feat(water-leak): add durable journal-backed recovery service

docs/durable-recovery.md

@@ -0,0 +1,155 @@
+# Durable warning recovery, explicit gaps and stated limits
+
+This extension implements a finite, disk-backed version of the water-warning
+experiment. Its purpose is to preserve accepted observations, pending warning
+obligations and the first acknowledgement receipt through **specified local
+process crashes**. It also exposes source gaps, arrival silence and storage limits.
+It does not maintain uninterrupted decisions while its only process is stopped.
+
+## Reproduce
+
+Use the pinned Elixir/OTP environment:
+
+```sh
+mix test --seed 0 test/firmboot/durable_test.exs
+```
+
+This change adds the service and its tests only. It ships no timing-measurement
+runner and no recorded measurement, so this document claims no latency or
+throughput figures.
+
+## Operations and commit boundary
+
+`Durable.create(path, options)` creates a new file exclusively. `Durable.start(path)`
+recovers an existing file; a missing file is an error, not permission to reset the
+service. Same-expanded-path registration prevents another cooperating writer in
+the same VM. Path aliases, external file replacement and concurrent other VMs
+are outside this writer-exclusion mechanism.
+
+Call `Durable.submit(pid, operation_id, command)` with one of:
+
+```elixir
+{:sample, %{seq: 1, inlet_ml: 1000, outlet_ml: 850}}
+{:update, :v2, 1, 1, 1, :none} # target, preparation ticks/cost, commit cost, test fault
+{:acknowledge, {"sim-loop", 3}, "operator-A"}
+```
+
+Operation IDs and labels are nonempty valid UTF-8 strings of at most 64 bytes.
+Readings have exactly three fields and unsigned 64-bit integer quantities; sequence
+numbers start at one. Update counts/costs fit 16 bits and are still checked by the
+existing model's stricter admission rules. Known update faults are local test
+controls, not an arbitrary replacement-code interface.
+
+Each new well-shaped command is evaluated by the deterministic transition logic,
+appended to the journal and synchronized before its reply. A domain rejection
+(such as an unknown warning) is also recorded, so the same operation ID retains
+the same result. Invalid schemas and conflicting reuse of an operation ID are
+rejected without appending. A different operation ID is a different attempt.
+
+An interrupted call or write/sync error has an **unknown outcome**. Retry with the
+same ID and identical command. If a complete command survived, recovery returns
+its original result; otherwise the retry can commit it once. A different payload
+under the same ID is rejected. Repeating an acknowledgement cannot overwrite its
+first receipt. An original reply describes its original boundary, not current
+service status; use `status`/`snapshot` for current state.
+
+The sync primitive requests that operating-system buffers reach storage, according
+to Erlang's [file interface](https://www.erlang.org/doc/apps/kernel/file.html#sync/1).
+Its successful return is an assumption at this boundary, not qualification of the
+filesystem, device, host power supply or WSL storage stack. Directory durability,
+host/kernel failure, arbitrary bit loss and power loss have not been qualified.
+
+## Gap handling and freshness
+
+A sample beyond the next expected sequence is durably buffered. The detector
+processes nothing across that gap. Missing predecessors cause all newly contiguous
+readings to drain in source order. Equal duplicate readings do not create another
+observation; conflicting values for an accepted sequence are rejected.
+
+For arrivals `3, 1, 2`, processed-through positions are `0, 1, 3`. Current gap
+ranges are respectively `[1..2]`, `[2..2]`, then empty. These ranges concern only
+the horizon of readings received; the service cannot know future missing indices.
+The sample reply means **acceptance**, not necessarily decision completion.
+
+`Durable.status(pid)` reports gaps, buffered count, last new arrival age, storage
+occupancy, observed process memory, observed mailbox length and recent commit
+durations. Arrival status starts as `unknown_after_start`, including after recovery.
+Only a new accepted reading refreshes the local arrival clock; duplicate/retried
+data does not. Silence is evaluated against the contract's local monotonic-time
+threshold when status is queried. There is no autonomous external alarm service.
+
+Recent arrival does not prove recent physical measurement. A sensor timestamp,
+clock-domain relation, calibration and source identity need their own evidence.
+An external observer is needed to detect an unresponsive or stopped service; a
+status request cannot guarantee a timely answer during a blocked sync or overload.
+
+## Journal format, bounds and integrity
+
+Each slot is exactly 512 bytes: eight-byte magic `FBWL001!`, a big-endian 16-bit
+payload length, 32-byte SHA-256 digest and a 470-byte payload/padding region.
+The digest covers the preceding slot digest, payload length and payload. Padding
+must be zero. The first slot uses an all-zero preceding digest and stores the
+contract and executable fingerprint. Subsequent slots contain fixed-schema sample,
+update, acknowledgement or completed-tail-repair records. No Erlang external
+term, source expression, arbitrary atom or function is decoded from the journal.
+
+`max_records` is between 1 and 4096 (default 512). The byte cap is exactly:
+
+```text
+(max_records + 1 header) × 512 bytes
+```
+
+New operation IDs are refused before exceeding the cap. Existing-ID retries
+remain available when full. Exhaustion can also prevent a new acknowledgement:
+this prototype does not reserve alarm/receipt capacity, rotate or delete records,
+compact the journal, or export obligations to another storage service. That is
+an explicit service limit, not a claim of indefinite bounded retention.
+
+The cap also bounds accepted command/reading/warning counts; input labels and
+integers have bounded widths, and cached replies have bounded fields. Actual heap
+layout, callers, mailbox growth, VM allocation and CPU scheduling are not proved
+bounded. Status memory readings are samples, not measurements of every transient
+allocation. Draining a backlog and replaying the journal take work proportional
+to the retained workload; no worst-case time bound follows.
+
+The fingerprint binds the available BEAM object bytes for the detector, model,
+warning wrapper, update type, codec, transition, journal and service, plus Elixir
+and ERTS version labels. Each object's module MD5 is checked against the loaded
+module before computing the fingerprint. Recovery refuses mismatches. Debug/build
+differences can conservatively change the fingerprint. This is not a cross-build
+compatibility proof, signed deployment protocol or proof that candidate code is safe.
+
+Complete malformed slots, wrong checksums, reordered slots and mismatched executable
+fingerprints cause refusal. A partial final slot is refused by default. An operator
+may explicitly start with `repair_tail: true` only when the append-prefix crash
+assumption is justified: remove that partial final slot, sync, and append a repair
+record before resuming. A completed repair consumes one slot and survives later
+restarts. Interrupted repair itself may need a renewed recovery decision; no claim
+is made that every interrupted repair attempt leaves its own durable audit entry.
+
+The chain has **no external anchor or authentication**. Replacing the file with
+an older complete prefix can erase a receipt without local detection. The test
+suite includes this counterexample. A malicious writer able to rewrite the file
+can also recompute hashes. The chain does not establish rollback protection or
+tamper resistance against that actor.
+
+## Executed evidence
+
+Fourteen recovery tests cover exact replay, gaps, duplicates/conflicts, malformed
+commands, capacity, writer exclusion, corruption, executable mismatch and the
+complete-prefix rollback limitation. One test enumerates **all 720 permutations
+of the six-reading fixture** and independently checks each completed ordered
+result. A further **56 operation/boundary cases** apply four crash points to each
+of the twelve readings, update and acknowledgement in the domain fixture. The
+crash points kill the actual service process: before writing,
+after a partial write, after a complete write, and after synchronization before
+reply. A separate injected sync error checks the writer's refusal to continue
+from uncertain volatile state. It does not reproduce a failing physical device.
+
+The existing buffered-restart model remains a separate [comparison](../REPORT.md).
+Nothing here establishes performance superiority over restart, standby or other
+recovery designs.
+
+The existing Lean/Agda proofs continue to cover their defined transition models.
+No new theorem about this file format, OS synchronization or concrete replay
+implementation is claimed.

lib/firmboot/water_leak/durable.ex

@@ -0,0 +1,181 @@
+defmodule Firmboot.WaterLeak.Durable do
+  @moduledoc """
+  A finite, single-writer journal service for the water-warning experiment.
+
+  Every new valid command (including a business rejection) is journaled before
+  its reply. An interrupted call has an unknown outcome: retry its operation ID
+  and identical command. A successful sample reply means durable acceptance;
+  samples beyond a gap remain buffered until their predecessors arrive.
+
+  Limits bound journal bytes/records and replay state entries, not the VM heap,
+  callers or mailbox. Timing is measured, not guaranteed. Labels are not identities
+  authenticated over a network. Same-path writer exclusion covers cooperating
+  processes in this VM; other VMs, aliases and external writers are excluded.
+  """
+  use GenServer
+  alias Firmboot.WaterLeak.Durable.{Codec, Journal, Transition}
+
+  @doc """
+  Creates a new journal at `path`, failing if the file already exists.
+
+  Options: `:source`, `:threshold_ml`, `:max_records` and `:silence_ms`. The header
+  records them together with the current engine fingerprint.
+  """
+  def create(path, options \\ []), do: Journal.create(Path.expand(path), options)
+
+  @doc """
+  Starts the single writer for an existing journal at `path` and replays it.
+
+  A missing file is an error, never a fresh service. An incomplete final slot stops
+  recovery unless `repair_tail: true` consents to the append-prefix crash model.
+  """
+  def start(path, options \\ []) do
+    options = Keyword.validate!(options, repair_tail: false)
+
+    GenServer.start(__MODULE__, {Path.expand(path), options[:repair_tail]},
+      name: {:global, {__MODULE__, Path.expand(path)}}
+    )
+  end
+
+  @doc """
+  Journals and applies one command under `operation_id`, replying only after sync.
+
+  Retrying an ID with the identical command returns the original reply; the same ID
+  with a different command is `{:error, :operation_id_conflict}`.
+  """
+  def submit(pid, operation_id, command) do
+    if Codec.label?(operation_id) and Codec.valid_command?(command),
+      do: GenServer.call(pid, {:submit, operation_id, command}, :infinity),
+      else: {:error, :invalid_command}
+  end
+
+  @doc "Returns the replayed domain state: water model, buffered readings and operations."
+  def snapshot(pid), do: GenServer.call(pid, :snapshot)
+
+  @doc "Reports gaps, arrival silence, journal usage, tail repairs and measured commit times."
+  def status(pid), do: GenServer.call(pid, :status)
+
+  @doc "Stops the writer and closes its journal file."
+  def stop(pid), do: GenServer.stop(pid)
+
+  @doc "Local fault-injection control; never an untrusted remote update interface."
+  def arm_fault(pid, point)
+      when point in [:before_write, :partial_write, :after_write, :after_sync, :sync_error],
+      do: GenServer.call(pid, {:fault, point})
+
+  @impl true
+  def init({path, repair_tail}) do
+    with true <- is_boolean(repair_tail),
+         {:ok, journal, records} <- Journal.open(path, repair_tail) do
+      case Transition.replay(journal.config, records) do
+        {:ok, domain} ->
+          {:ok,
+           %{
+             journal: journal,
+             domain: domain,
+             last_arrival: nil,
+             fault: nil,
+             durations_us: [],
+             commits: 0,
+             max_commit_us: 0
+           }}
+
+        {:error, reason} ->
+          Journal.close(journal)
+          {:stop, reason}
+      end
+    else
+      false -> {:stop, :invalid_repair_option}
+      {:error, reason} -> {:stop, reason}
+    end
+  end
+
+  @impl true
+  def handle_call(:snapshot, _from, s), do: {:reply, s.domain, s}
+  def handle_call({:fault, fault}, _from, s), do: {:reply, :ok, %{s | fault: fault}}
+
+  def handle_call(:status, _from, s) do
+    age = if s.last_arrival, do: System.monotonic_time(:millisecond) - s.last_arrival, else: nil
+    {:memory, memory} = Process.info(self(), :memory)
+    {:message_queue_len, queued} = Process.info(self(), :message_queue_len)
+
+    {:reply,
+     %{
+       processed_through: s.domain.water.model.seq,
+       gaps: Transition.gaps(s.domain),
+       buffered_readings: map_size(s.domain.waiting),
+       arrival_status:
+         cond do
+           age == nil -> :unknown_after_start
+           age >= s.journal.config.silence_ms -> :silent
+           true -> :recent_arrival
+         end,
+       last_new_arrival_age_ms: age,
+       journal_bytes: Journal.bytes(s.journal),
+       journal_capacity_bytes: Journal.capacity(s.journal),
+       records: s.journal.count,
+       storage_full: s.journal.count == s.journal.config.max_records,
+       tail_repairs: Enum.reverse(s.domain.repairs),
+       process_memory_bytes: memory,
+       queued_messages_observed: queued,
+       commits_this_process: s.commits,
+       max_commit_us: s.max_commit_us,
+       recent_commit_us: Enum.reverse(s.durations_us)
+     }, s}
+  end
+
+  def handle_call({:submit, id, command}, _from, s) do
+    # Validate again at the actual GenServer boundary, not only in the public helper.
+    if Codec.label?(id) and Codec.valid_command?(command) do
+      case Transition.retry(s.domain, id, command) do
+        {:retry, reply} -> {:reply, reply, s}
+        {:conflict, reply} -> {:reply, reply, s}
+        :new -> commit(s, id, command)
+      end
+    else
+      {:reply, {:error, :invalid_command}, s}
+    end
+  end
+
+  defp commit(s, id, command) do
+    if s.journal.count >= s.journal.config.max_records do
+      {:reply, {:error, :storage_capacity}, s}
+    else
+      began = System.monotonic_time(:microsecond)
+      {domain, reply} = Transition.apply_new(s.domain, id, command)
+
+      case Journal.append(s.journal, {id, command}, s.fault) do
+        {:ok, journal} ->
+          elapsed = System.monotonic_time(:microsecond) - began
+
+          new_reading =
+            match?({:sample, _}, command) and
+              (domain.water.model.seq != s.domain.water.model.seq or
+                 domain.waiting != s.domain.waiting)
+
+          last_arrival =
+            if new_reading, do: System.monotonic_time(:millisecond), else: s.last_arrival
+
+          {:reply, reply,
+           %{
+             s
+             | journal: journal,
+               domain: domain,
+               fault: nil,
+               last_arrival: last_arrival,
+               commits: s.commits + 1,
+               durations_us: Enum.take([elapsed | s.durations_us], 128),
+               max_commit_us: max(s.max_commit_us, elapsed)
+           }}
+
+        {:error, reason} ->
+          # A failed write/sync can have changed the file. Never continue from
+          # volatile state or label the unknown outcome an uncommitted rejection.
+          {:stop, {:journal_failure, reason}, s}
+      end
+    end
+  end
+
+  @impl true
+  def terminate(_reason, s), do: Journal.close(s.journal)
+end