From b154e0a736768161ee350c6c660a61666cf1bdd5 Mon Sep 17 00:00:00 2001 From: "Jonathan D.A. Jewell" <6759885+hyperpolymath@users.noreply.github.com> Date: Thu, 8 Oct 2026 20:34:19 +0100 Subject: [PATCH] feat(water-leak): add durable journal-backed recovery service Adds Firmboot.WaterLeak.Durable, a single-writer service that journals every new valid command in 512-byte SHA-256-chained slots and syncs before replying, so accepted readings, pending warnings and the first acknowledgement receipt survive the tested local process crashes. Sequence gaps are buffered and reported, never filled in. - lib/firmboot/water_leak/durable{,/codec,/journal,/transition}.ex - test/firmboot/durable_test.exs: 14 tests, including all 720 arrival permutations of a six-reading fixture and 56 operation x crash-point cases (before write, partial write, after write, after sync) - docs/durable-recovery.md: API, commit boundary, format and limits - mix.exs: extra_applications [:crypto] for the slot chain Provenance: extracted from the unpushed local commit 7c29853 ("chore: land estate governance sync", 2026-09-15), which had swept this work in alongside governance files that must not land. That branch is kept at refs/rescued/20261008-firmboot-unpushed/ chore/bump-standards-pins in the local clone; triage note dev-notes/inbox/2026-10-08-firmboot-and-fleet-branch-triage.md. Not carried over: the branch's rewrite of recovery_experiment.ex and water_leak_test.exs (they would revert #4's checkpoint tests), its timing-measurement runner and recorded result, and its assurance register, which depends on scripts/assurance.py (Python is banned in the estate). The doc was trimmed to claim only what this change's code and tests reproduce. @doc added to every public function. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015bTuGfwCcvjrmNFejydTML --- docs/durable-recovery.md | 155 +++++++ lib/firmboot/water_leak/durable.ex | 181 ++++++++ lib/firmboot/water_leak/durable/codec.ex | 134 ++++++ lib/firmboot/water_leak/durable/journal.ex | 220 ++++++++++ lib/firmboot/water_leak/durable/transition.ex | 126 ++++++ mix.exs | 2 + test/firmboot/durable_test.exs | 392 ++++++++++++++++++ 7 files changed, 1210 insertions(+) create mode 100644 docs/durable-recovery.md create mode 100644 lib/firmboot/water_leak/durable.ex create mode 100644 lib/firmboot/water_leak/durable/codec.ex create mode 100644 lib/firmboot/water_leak/durable/journal.ex create mode 100644 lib/firmboot/water_leak/durable/transition.ex create mode 100644 test/firmboot/durable_test.exs diff --git a/docs/durable-recovery.md b/docs/durable-recovery.md new file mode 100644 index 0000000..0126951 --- /dev/null +++ b/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. diff --git a/lib/firmboot/water_leak/durable.ex b/lib/firmboot/water_leak/durable.ex new file mode 100644 index 0000000..1a1457d --- /dev/null +++ b/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 diff --git a/lib/firmboot/water_leak/durable/codec.ex b/lib/firmboot/water_leak/durable/codec.ex new file mode 100644 index 0000000..bdf92f6 --- /dev/null +++ b/lib/firmboot/water_leak/durable/codec.ex @@ -0,0 +1,134 @@ +defmodule Firmboot.WaterLeak.Durable.Codec do + @moduledoc "Fixed, bounded journal schema. Decoding creates no atoms or executable terms." + @max_integer 18_446_744_073_709_551_615 + @max_records 4096 + + @doc "The largest record capacity any journal header may declare." + def max_records, do: @max_records + + @doc "True for a nonblank valid UTF-8 string of 1 to 64 bytes (operation IDs, sources, operators)." + def label?(s), + do: is_binary(s) and byte_size(s) in 1..64 and String.valid?(s) and String.trim(s) != "" + + @doc "True for an integer that fits an unsigned 64-bit field." + def uint?(n), do: is_integer(n) and n >= 0 and n <= @max_integer + + @doc """ + Builds a validated journal configuration from `options` and an engine `fingerprint`. + + Returns `{:ok, config}` or `{:error, :invalid_contract}`; unknown options raise. + """ + def config(options, fingerprint) do + options = + Keyword.validate!(options, + source: "sim-loop", + threshold_ml: 100, + max_records: 512, + silence_ms: 5_000 + ) + + config = Map.new(options) |> Map.put(:fingerprint, fingerprint) + + if valid_config?(config), do: {:ok, config}, else: {:error, :invalid_contract} + end + + @doc "True when every configuration field is within its schema bound." + def valid_config?(c) do + label?(c.source) and uint?(c.threshold_ml) and c.threshold_ml > 0 and + is_integer(c.max_records) and c.max_records in 1..@max_records and + is_integer(c.silence_ms) and c.silence_ms in 1..86_400_000 and + is_binary(c.fingerprint) and byte_size(c.fingerprint) == 32 + end + + @doc "True for a well-formed `:sample`, `:update` or `:acknowledge` command within schema bounds." + def valid_command?({:sample, %{seq: seq, inlet_ml: inlet, outlet_ml: outlet} = r}), + do: map_size(r) == 3 and uint?(seq) and seq > 0 and uint?(inlet) and uint?(outlet) + + def valid_command?({:update, target, ticks, prepare, commit, fault}), + do: + target in [:v1, :v2] and fault in [:none, :position, :reported, :history] and + Enum.all?([ticks, prepare, commit], &(is_integer(&1) and &1 in 0..65_535)) + + def valid_command?({:acknowledge, {source, episode}, operator}), + do: label?(source) and uint?(episode) and episode > 0 and label?(operator) + + def valid_command?(_), do: false + + @doc "Encodes a header, a journaled command or a tail-repair marker as a tagged binary payload." + def encode({:header, c}), + do: + <<0, c.max_records::32, c.threshold_ml::64, c.silence_ms::32, + c.fingerprint::binary-size(32), byte_size(c.source), c.source::binary>> + + def encode({id, {:sample, r}}), + do: <<1, byte_size(id), id::binary, r.seq::64, r.inlet_ml::64, r.outlet_ml::64>> + + def encode({id, {:update, target, ticks, prepare, commit, fault}}), + do: + <<2, byte_size(id), id::binary, version(target), ticks::16, prepare::16, commit::16, + fault(fault)>> + + def encode({id, {:acknowledge, {source, episode}, operator}}), + do: + <<3, byte_size(id), id::binary, byte_size(source), source::binary, episode::64, + byte_size(operator), operator::binary>> + + def encode({:tail_repair, bytes}), do: <<4, bytes::16>> + + @doc """ + Decodes one payload and revalidates it against the schema. + + Returns `{:ok, record}` or `{:error, :schema}`; it creates no atoms from input. + """ + def decode(binary) do + case parse(binary) do + {:header, c} = record -> + if valid_config?(c), do: {:ok, record}, else: {:error, :schema} + + {:tail_repair, n} = record when n in 1..511 -> + {:ok, record} + + {id, command} = record when is_binary(id) -> + if label?(id) and valid_command?(command), do: {:ok, record}, else: {:error, :schema} + + _ -> + {:error, :schema} + end + end + + defp parse( + <<0, cap::32, threshold::64, silence::32, fingerprint::binary-size(32), n, + source::binary-size(n)>> + ), + do: + {:header, + %{ + source: source, + threshold_ml: threshold, + max_records: cap, + silence_ms: silence, + fingerprint: fingerprint + }} + + defp parse(<<1, n, id::binary-size(n), seq::64, inlet::64, outlet::64>>), + do: {id, {:sample, %{seq: seq, inlet_ml: inlet, outlet_ml: outlet}}} + + defp parse(<<2, n, id::binary-size(n), v, ticks::16, prepare::16, commit::16, f>>) + when v in 1..2 and f in 0..3, + do: + {id, + {:update, Enum.at([:v1, :v2], v - 1), ticks, prepare, commit, + Enum.at([:none, :position, :reported, :history], f)}} + + defp parse( + <<3, n, id::binary-size(n), m, source::binary-size(m), episode::64, k, + operator::binary-size(k)>> + ), + do: {id, {:acknowledge, {source, episode}, operator}} + + defp parse(<<4, bytes::16>>), do: {:tail_repair, bytes} + defp parse(_), do: :invalid + defp version(:v1), do: 1 + defp version(:v2), do: 2 + defp fault(f), do: Enum.find_index([:none, :position, :reported, :history], &(&1 == f)) +end diff --git a/lib/firmboot/water_leak/durable/journal.ex b/lib/firmboot/water_leak/durable/journal.ex new file mode 100644 index 0000000..8f4278e --- /dev/null +++ b/lib/firmboot/water_leak/durable/journal.ex @@ -0,0 +1,220 @@ +defmodule Firmboot.WaterLeak.Durable.Journal do + @moduledoc """ + 512-byte chained journal slots with bounded binary payloads and sync-before-reply. + + The SHA-256 chain detects tested corruption/reordering; it is not authentication + or protection against replacing/truncating an entire valid journal. Tail repair + requires explicit consent to the append-prefix crash model. No power-loss or + filesystem correctness proof is asserted. One compliant writer must own the file. + """ + alias Firmboot.WaterLeak.Durable.Codec + @slot 512 + @magic "FBWL001!" + @zero <<0::256>> + @payload_size 470 + defstruct [:fd, :config, :hash, count: 0] + + @doc "The fixed size in bytes of every journal slot, header included." + def slot_size, do: @slot + + @doc """ + SHA-256 over the Elixir and OTP versions and the object code of every module that + decides replay, so a journal never replays under different engine code. + + Raises if a loaded module differs from its available object code. + """ + def fingerprint do + modules = [ + Firmboot.WaterLeak, + Firmboot.Detector, + Firmboot.Model, + Firmboot.Update, + Codec, + Firmboot.WaterLeak.Durable.Transition, + __MODULE__, + Firmboot.WaterLeak.Durable + ] + + objects = + Enum.map(modules, fn mod -> + Code.ensure_loaded!(mod) + {^mod, binary, _} = :code.get_object_code(mod) + + unless :code.module_md5(binary) == mod.module_info(:md5), + do: raise("loaded code differs from its available object code: #{inspect(mod)}") + + [Atom.to_string(mod), :crypto.hash(:sha256, binary)] + end) + + :crypto.hash(:sha256, [System.version(), :erlang.system_info(:version), objects]) + end + + @doc "Exclusively creates `path` and writes and syncs its header slot; an existing file is an error." + def create(path, options) do + with {:ok, config} <- Codec.config(options, fingerprint()), + {:ok, fd} <- :file.open(path, [:write, :binary, :raw, :exclusive]) do + {slot, _} = frame({:header, config}, @zero) + result = with :ok <- :file.write(fd, slot), do: :file.sync(fd) + closed = :file.close(fd) + if result == :ok, do: closed, else: result + end + end + + @doc """ + Opens an existing journal, verifies its chain and fingerprint, and returns + `{:ok, journal, records}` positioned for append. + + An incomplete final slot is `{:error, {:incomplete_tail, bytes}}` unless `repair_tail` + is true, in which case the tail is truncated and a repair marker is journaled. + """ + def open(path, repair_tail) do + # Stat precedes open so a missing journal cannot silently create a new service. + # The directory and file are trusted; concurrent external path replacement is excluded. + with {:ok, info} <- File.stat(path), + true <- info.type == :regular and info.size <= (Codec.max_records() + 1) * @slot, + {:ok, fd} <- :file.open(path, [:read, :write, :binary, :raw]) do + result = recover(fd, repair_tail) + if not match?({:ok, _, _}, result), do: :file.close(fd) + result + else + false -> {:error, :invalid_journal_size_or_type} + error -> error + end + end + + @doc "Closes the journal's file descriptor." + def close(journal), do: :file.close(journal.fd) + + @doc "Bytes currently used: the header slot plus one slot per record." + def bytes(journal), do: (journal.count + 1) * @slot + + @doc "Bytes the journal may reach: the header slot plus `max_records` slots." + def capacity(journal), do: (journal.config.max_records + 1) * @slot + + @doc """ + Frames `record` into the next chained slot, writes it and syncs before returning. + + `fault` is a local test injection point; `nil` in normal use. + """ + def append(journal, record, fault \\ nil) do + if journal.count >= journal.config.max_records do + {:error, :storage_capacity} + else + {slot, hash} = frame(record, journal.hash) + crash(fault, :before_write) + + if fault == :partial_write do + :ok = :file.write(journal.fd, binary_part(slot, 0, div(@slot, 2))) + Process.exit(self(), :kill) + end + + with :ok <- :file.write(journal.fd, slot) do + crash(fault, :after_write) + + synced = + if fault == :sync_error, + do: {:error, :injected_sync_error}, + else: :file.sync(journal.fd) + + with :ok <- synced do + crash(fault, :after_sync) + {:ok, %{journal | hash: hash, count: journal.count + 1}} + end + end + end + end + + defp crash(point, point) when not is_nil(point), do: Process.exit(self(), :kill) + defp crash(_, _), do: :ok + + defp recover(fd, repair_tail) do + with {:ok, first} <- :file.read(fd, @slot), + {:ok, {:header, config}, hash} <- unframe(first, @zero), + true <- config.fingerprint == fingerprint(), + {:ok, records, last_hash, tail} <- read_records(fd, hash, config.max_records, []), + {:ok, _} <- Firmboot.WaterLeak.Durable.Transition.replay(config, records), + {:ok, _} <- :file.position(fd, (length(records) + 1) * @slot) do + journal = %__MODULE__{fd: fd, config: config, hash: last_hash, count: length(records)} + + cond do + tail == 0 -> + # A complete write may have survived without its caller seeing a reply. + # Synchronize the recovered prefix before serving an idempotent retry. + with :ok <- :file.sync(fd), do: {:ok, journal, records} + + not repair_tail -> + {:error, {:incomplete_tail, tail}} + + journal.count >= config.max_records -> + {:error, :storage_capacity} + + true -> + with :ok <- :file.truncate(fd), + :ok <- :file.sync(fd), + {:ok, repaired} <- append(journal, {:tail_repair, tail}) do + {:ok, repaired, records ++ [{:tail_repair, tail}]} + end + end + else + false -> {:error, :engine_fingerprint_mismatch} + :eof -> {:error, :missing_header} + {:ok, _, _} -> {:error, :missing_header} + error -> error + end + end + + defp read_records(fd, hash, remaining, records) do + case :file.read(fd, @slot) do + :eof -> + {:ok, Enum.reverse(records), hash, 0} + + {:ok, binary} when byte_size(binary) < @slot -> + {:ok, Enum.reverse(records), hash, byte_size(binary)} + + {:ok, _} when remaining == 0 -> + {:error, :record_limit_exceeded} + + {:ok, binary} -> + case unframe(binary, hash) do + {:ok, {:header, _}, _} -> + {:error, :unexpected_header} + + {:ok, record, next_hash} -> + read_records(fd, next_hash, remaining - 1, [record | records]) + + error -> + error + end + + error -> + error + end + end + + defp frame(record, previous) do + payload = Codec.encode(record) + size = byte_size(payload) + true = size <= @payload_size + hash = :crypto.hash(:sha256, [previous, <>, payload]) + + {<<@magic, size::16, hash::binary-size(32), payload::binary, + 0::size((@payload_size - size) * 8)>>, hash} + end + + defp unframe( + <<@magic, size::16, hash::binary-size(32), rest::binary-size(@payload_size)>>, + previous + ) + when size <= @payload_size do + <> = rest + expected = :crypto.hash(:sha256, [previous, <>, payload]) + + if hash == expected and padding == :binary.copy(<<0>>, @payload_size - size) do + with {:ok, record} <- Codec.decode(payload), do: {:ok, record, hash} + else + {:error, :journal_integrity} + end + end + + defp unframe(_, _), do: {:error, :journal_framing} +end diff --git a/lib/firmboot/water_leak/durable/transition.ex b/lib/firmboot/water_leak/durable/transition.ex new file mode 100644 index 0000000..8eb3b50 --- /dev/null +++ b/lib/firmboot/water_leak/durable/transition.ex @@ -0,0 +1,126 @@ +defmodule Firmboot.WaterLeak.Durable.Transition do + @moduledoc "Deterministic replay of bounded commands; source gaps never manufacture samples." + alias Firmboot.{Update, WaterLeak} + defstruct [:water, waiting: %{}, operations: %{}, repairs: []] + + @doc "Initial domain state for a journal configuration: an empty water model and no operations." + def new(config), + do: %__MODULE__{ + water: + WaterLeak.new( + source: config.source, + threshold_ml: config.threshold_ml + ) + } + + @doc "Rebuilds domain state from journaled records; a repeated operation ID is an error." + def replay(config, records) do + Enum.reduce_while(records, {:ok, new(config)}, fn + {:tail_repair, bytes}, {:ok, s} -> + {:cont, {:ok, %{s | repairs: [bytes | s.repairs]}}} + + {id, command}, {:ok, s} -> + if Map.has_key?(s.operations, id) do + {:halt, {:error, :duplicate_operation_in_journal}} + else + {next, _} = apply_new(s, id, command) + {:cont, {:ok, next}} + end + end) + end + + @doc """ + Classifies `id`: `:new` if unseen, `{:retry, reply}` with the stored reply if seen + with the identical command, or `{:conflict, error}` if seen with a different one. + """ + def retry(s, id, command) do + case Map.fetch(s.operations, id) do + :error -> :new + {:ok, {^command, reply}} -> {:retry, reply} + {:ok, _} -> {:conflict, {:error, :operation_id_conflict}} + end + end + + @doc "Executes a new command and records it with its reply under `id`; returns `{state, reply}`." + def apply_new(s, id, command) do + {s, reply} = execute(s, id, command) + {%{s | operations: Map.put(s.operations, id, {command, reply})}, reply} + end + + @doc "Missing sequence ranges, as inclusive `{first, last}` pairs, below the buffered readings." + def gaps(s) do + {ranges, _} = + s.waiting + |> Map.keys() + |> Enum.sort() + |> Enum.reduce({[], s.water.model.seq + 1}, fn seq, {ranges, expected} -> + ranges = if seq > expected, do: [{expected, seq - 1} | ranges], else: ranges + {ranges, seq + 1} + end) + + Enum.reverse(ranges) + end + + defp execute(s, _id, {:sample, reading}) do + existing = + Map.get(s.waiting, reading.seq) || + Enum.find(s.water.observations, &(&1.seq == reading.seq)) + + cond do + existing == reading -> + {s, {:ok, :duplicate_reading}} + + existing != nil -> + {s, {:error, :conflicting_reading}} + + true -> + next = drain(%{s | waiting: Map.put(s.waiting, reading.seq, reading)}) + + {next, + {:ok, + %{ + accepted_seq: reading.seq, + processed_through: next.water.model.seq, + gaps_present: map_size(next.waiting) > 0 + }}} + end + end + + defp execute(s, id, {:update, target, ticks, prepare, commit, fault}) do + water = + WaterLeak.request(s.water, %Update{ + id: id, + target: target, + prepare_ticks: ticks, + prepare_cost: prepare, + commit_cost: commit, + fault: fault + }) + + reply = + case hd(water.model.audit) do + %{kind: :accepted} -> {:ok, :update_accepted} + %{kind: :rejected, reason: reason} -> {:error, {:update_rejected, reason}} + end + + {%{s | water: water}, reply} + end + + defp execute(s, _id, {:acknowledge, key, operator}) do + case WaterLeak.acknowledge(s.water, key, operator) do + {:ok, water} -> {%{s | water: water}, {:ok, water.warnings[key].acknowledgement}} + {:error, reason} -> {s, {:error, reason}} + end + end + + defp drain(s) do + case Map.pop(s.waiting, s.water.model.seq + 1) do + {nil, _} -> + s + + {reading, waiting} -> + {:ok, water} = WaterLeak.sample(s.water, reading) + drain(%{s | water: water, waiting: waiting}) + end + end +end diff --git a/mix.exs b/mix.exs index d7a74c4..18bdc18 100644 --- a/mix.exs +++ b/mix.exs @@ -10,4 +10,6 @@ defmodule FirmbootReference.MixProject do deps: [] ] end + + def application, do: [extra_applications: [:crypto]] end diff --git a/test/firmboot/durable_test.exs b/test/firmboot/durable_test.exs new file mode 100644 index 0000000..8edc825 --- /dev/null +++ b/test/firmboot/durable_test.exs @@ -0,0 +1,392 @@ +defmodule Firmboot.DurableTest do + use ExUnit.Case, async: false + @moduletag :capture_log + alias Firmboot.WaterLeak + alias Firmboot.WaterLeak.{Checker, Durable, Experiment} + alias Firmboot.WaterLeak.Durable.{Codec, Journal} + @key {"sim-loop", 3} + + setup do + dir = Path.join(System.tmp_dir!(), "firmboot-durable-#{System.unique_integer([:positive])}") + File.mkdir_p!(dir) + on_exit(fn -> File.rm_rf!(dir) end) + %{dir: dir, path: Path.join(dir, "water.journal")} + end + + test "recovery replays the exact warning, migration and first receipt", %{path: path} do + assert :ok = Durable.create(path) + pid = start(path) + seed(pid) + command = {:acknowledge, @key, "operator-A"} + assert {:ok, receipt} = Durable.submit(pid, "ack", command) + before = Durable.snapshot(pid) + assert before.water.model.version == :v2 + assert WaterLeak.outstanding(before.water) == [] + assert :ok = Durable.stop(pid) + recovered = start(path) + assert Durable.snapshot(recovered) == before + assert {:ok, ^receipt} = Durable.submit(recovered, "ack", command) + + assert {:error, :operation_id_conflict} = + Durable.submit(recovered, "ack", {:acknowledge, @key, "operator-B"}) + + assert Durable.snapshot(recovered) == before + + assert :ok = + Checker.verify( + Enum.take(Experiment.readings(), 6), + [receipt], + before.water, + Experiment.contract() + ) + end + + test "pending source gaps survive restart and drain only when missing readings arrive", %{ + path: path + } do + :ok = Durable.create(path) + pid = start(path) + [one, two, three | _] = Experiment.readings() + + assert {:ok, %{processed_through: 0, gaps_present: true}} = + Durable.submit(pid, "three", {:sample, three}) + + :ok = Durable.stop(pid) + pid = start(path) + assert Durable.status(pid).gaps == [{1, 2}] + assert Durable.status(pid).arrival_status == :unknown_after_start + + assert {:ok, %{processed_through: 1, gaps_present: true}} = + Durable.submit(pid, "one", {:sample, one}) + + assert {:ok, %{processed_through: 3, gaps_present: false}} = + Durable.submit(pid, "two", {:sample, two}) + + assert Durable.snapshot(pid).waiting == %{} + assert Enum.reverse(Durable.snapshot(pid).water.observations) == [one, two, three] + + assert :ok = + Checker.verify( + [one, two, three], + [], + Durable.snapshot(pid).water, + Experiment.contract() + ) + end + + test "duplicates cannot rewrite an accepted reading or claim new arrival freshness", %{ + path: path + } do + :ok = Durable.create(path, silence_ms: 1) + pid = start(path) + first = hd(Experiment.readings()) + assert {:ok, _} = Durable.submit(pid, "first", {:sample, first}) + Process.sleep(3) + assert Durable.status(pid).arrival_status == :silent + assert {:ok, :duplicate_reading} = Durable.submit(pid, "duplicate", {:sample, first}) + assert Durable.status(pid).arrival_status == :silent + + assert {:error, :conflicting_reading} = + Durable.submit(pid, "conflict", {:sample, %{first | inlet_ml: 999}}) + + assert Durable.snapshot(pid).water.observations == [first] + end + + test "all 720 arrival permutations preserve a six-reading fixture", %{dir: dir} do + readings = Enum.take(Experiment.readings(), 6) + + for {order, index} <- Enum.with_index(permutations(readings)) do + path = Path.join(dir, "permutation-#{index}") + :ok = Durable.create(path, max_records: 6) + pid = start(path) + + for reading <- order do + assert {:ok, _} = Durable.submit(pid, "s#{reading.seq}", {:sample, reading}) + end + + state = Durable.snapshot(pid) + assert :ok = Checker.verify(readings, [], state.water, Experiment.contract()) + assert state.waiting == %{} + assert :ok = Durable.stop(pid) + end + end + + @tag :capture_log + test "real process crashes at four append boundaries preserve confirmed obligations", %{ + dir: dir + } do + for point <- [:before_write, :partial_write, :after_write, :after_sync] do + path = Path.join(dir, Atom.to_string(point)) + :ok = Durable.create(path) + pid = start(path) + seed(pid) + ref = Process.monitor(pid) + :ok = Durable.arm_fault(pid, point) + command = {:acknowledge, @key, "operator-A"} + assert catch_exit(Durable.submit(pid, "uncertain-ack", command)) + assert_receive {:DOWN, ^ref, :process, ^pid, :killed} + + if point == :partial_write do + size = File.stat!(path).size + assert {:error, {:incomplete_tail, 256}} = Durable.start(path) + assert File.stat!(path).size == size + end + + recovered = start(path, repair_tail: point == :partial_write) + + if point in [:before_write, :partial_write] do + assert WaterLeak.outstanding(Durable.snapshot(recovered).water) == [@key] + else + assert WaterLeak.outstanding(Durable.snapshot(recovered).water) == [] + end + + assert {:ok, receipt} = Durable.submit(recovered, "uncertain-ack", command) + count = Durable.status(recovered).records + assert {:ok, ^receipt} = Durable.submit(recovered, "uncertain-ack", command) + assert Durable.status(recovered).records == count + + assert :ok = + Checker.verify( + Enum.take(Experiment.readings(), 6), + [receipt], + Durable.snapshot(recovered).water, + Experiment.contract() + ) + + if point == :partial_write do + assert Durable.status(recovered).tail_repairs == [256] + :ok = Durable.stop(recovered) + again = start(path) + assert Durable.status(again).tail_repairs == [256] + end + end + end + + @tag :capture_log + test "a sync error ends the writer and leaves the caller to resolve an unknown outcome", %{ + path: path + } do + :ok = Durable.create(path) + pid = start(path) + ref = Process.monitor(pid) + :ok = Durable.arm_fault(pid, :sync_error) + command = {:sample, hd(Experiment.readings())} + assert catch_exit(Durable.submit(pid, "uncertain", command)) + assert_receive {:DOWN, ^ref, :process, ^pid, {:journal_failure, :injected_sync_error}} + recovered = start(path) + assert {:ok, %{processed_through: 1}} = Durable.submit(recovered, "uncertain", command) + assert Durable.status(recovered).records == 1 + end + + test "capacity is enforced before another write, including after restart", %{path: path} do + :ok = Durable.create(path, max_records: 2) + pid = start(path) + [one, two, three | _] = Experiment.readings() + assert {:ok, reply} = Durable.submit(pid, "one", {:sample, one}) + assert {:ok, _} = Durable.submit(pid, "two", {:sample, two}) + :ok = Durable.stop(pid) + pid = start(path) + before = Durable.snapshot(pid) + assert {:error, :storage_capacity} = Durable.submit(pid, "three", {:sample, three}) + assert {:ok, ^reply} = Durable.submit(pid, "one", {:sample, one}) + assert Durable.snapshot(pid) == before + assert Durable.status(pid).storage_full + assert File.stat!(path).size == 3 * Journal.slot_size() + end + + test "56 operation/boundary crash cases preserve the independent final fixture", %{dir: dir} do + commands = + Enum.flat_map(Experiment.readings(), fn reading -> + sample = {"s#{reading.seq}", {:sample, reading}} + + if reading.seq == 5 do + [ + {"update", {:update, :v2, 1, 1, 1, :none}}, + sample, + {"ack", {:acknowledge, @key, "operator-A"}} + ] + else + [sample] + end + end) + + assert length(commands) == 14 + + for index <- 0..13, point <- [:before_write, :partial_write, :after_write, :after_sync] do + path = Path.join(dir, "cut-#{index}-#{point}") + :ok = Durable.create(path) + pid = start(path) + + for {id, command} <- Enum.take(commands, index), + do: assert(match?({:ok, _}, Durable.submit(pid, id, command))) + + {id, command} = Enum.at(commands, index) + ref = Process.monitor(pid) + :ok = Durable.arm_fault(pid, point) + assert catch_exit(Durable.submit(pid, id, command)) + assert_receive {:DOWN, ^ref, :process, ^pid, :killed} + pid = start(path, repair_tail: point == :partial_write) + + assert Map.has_key?(Durable.snapshot(pid).operations, id) == + point in [:after_write, :after_sync] + + assert {:ok, _} = Durable.submit(pid, id, command) + + for {rest_id, rest} <- Enum.drop(commands, index + 1), + do: assert(match?({:ok, _}, Durable.submit(pid, rest_id, rest))) + + receipt = %{id: @key, by: "operator-A", after_seq: 5} + + assert :ok = + Checker.verify( + Experiment.readings(), + [receipt], + Durable.snapshot(pid).water, + Experiment.contract() + ) + + assert :ok = Durable.stop(pid) + end + end + + test "semantic corruption is refused before any requested tail repair", %{path: path} do + :ok = Durable.create(path) + pid = start(path) + reading = hd(Experiment.readings()) + assert {:ok, _} = Durable.submit(pid, "same", {:sample, reading}) + :ok = Durable.stop(pid) + original = File.read!(path) + + <<_header::binary-size(512), _prefix::binary-size(10), previous::binary-size(32), _::binary>> = + original + + payload = Codec.encode({"same", {:sample, reading}}) + size = byte_size(payload) + hash = :crypto.hash(:sha256, [previous, <>, payload]) + forged = <<"FBWL001!", size::16, hash::binary, payload::binary, 0::size((470 - size) * 8)>> + damaged = original <> forged <> <<1, 2, 3>> + File.write!(path, damaged) + assert {:error, :duplicate_operation_in_journal} = Durable.start(path, repair_tail: true) + assert File.read!(path) == damaged + end + + test "missing files and a second same-path writer cannot reset or fork the ledger", %{ + path: path + } do + assert {:error, :enoent} = Durable.start(path) + refute File.exists?(path) + assert :ok = Durable.create(path) + pid = start(path) + assert {:error, {:already_started, ^pid}} = Durable.start(path) + assert {:error, :eexist} = Durable.create(path) + end + + test "malformed, excessive and unknown inputs are rejected at the server boundary", %{ + path: path + } do + :ok = Durable.create(path) + pid = start(path) + + for command <- [ + {:sample, %{seq: 0, inlet_ml: 1, outlet_ml: 1}}, + {:sample, %{seq: 1, inlet_ml: 18_446_744_073_709_551_616, outlet_ml: 1}}, + {:sample, %{seq: 1, inlet_ml: 1.0, outlet_ml: 1}}, + {:acknowledge, @key, String.duplicate("x", 65)}, + {:update, :unknown, 1, 1, 1, :none}, + {:execute, "arbitrary"} + ] do + assert {:error, :invalid_command} = GenServer.call(pid, {:submit, "invalid", command}) + end + + assert Durable.status(pid).records == 0 + end + + test "complete-slot corruption and reordering are rejected without repairing evidence", %{ + dir: dir + } do + for mutation <- [:payload, :magic, :reorder] do + path = Path.join(dir, Atom.to_string(mutation)) + :ok = Durable.create(path) + pid = start(path) + + for reading <- Enum.take(Experiment.readings(), 2), + do: Durable.submit(pid, "s#{reading.seq}", {:sample, reading}) + + :ok = Durable.stop(pid) + bytes = File.read!(path) + + damaged = + case mutation do + :payload -> + flip(bytes, 512 + 60) + + :magic -> + flip(bytes, 512) + + :reorder -> + <> = bytes + header <> two <> one + end + + File.write!(path, damaged) + assert {:error, reason} = Durable.start(path, repair_tail: true) + assert reason in [:journal_integrity, :journal_framing] + assert File.read!(path) == damaged + end + end + + test "a well-framed journal for another executable is refused", %{path: path} do + {:ok, config} = Codec.config([], <<0::256>>) + payload = Codec.encode({:header, config}) + size = byte_size(payload) + hash = :crypto.hash(:sha256, [<<0::256>>, <>, payload]) + + File.write!( + path, + <<"FBWL001!", size::16, hash::binary, payload::binary, 0::size((470 - size) * 8)>> + ) + + assert {:error, :engine_fingerprint_mismatch} = Durable.start(path) + end + + test "limitation witness: truncation to a complete older prefix needs an external anchor", %{ + path: path + } do + :ok = Durable.create(path) + pid = start(path) + seed(pid) + assert {:ok, _} = Durable.submit(pid, "ack", {:acknowledge, @key, "operator-A"}) + :ok = Durable.stop(pid) + bytes = File.read!(path) + File.write!(path, binary_part(bytes, 0, byte_size(bytes) - 512)) + recovered = start(path) + assert WaterLeak.outstanding(Durable.snapshot(recovered).water) == [@key] + end + + defp start(path, options \\ []) do + assert {:ok, pid} = Durable.start(path, options) + on_exit(fn -> if Process.alive?(pid), do: Durable.stop(pid) end) + pid + end + + defp seed(pid) do + for reading <- Enum.take(Experiment.readings(), 6) do + if reading.seq == 5 do + assert {:ok, :update_accepted} = + Durable.submit(pid, "update", {:update, :v2, 1, 1, 1, :none}) + end + + assert {:ok, _} = Durable.submit(pid, "s#{reading.seq}", {:sample, reading}) + end + end + + defp permutations([]), do: [[]] + + defp permutations(xs), + do: for(x <- xs, tail <- permutations(List.delete(xs, x)), do: [x | tail]) + + defp flip(binary, offset) do + <> = binary + before <> <> <> after_bytes + end +end