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