diff --git a/README.md b/README.md index 78e989d..5ce4f0b 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,17 @@ which brings up `postgres:17` on `localhost:5432` with user/password `config/test.exs` for the defaults) if a server is already running elsewhere. +## Surviving a restart + +`docs/restart-demo.md` walks through the demo embedder that drives this +package's whole surface across a simulated restart with no Session +process: persist mid-run with a pending durable timer and an in-flight +async invocation, drop everything volatile, cold-boot from the run id +alone, and finish with zero duplicate side effects and a replay that +reproduces the path. The executable version lives in +`test/statifier_persistence/demo/restart_demo_test.exs` (and its +Postgres variant beside it). + ## The contract this package builds on The persisted-position story is already specified upstream, and this package diff --git a/docs/adr/0004-run-lifecycle-executor-seam-and-serialization.md b/docs/adr/0004-run-lifecycle-executor-seam-and-serialization.md index 3cf11d7..3ec4cb4 100644 --- a/docs/adr/0004-run-lifecycle-executor-seam-and-serialization.md +++ b/docs/adr/0004-run-lifecycle-executor-seam-and-serialization.md @@ -196,3 +196,24 @@ exit from `fun` (a raise included) releases them with the transaction. The rowless hole and the fix are pinned by a live two-connection test outside the SQL sandbox, whose single shared connection would otherwise serialize the callers by ownership and mask a broken lock. + +## Validation note (2026-08-22, sp-4an.4): driven end to end by a demo embedder + +Decisions 3, 4 and 6 were exercised end to end by a demo embedder +(`test/statifier_persistence/demo/`, walkthrough in +`docs/restart-demo.md`) running a multi-step chart across a simulated +restart with no Session process - persist mid-run with a pending durable +timer and an in-flight async invocation, drop everything volatile, boot +from the run id alone, recover, finish - over both adapters, with the +executor call log asserted on exact contents and a replay reproducing +the path struct for struct. Not an amendment: nothing decided here +changes. + +One finding about the API surface: a byte-identical replay needs the +session id, the one input `Runs.create/4` otherwise generates fresh +(`MachineState.new/2` stamps it into the `:datamodel_init` effect's +`_sessionid`/`_ioprocessors` system variables). The existing +`initialize: [session_id: ...]` pass-through already covers it - no new +surface needed - but a host that wants replayable runs must record that +id alongside its input tape, which `docs/restart-demo.md` now says out +loud. diff --git a/docs/plans/260822-sp-4an.4-restart-demo-host.md b/docs/plans/260822-sp-4an.4-restart-demo-host.md new file mode 100644 index 0000000..e5b5611 --- /dev/null +++ b/docs/plans/260822-sp-4an.4-restart-demo-host.md @@ -0,0 +1,941 @@ +# Demo host survives a simulated restart (sp-4an.4) Implementation Plan + +## Overview + +Build the charter's validation gate: a demo embedder that runs a multi-step +chart across a simulated restart with no `Statifier.Session` process at all - +persist mid-run, drop every volatile process and struct, cold-boot from the +run id alone, continue, finish. At the kill point the run has one **pending +durable timer** and one **in-flight async invoke**, so the demo shows the +timer re-firing from the durable store and the invocation being +*re-established* through the ADR-0051 handler palette rather than restored, +per st-ADR-0060's "resume restores position, not liveness". Success is +finishing with no duplicate side effects (the executor call log is asserted) +and a replay of the recorded inputs reproducing the same path. Beads issue: +`sp-4an.4`. + +## Current State Analysis + +Everything the demo drives is already on main and gate-green; this bead adds +no library code. + +- The loop: `lib/statifier_persistence/runs.ex` - `create/4`, `step/5`, + `fail/4`, all taking `executor:`, `routes:`, `invoke_types:`, + `serialization:` opts (`runs.ex:95-101`). `step/5` returns + `{:ok, Run.t(), MachineState.t()} | {:discarded, Run.t()} | {:error, error}` + (`runs.ex:141-147`). +- The seam: `lib/statifier_persistence/executor.ex` - one callback + `execute(effect, context) :: :ok | {:error, term()}`, context + `%{run_id: String.t(), content_hash: String.t()}` (`executor.ex:17`, + `:42`). A module **or an arity-2 fun** is accepted (`executor.ex:24`). +- The facade: `lib/statifier_persistence/storage.ex` - `save_chart/3`, + `fetch_chart/2`, `insert_run/5`, `fetch_run/2`, `update_run/5`, + `load_run_position/3` (the only guarded blob -> `MachineState.t()` path, + ADR-0003 decision 2). +- Two adapters pass the same conformance suite: + `storage/in_memory.ex` (Agent) and `storage/ecto.ex` (Postgres, with + `isolate/1` and `lock_run/3`). +- Test fixtures and patterns to model on: + `test/statifier_persistence/runs_test.exs` (chart heredocs, `compile!/1`, + `active_ids/1` at `:205-210`, `failing_executor/1`, sabotage notes on every + test), `test/support/recording_executor.ex`, + `test/support/ecto_hosts.ex`, `test/test_helper.exs` (Postgres harness in + `:manual` sandbox, ADR-0005). +- `coveralls.json` skips `test/support/`, so demo-host modules placed there + carry no coverage obligation; the 90% floor applies to `lib/` only. +- No demo, no example, and no test in this repo yet crosses a restart. + +### Key Discoveries + +Verified against the pinned dep (`deps/statifier`) and, where marked, +by running the scenario end to end before writing this plan. + +- **`active_invocations` survives the blob and is the liveness authority.** + `%MachineState{}.active_invocations` is + `%{{state_index, invoke_index} => invoke_id}` + (`deps/statifier/lib/statifier/machine_state.ex:444`) and is carried + verbatim through `Position.to_binary/1` / `from_binary/2` + (`deps/statifier/lib/statifier/position.ex:229`, `:336`, `:525`). + **Probe-verified**: `%{{2, 0} => "enrich"}` before and after the + round-trip. It records *what was invoked*, never a pid - st-ADR-0060 + decision 7 says it deliberately diverges from the empty live table after a + resume. That divergence is exactly what the demo makes visible. +- **Nothing about a pending delayed send is in the position.** `delay_ms` is + relative and no wall-clock instant is stored anywhere + (`deps/statifier/docs/persistence.md:210-254`). Re-arming is unambiguously + the host's job, off the `SendDelayed`/`Cancel` effect vocabulary it + recorded at schedule time. +- **`routes` and `invoke_types` come back `nil` unconditionally** on decode, + whatever the blob carried (st-ADR-0064, `position.ex:135-142`). `Runs` + already pattern-matches both as `nil` as a tripwire and re-stamps them per + call (`runs.ex:243-248`). This is the mechanical form of "registration does + not survive a restart": the demo host re-supplies the palette on every + single step. +- **The process-less `done.invoke` shape is documented and pinned.** + `Statifier.Session.done_invocation/3` needs a session, so the demo cannot + use it; it builds the event itself, matching the single construction site + at `deps/statifier/lib/statifier/session.ex:1839-1853`: + `Event.external("done.invoke." <> invoke_id, data: donedata, invokeid: + invoke_id, origin: ..., origintype: ...)`. `session.ex:1838` names this + "the shape `docs/extending.md` documents for a process-less host to match". +- **`Statifier.Invoke.Handler` is a real behaviour to implement** + (`deps/statifier/lib/statifier/invoke/handler.ex`): pure `start/2`, + `cancel/2`, `forward/3` planning callbacks returning instructions, plus the + impure optional `perform/2` that **MUST be idempotent on `invoke_id`** + (`deps/statifier/docs/extending.md:186-199`). A demo host that implements + it is shaped like a real embedder rather than pattern-matching effect tags + inline. +- **`Statifier.Invoke.Types.registered?/2` gates `active_invocations`.** The + interpreter records an active invocation only for a registered type + (`deps/statifier/lib/statifier/invoke/types.ex:41-60`, the shared + classifier ADR-0051 decision 3 requires). So the demo's `myapp:enrich` + invocation is recorded only because `invoke_types` declares it - the + registry and the resume story are one mechanism, not two. +- **`Statifier.Chart.to_binary/1` / `from_binary/1` + (`deps/statifier/lib/statifier/chart.ex`) recompiles the machine from + stored SCXML source on every load.** This gives the cold boot a real + freshly-interned `%Machine{}` - not the pre-restart struct - which is what + makes the identity guard exercise genuine rather than trivially satisfied. +- **A terminal run discards a late timer.** `Runs.step/5` checks the run + record's status before any position decode and returns `{:discarded, run}` + (`runs.ex:158-171`), which is precisely the liveness check + `deps/statifier/docs/durable-timers.md:286` requires of a host before + feeding a fired timer event back. +- **The whole scenario runs today.** A probe drove the exact chart below + through `initialize/2`, a step, `Chart.to_binary/1` + `Position.to_binary/1`, + a fresh `Chart.from_binary/1` + `Position.from_binary/2`, and three more + steps to `status: :done`. Effects observed at each step: + `[send_delayed: "sla-timer", invoke: "enrich"]`, then + `[cancel_invoke: "enrich", send_delayed: "reminder-timer"]`, then + `[cancel: "reminder-timer"]`, then `[done: nil]`. No design risk remains in + the chart. +- **The nearest upstream reference is + `deps/statifier/test/statifier/interpreter_rehydration_test.exs`**, which + compiles the same source twice to simulate a fresh host process. Its + `rehydrate!/2` helper is the shape to copy; it has no durable store, no + timers, and no invokes, which is the gap this bead fills. + +## Desired End State + +`mix quality` runs a demo that reads as an embedder, not as a unit test: + +```elixir +# Node 1 +host = Demo.Host.boot(store, ledger, run_id) +host = Demo.Host.start_run(host) # Runs.create/4 +host = Demo.Host.submit(host, "submit") # -> enriching +# durable: 1 timer row (sla-timer), 1 open invocation row (enrich) +# volatile: 1 armed timer, 1 live worker pid + +# The node dies. Everything volatile goes with it. +Demo.Runtime.stop() + +# Node 2 knows only the run id. +host = Demo.Host.boot(store, ledger, run_id) # chart blob -> recompiled Machine +host = Demo.Host.recover(host) # re-arm timers, re-establish invocations + +host = Demo.Host.finish_invocation(host, "enrich", %{"score" => 7}) +host = Demo.Host.tick(host, :timer.minutes(20)) # sla-timer fires from the durable store +host = Demo.Host.submit(host, "ack") # -> settled (final) + +%Run{status: :completed} = Demo.Host.run(host) +``` + +Verifiable end state: + +1. The run reaches `:completed` through `:done` only (ADR-0004 decision 6). +2. The configuration path is exactly + `intake -> enriching -> cooling -> settling -> (final)`. +3. The worker pid serving `enrich` after `recover/1` is **alive and + different** from the pre-restart pid, and the pre-restart pid is dead. +4. The `sla-timer` that fires after the restart was armed **before** it, and + fired from a durable row, not from a surviving in-memory timer. +5. The executor call log contains exactly one `{:send_delayed, sla-timer}`, + one `{:invoke, enrich}`, one `{:cancel_invoke, enrich}`, one + `{:send_delayed, reminder-timer}` and one `{:cancel, reminder-timer}` - + no effect executed twice across the restart. +6. The host's own side-effect log (its idempotency table, keyed by + `{run_id, ordinal}` for timers and `invoke_id` for invocations) has one + row per key even though `recover/1` re-ran the invocation start. +7. Replaying the recorded input tape against a fresh store and a fresh run id + reproduces the identical configuration sequence and the identical effect + sequence, field for field. +8. All of the above run green against `Storage.InMemory` **and** + `Storage.Ecto`/Postgres. + +## What We're NOT Doing + +- **No new mix dependencies, and specifically no `oban` / `statifier_oban`.** + The criteria are fully achievable without one: the demo host owns durable + timer rows in its own store and re-arms them after the restart, which is + the pattern `deps/statifier/docs/durable-timers.md:210` calls "Route B: a + process-less host". Pulling Oban in would be an out-of-scope cross-repo + contract question, and would replace the thing under test with a + third-party scheduler. +- **No library code.** `lib/` is untouched. `Runs`, `Executor`, `Storage`, + `Serialization` and both adapters are the subjects, not the deliverables. + If the demo exposes a real gap in one of them, that is a finding to file as + a new bead, not to patch inside this one. +- **No `.quality.exs` edits, no version bump, no release.** +- **No wall-clock timers.** The demo drives a monotonic mock clock + (`Demo.Runtime.advance/1`), so the gate stays fast and deterministic. A + `Process.send_after` variant would prove nothing the mock clock does not + and would make the suite time-dependent. Recorded here rather than left + implicit because it is the one place the demo is unlike a production host. +- **No graduation of `Demo.*` into `lib/statifier_persistence/testing/`.** + ADR-0003 decision 5's precedent exists for a downstream *adapter* to reuse + a conformance suite; nothing downstream asks to reuse a restart scenario + yet, and shipping it in `lib/` would create public surface with no + consumer. See Open Questions. +- **No changelog fragment.** Per `changelog.d/README.md`, tests and docs with + no public surface do not get one. If a reviewer disagrees, the fragment is + a one-line addition. +- **No new ADR.** This bead validates ADR-0004's decisions 3, 4 and 5; it + does not decide anything new. Phase 5 records that validation as a dated + note on ADR-0004 rather than opening ADR-0006. +- **No crash-injection / at-least-once test.** `runs_test.exs:645-681` + already pins re-driven effects re-emitting identical deterministic keys via + `FlakyAdapter`. The demo's restart is a clean one, between steps; adding a + dirty restart is a separate scenario worth its own bead. + +## Implementation Approach + +Five phases. Phases 1-3 build the demo against `Storage.InMemory` and are +each independently committable and full-gate green: Phase 1 is a +straight-through run with no restart (which proves the host, the ledger and +the seam), Phase 2 adds the restart, Phase 3 adds replay determinism and the +identity-guard refusal. Phase 4 re-runs the identical scenario over Postgres +by extracting the body into a shared scenario module. Phase 5 is docs. + +The demo host is split along the axis the bead is about - what survives a +restart and what does not: + +| Module | Lives in | Survives a restart | +|---|---|---| +| `Demo.Ledger` | the host's own durable store (Agent) | **yes** - it is the host's database table | +| `StatifierPersistence.Storage` | the package's store (InMemory Agent / Postgres) | **yes** | +| `Demo.Runtime` | a `Supervisor` over volatile children | **no** - stopped and rebuilt | +| `Demo.Host` | a plain struct, rebuilt by `boot/1` | **no** | +| the `%Machine{}` | recompiled from the stored chart blob | **no** | + +Everything the host does to the outside world goes through the +`StatifierPersistence.Executor` seam - one arity-2 fun built by +`Demo.Host.executor/1` - so no phase inlines an effect. + +All demo modules live under `test/support/demo/` (compiled only in `:test` +via `elixirc_paths/1`, and skipped by `coveralls.json`). The demo tests are +ordinary `mix test` tests, so `mix quality` - the acceptance gate - runs +them; that is why the demo is a test rather than a script under `examples/`, +which nothing in the gate would execute. + +Timer and invocation services never call `Runs` themselves; they return due +work to the driving process, which steps. This keeps every storage call in +the test process, which is what makes the Postgres variant work under the +SQL sandbox in Phase 4. + +### The demo chart + +Probe-verified end to end (see Key Discoveries). It lives as a heredoc in +`test/support/demo/scenario.ex`, following `runs_test.exs`'s fixture style, +and uses only firewall-safe vocabulary (`myapp:enrich`, per the umbrella's +`docs/terminology-firewall.md`): + +```xml + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +``` + +The kill point is `enriching`: `sla-timer` pending, `enrich` in flight. The +`escalated` state is the negative target - reaching it means the demo lost +the race it is supposed to control, so asserting the path never touches it is +a real assertion, not decoration. `reminder-timer` exists so the tail +exercises `{:cancel, _}` on a timer that was armed *after* the restart and +must never fire. + +--- + +## Phase 1: the demo host, the ledger, and a straight-through run + +### Overview + +Everything except the restart: the durable ledger, the volatile runtime, the +ADR-0051 handler, the executor seam, and one test that drives the chart from +`Runs.create/4` to `:completed` in a single process. This phase alone proves +the host is shaped like an embedder and that the seam carries every effect. + +### Changes Required + +#### 1. The durable host store + +**File**: `test/support/demo/ledger.ex` (new) - +`StatifierPersistence.Demo.Ledger` + +**Changes**: An `Agent` standing in for the embedder's own tables. It is +started per test with `start_supervised!/1` and its pid is carried on the +`Demo.Host` struct, so tests stay `async: true`. `@spec` on every public +function; every function returns `:ok` or a value, never a bare raise. + +State: + +``` +%{ + timers: %{{run_id, ordinal} => %{send_id:, event:, data:, due_at_ms:, ordinal:}}, + invocations: %{{run_id, invoke_id} => %{type:, params:, status: :open | :done}}, + side_effects: [{op, key}], # append-only, the idempotency table + executor_calls: [{effect, context}] # the call log the bead asks to assert +} +``` + +API: `start_link/1`, `arm_timer/3`, `cancel_timer/3` (by `{run_id, send_id}` - +spec 6.3's cancel key may match more than one row), `drop_timer/3`, +`open_timers/2`, `open_invocations/2`, `record_invocation/3`, +`close_invocation/3`, `record_call/3`, `calls/1`, +`record_side_effect/2`, `side_effects/1`. + +`arm_timer/3` and `record_invocation/3` are **idempotent on their key** - +`{run_id, ordinal}` (st-ADR-0059's compact dedup key) and `invoke_id` +respectively - and append to `side_effects` only on a genuinely new key. +That list is what makes "no duplicate side effects" checkable independently +of the effect log. + +#### 2. The volatile runtime + +**File**: `test/support/demo/runtime.ex` (new) - +`StatifierPersistence.Demo.Runtime` + +**Changes**: A `Supervisor` over two `Agent` children, plus a mock clock. +Stopping the supervisor is the simulated node death. + +- `Demo.Runtime.Timers` - `%{now_ms: integer(), armed: MapSet.t(key)}`. + `arm/2`, `disarm/2`, `armed/1`, `advance/2` (advances `now_ms` and returns + the armed keys whose `due_at_ms` has passed, looked up against the ledger + by the caller). +- `Demo.Runtime.Workers` - `%{invoke_id => pid()}`, where each worker is a + real spawned `Agent` so "the process died with the node" and "a different + process serves it now" are both directly observable. + +API: `start_link/1`, `stop/1`, `arm/3`, `disarm/3`, `due/3`, +`start_worker/3`, `worker/2`, `stop_worker/2`, `now_ms/1`. + +#### 3. The invoke handler + +**File**: `test/support/demo/enrich_handler.ex` (new) - +`StatifierPersistence.Demo.EnrichHandler` + +**Changes**: `@behaviour Statifier.Invoke.Handler`. `start/2` plans +`[{:handler, __MODULE__, {:start, invoke_id, params}}]`; `cancel/2` plans +`[{:handler, __MODULE__, {:cancel, invoke_id}}]`; `forward/3` returns +`{:ok, []}` (this handler's jobs take no autoforwarded events). + +`perform/2` is **not** implemented here, because it needs the ledger and the +runtime and `ctx` carries neither (`ctx` is +`%{session_id:, invoke_types:, invoke_handlers:}`, +`deps/statifier/lib/statifier/invoke/handler.ex`). `Demo.Host` performs the +planned instructions itself, which is exactly what a process-less host does +and is where the idempotency-on-`invoke_id` obligation is honored. + +#### 4. The host + +**File**: `test/support/demo/host.ex` (new) - +`StatifierPersistence.Demo.Host` + +**Changes**: A struct +`%Host{store, ledger, runtime, machine, run_id, invoke_handlers, +invoke_types, tape, run}` plus: + +- `boot/4` - cold boot from `run_id` alone: `Storage.fetch_run/2` for the + `content_hash`, `Storage.fetch_chart/2` for the blob, + `Statifier.Chart.from_binary/1` for a **freshly compiled** `%Machine{}`. + Returns `{:ok, %Host{}} | {:error, term()}`. On the very first boot (no run + yet) `start_run/2` compiles from source, `Storage.save_chart/3`s the + `Chart.to_binary/1` blob, then `Runs.create/4`. +- `executor/1` - returns the arity-2 fun `Runs` will call. It records every + `{effect, context}` into the ledger, then dispatches: + + | effect | host action | + |---|---| + | `{:send_delayed, %SendDelayed{}}` | `Ledger.arm_timer` (durable, idempotent on `{run_id, ordinal}`), then `Runtime.arm` | + | `{:cancel, %Cancel{}}` | `Ledger.cancel_timer` by `{run_id, send_id}`, then `Runtime.disarm` | + | `{:invoke, %Invoke{}}` | look `type` up in `invoke_handlers`; `handler.start/2`; perform the instructions (`Ledger.record_invocation` + `Runtime.start_worker`) | + | `{:cancel_invoke, %CancelInvoke{}}` | `handler.cancel/2`; perform (`Ledger.close_invocation` + `Runtime.stop_worker`) | + | `{:send, _}` | `Ledger.record_side_effect` only - nothing external to reach in a demo | + | `{:log, _}`, `{:datamodel_*, _}`, `{:trace, _}` | `:ok`, observational | + + An `{:invoke, _}` whose type is in no palette entry returns + `{:error, {:no_handler, type}}`, so the seam stays honest rather than + silently succeeding. Unreachable on the happy path. +- `submit/3` - appends the event to `tape`, then + `Runs.step(store, run_id, machine, event, executor: executor(host), + invoke_types: host.invoke_types, routes: Routes.new())`. Handles all three + result arms; `{:discarded, run}` is recorded, not raised. +- `finish_invocation/4` - builds the `done.invoke.` event in + `session.ex:1839`'s exact shape and submits it. +- `tick/2` - `Runtime.due/3` -> for each due key, load the ledger row, + submit `Event.external(event, sendid: send_id, data: data)`, and drop the + durable row **after** the step returns. A `{:discarded, _}` (terminal run) + drops the row without stepping, which is + `deps/statifier/docs/durable-timers.md:286`'s liveness rule. +- `run/1`, `position/1` (through `Storage.load_run_position/3`), + `config/1` (leaf ids, sorted, modelled on `runs_test.exs:205-210`), + `tape/1`. + +#### 5. The scenario and the first test + +**File**: `test/support/demo/scenario.ex` (new) - +`StatifierPersistence.Demo.Scenario` + +**Changes**: The chart heredoc, `source/0`, `machine!/0`, `invoke_types/0` +(`InvokeTypes.new(types: ["myapp:enrich"])`), `handlers/0` +(`%{"myapp:enrich" => Demo.EnrichHandler}`), and `straight_through/1` - the +scenario body parameterized on a `{adapter, opts}` pair so Phase 4 can reuse +it unchanged. + +**File**: `test/statifier_persistence/demo/restart_demo_test.exs` (new, +`async: true`) + +**Changes**: `setup` starts the ledger, the runtime, and +`Storage.new(InMemory, [])`. One test: the whole chart driven with no +restart, asserting the configuration after every step, `status: :completed`, +that `escalated` was never entered, and the exact executor call log. + +Sabotage note above it per the repo convention. **The mutation must be in +`lib/`** - the convention says break the code the test covers, and what this +test covers is `Runs`/`Storage`, not the demo host. Mutating a `test/support/` +module proves nothing about the library. The mutation for this test: +`Runs.run_status/2` (`runs.ex:464-471`) returns `:active` instead of +`:completed` when `machine_state.status == :done` - the final +`status: :completed` assertion then goes red, and nothing else in the suite +depends on that clause in a way that would mask it. + +### Success Criteria + +#### Automated Verification: +- [x] Full quality gate passes (`mix quality`), coverage floor still met. +- [x] `mix gate.verify` attests the run was not narrowed. +- [x] The new test appears in `mix test --trace` output under the demo + module (a file that fails to be picked up still "passes"). + +#### Manual Verification: +- [x] The sabotage mutation for the new test was confirmed red and reverted, + and the one-line note above the test names the mutation that actually + went red. +- [ ] Reading `host.ex` top to bottom, nothing an embedder would do to the + outside world happens outside `executor/1`. + +**Implementation Note**: Use `mix quality --profile loop` between edits; the +full `mix quality` is the phase gate. In looped execution the Automated +Verification block gates advancement and the Manual items are deferred. + +--- + +## Phase 2: the simulated restart + +### Overview + +Kill the volatile layer mid-run and prove the run finishes anyway: timers +re-fire from the durable store, the invocation is re-established through the +handler palette, and nothing executes twice. + +### Changes Required + +#### 1. Recovery on the host + +**File**: `test/support/demo/host.ex` + +**Changes**: two functions. + +- `recover/1` - the cold-boot obligations st-ADR-0060 decision 7 leaves to + the host, in this order: + 1. **Timers.** `Ledger.open_timers(run_id)` -> `Runtime.arm` each. The + engine is not consulted at all, because the position holds nothing about + them. + 2. **Invocations.** Load the position through + `Storage.load_run_position/3` and read + `machine_state.active_invocations`. For each `{_key, invoke_id}` the + engine still considers active **and** the ledger still shows `:open`, + re-run `handler.start/2` and perform the instructions. The engine is the + liveness authority (which ids are still live), the ledger is the payload + source (what `type` and `params` that id was started with) - neither + alone is enough, and that split is the point of the phase. + 3. Return the rebuilt `%Host{}`. + + `recover/1` performs no step and emits no event; it re-establishes + liveness only. +- `restart/1` - `Runtime.stop/1`, discard the `%Host{}`, start a fresh + `Runtime`, `boot/4`, `recover/1`. It returns a new struct rather than + mutating one, so a test that keeps using the old binding is a test that + fails. + +#### 2. The restart test + +**File**: `test/statifier_persistence/demo/restart_demo_test.exs` + +**Changes**: `Scenario.across_restart/1` in `scenario.ex` plus the test that +calls it. Assertions, in the order the demo reaches them: + +- **At the kill point**: `config == ["enriching"]`; exactly one open ledger + timer row (`sla-timer`); exactly one open invocation row (`enrich`); the + position's `active_invocations` has exactly one entry; `Runtime.worker` + returns a live pid, captured as `pid_before`. +- **After `Runtime.stop/1`**: `Process.alive?(pid_before) == false` and + nothing is armed - the volatile layer is genuinely gone. +- **After `boot/4`**: the machine was rebuilt from stored bytes, not carried + over. Note that `%Machine{}` structs compare **by value**, so `!==` against + the pre-restart struct proves nothing - two compilations of the same source + are equal terms. Assert the path instead: `Demo.Host.boot/4` records a + `{:chart_fetched, content_hash}` marker in the ledger's side-effect log + when it calls `Storage.fetch_chart/2` + `Statifier.Chart.from_binary/1`, + and the test asserts that marker is present for the post-restart boot and + that `machine.identity.content_hash` equals the run record's. That the + freshly interned compilation is usable at all is what + `load_run_position/3`'s guard then proves on the next step. +- **After `recover/1`**: `Runtime.worker("enrich")` is a live pid **different + from `pid_before`** (liveness re-established, not restored), and + `sla-timer` is armed again. +- **The tail**: `finish_invocation("enrich", ...)` -> `["cooling"]` with + `{:cancel_invoke, enrich}` and `{:send_delayed, reminder-timer}` executed; + `tick(:timer.minutes(20))` -> `["settling"]` with + `{:cancel, reminder-timer}` executed and the `reminder-timer` row gone; + `submit("ack")` -> `status: :completed`, `escalated` never entered. +- **No duplicate side effects**: the ledger's `side_effects` list has exactly + one entry per key across the whole run, `recover/1`'s re-run of + `handler.start/2` included; and the executor call log contains exactly one + `{:invoke, _}` and one `{:send_delayed, "sla-timer"}` - the resume did not + re-emit either, which is st-ADR-0060's claim stated as an assertion. + +Each new test carries its sabotage note, and again the mutation goes in +`lib/`. The mutation for this test: `Runs.write_run/6` +(`runs.ex:512-525`) passes `position: :skip` on the `:update` path, so the +stored blob never advances past creation. The restarted host then resumes in +`intake` instead of `enriching`, the recovered timer fires into the wrong +configuration, and the run never completes - red, and red for exactly the +reason the test exists. + +Note for the implementer: `recover/1` skipping the re-arm, or skipping the +invocation re-establishment, are also useful things to break while developing +- they confirm the test is really watching recovery - but they are mutations +of `test/support/`, so neither is the note the convention asks for. + +### Success Criteria + +#### Automated Verification: +- [x] Full quality gate passes. +- [x] `mix gate.verify`. +- [x] `mix test test/statifier_persistence/demo/ --trace` lists the restart + test by name. + +#### Manual Verification: +- [x] Both sabotage mutations confirmed red and reverted, noted above their + tests. +- [ ] The restart test asserts a *different* live worker pid after recovery, + and that the pre-restart pid is dead - not merely that some pid exists. +- [ ] The executor call log assertion is on exact contents, not on a count. +- [ ] Read the test as prose: an outsider can follow "persist, drop, load, + continue, finish" without reading the support modules. + +**Implementation Note**: Same loop/full-gate discipline as Phase 1. + +--- + +## Phase 3: replay determinism and the identity-guard refusal + +### Overview + +The second half of the bead's success definition: the recorded inputs, +replayed, reproduce the same path - and a replay against the wrong chart +revision is refused rather than silently resuming the wrong configuration. + +### Changes Required + +#### 1. Replay + +**File**: `test/support/demo/scenario.ex` + +**Changes**: `replay/3` - given a tape and a fresh `{store, ledger}` and a +fresh `run_id`, drive `Runs.create/4` then `Runs.step/5` for each recorded +event, with **no `Demo.Runtime` at all**: no timers, no workers, nothing but +the tape. It collects the configuration after each step and the effects the +executor saw. + +**File**: `test/statifier_persistence/demo/restart_demo_test.exs` + +**Changes**: a test that runs `across_restart/1`, takes `Host.tape/1`, calls +`replay/3` against a brand-new store, and asserts: + +- the configuration sequences are equal, element for element; +- the effect sequences are equal, **struct for struct** - which holds because + the deterministic keys (`send_id`, `macrostep`/`microstep`/`round`, + `ordinal` from st-ADR-0054 decision 3 and st-ADR-0059) are pure fold state. + Only the executor `context` differs, because `run_id` differs; the test + compares effects, not contexts, and says so in a comment. + +#### 2. The wrong-revision refusal + +**File**: `test/statifier_persistence/demo/restart_demo_test.exs` + +**Changes**: a test that reaches the kill point, then boots a host whose +machine is a **different compilation** - the same chart with one extra state, +following `StatifierPersistence.Testing.Charts`' `chart_a`/`chart_b` rationale +(`lib/statifier_persistence/testing/charts.ex:8-15`) - and asserts +`Runs.step/5` returns `{:error, {:identity_mismatch, expected, actual}}` +with both identities present, and that the stored position is unchanged +afterwards. This is the demo's version of the guard ADR-0003 decision 2 makes +structural, exercised through the lifecycle rather than through the facade +directly. + +Sabotage notes on both tests, both mutations in `lib/`: + +- **Mismatch test**: `Storage.load_run_position/3` collapses the + `{:error, {:identity_mismatch, expected, actual}}` arm into + `{:error, :chart_not_found}` - the arm-collapsing ADR-0003 decision 4 + forbids. The assertion on both identities goes red. +- **Replay test**: the same `Runs.write_run/6` `position: :skip` mutation + Phase 2 uses. It is asymmetric between the two sides - the original run + crosses a restart and reloads the stored blob, the replay never reloads - + so the two configuration sequences diverge and the test goes red. A + mutation that hits both sides identically (reordering + `Runs.execute_effects/3`, for one) cannot sabotage a determinism test and + is not a candidate; that asymmetry is worth stating in the note. + +### Success Criteria + +#### Automated Verification: +- [x] Full quality gate passes. +- [x] `mix gate.verify`. +- [x] `mix test test/statifier_persistence/demo/ --trace` lists both new + tests by name. + +#### Manual Verification: +- [x] Sabotage mutations confirmed red and reverted. +- [ ] The replay assertion compares full effect structs, not tags or counts. +- [ ] The mismatch test asserts both identities in the error arm, per + ADR-0003 decision 4 (no arm collapsed). +- [ ] The replay test would fail if the effect fold stopped being pure - + confirm by checking that at least one counter-derived field + (`ordinal`) is part of the compared structs. + +**Implementation Note**: Same loop/full-gate discipline. + +--- + +## Phase 4: the same demo over Postgres + +### Overview + +Run the identical scenario against `Storage.Ecto` so the demo proves the +loop, not the `InMemory` adapter. The harness already exists (ADR-0005), so +this is parameterization plus one test module. + +### Changes Required + +#### 1. Parameterize the scenario + +**File**: `test/support/demo/scenario.ex` + +**Changes**: every scenario function already takes a `store`; confirm nothing +in `Demo.Host`, `Demo.Ledger` or `Demo.Runtime` names `InMemory`. The ledger +stays an `Agent` in both variants - it models the *host's* store, and +ADR-0003 decision 1 gives it no place in the adapter behaviour. + +#### 2. The Postgres variant + +**File**: `test/statifier_persistence/demo/restart_demo_ecto_test.exs` +(new, `async: false`) + +**Changes**: `setup` mirrors +`test/statifier_persistence/storage/ecto_conformance_test.exs`: +`Storage.new(Storage.Ecto, persistence: EctoHosts.Default, sandbox: true)`. +Then the same three scenario calls Phase 1-3 made - straight-through, across +restart, replay - against the Ecto store. + +Because the timer and worker services never call `Runs` themselves (see +Implementation Approach), every storage call still happens in the test +process, so the SQL sandbox's connection ownership is satisfied without +`allow/3` gymnastics. + +The default `serialization:` strategy is `AdapterLock`, so this variant also +exercises `Storage.Ecto.lock_run/3` around every step of the demo - the +ADR-0004 amendment's advisory-plus-row lock, driven by a realistic +sequence rather than by the conformance suite's synthetic contention. + +### Success Criteria + +#### Automated Verification: +- [x] Full quality gate passes with the Postgres server up + (`docker compose up -d db`). +- [x] `mix gate.verify`. +- [x] `mix test test/statifier_persistence/demo/ --trace` lists the Ecto + variant's tests as well as the InMemory ones - a variant that silently + failed to compile into the suite still "passes". + +#### Manual Verification: +- [x] The Ecto variant genuinely re-reads from Postgres after the restart: + confirm by observing that `boot/4` issues a `fetch_run` and a + `fetch_chart` (log the repo, or step it once by hand). +- [x] Rows are cleaned up / rolled back by the sandbox; a second `mix test` + run in a row is green. + +**Implementation Note**: Same loop/full-gate discipline. This is the phase +that is red on a machine with no database - that is ADR-0005's accepted +posture, not a defect. + +--- + +## Phase 5: documentation + +### Overview + +Make the demo findable and explain what it proves, and record on ADR-0004 +that its decisions were validated against a real pipeline - the charter's +"the first production embedder drives the API" rule. + +### Changes Required + +#### 1. A walkthrough + +**File**: `docs/restart-demo.md` (new) + +**Changes**: A how-to/explanation page: the chart, the kill point, the five +things the host must own that the position does not restore, and the table +from Implementation Approach mapping each module to whether it survives. +Every claim points at the test as executable truth rather than repeating +code. It states the boundary plainly: the durable timer store here is the +host's own; a production host would hand `SendDelayed`/`Cancel` to a real +scheduler (statifier_oban's charter, st-ADR-0054), and the seam is the same +either way. + +#### 2. README pointer + +**File**: `README.md` + +**Changes**: A short "Surviving a restart" subsection linking +`docs/restart-demo.md` and naming the test file. + +#### 3. ADR-0004 validation note + +**File**: `docs/adr/0004-run-lifecycle-executor-seam-and-serialization.md` + +**Changes**: A dated note (2026-08-22, sp-4an.4) under Consequences +recording that decisions 3, 4 and 6 were driven end to end by a demo +embedder across a simulated restart with no Session process, naming the test +file, and recording the one finding the demo produced about the API surface +(or explicitly "none", if none). Not an amendment - nothing decided here +changes. + +No changelog fragment: docs and tests, per `changelog.d/README.md`. + +### Success Criteria + +#### Automated Verification: +- [x] Full quality gate passes (docs-only diff, but run it - the repo's + authority table requires a green full gate before any commit that + touches Elixir, and this phase may touch none, in which case review of + the diff is the bar). +- [x] The umbrella's terminology scan + (`docs/terminology-firewall.md`) is clean over the full diff before + any push. + +#### Manual Verification: +- [ ] `docs/restart-demo.md` reads as a document for a host author, not a + changelog of this bead. +- [ ] The ADR note names a concrete finding or explicitly says there was + none. + +**Implementation Note**: Docs-only phases still run the full gate before the +commit, per this repo's `CLAUDE.md`. + +--- + +## Testing Strategy + +### Unit Tests + +The demo *is* the test; there are no separate unit tests for the demo +modules, which live in `test/support/` and are exercised only through the +scenario (and are excluded from coverage by `coveralls.json`). Coverage of +`lib/` can only rise: the demo drives `Runs.create/4`, `Runs.step/5`, +`Storage.save_chart/3`, `fetch_chart/2`, `fetch_run/2`, +`load_run_position/3`, `Serialization.AdapterLock` and both adapters through +a path no existing test takes. + +Every test asserting `lib/` behavior carries a sabotage note per the repo +convention, and **every mutation is in `lib/`** - breaking a `test/support/` +demo module proves nothing about the library and does not satisfy the +convention. The mutations, named definitively so the implementer runs a +decided one rather than hunting: + +| Test | Mutation (all in `lib/statifier_persistence/`) | Why it goes red | +|---|---|---| +| Phase 1, straight-through | `Runs.run_status/2` returns `:active` for a `:done` machine state | the `status: :completed` assertion fails | +| Phase 2, restart | `Runs.write_run/6` passes `position: :skip` on `:update` | the reload resumes in `intake`; the run never completes | +| Phase 3, replay | the same `write_run/6` mutation - asymmetric, because only the original side reloads a blob | the two configuration sequences diverge | +| Phase 3, mismatch | `Storage.load_run_position/3` collapses `{:identity_mismatch, _, _}` into `:chart_not_found` | the both-identities assertion fails | +| Phase 4, Ecto variant | reuses the Phase 1-3 notes; the scenario body is shared, so no new mutation is needed | - | + +The convention still requires each mutation to be *run* and confirmed red +before the note is written; naming them here removes the search, not the +verification. + +Key edge cases the demo covers deliberately: a timer firing into a terminal +run (`{:discarded, run}`), a `` on a timer armed after the restart, +and a `done.invoke` for an invocation whose worker is a different process +than the one that started it. + +### Manual Testing Steps + +1. `docker compose up -d db`, then `mix quality` - the full gate, both + variants. +2. `mix test test/statifier_persistence/demo/ --trace` and read the test + names: confirm both the InMemory and the Ecto variants are listed, and + that the count matches what the phases added. +3. Apply each sabotage mutation named above, confirm red, revert. +4. Read `docs/restart-demo.md` against the test and confirm no claim in the + prose is unsupported by an assertion. + +## Open Questions + +These do not block implementation - the plan proceeds without resolving any +of them - but they were raised during planning and are recorded here rather +than dropped, because no human was available to decide them. + +1. **Should this scenario be mirrored into statifier_oban?** The demo + deliberately uses the host's own durable timer rows and a mock clock, not + Oban. The natural follow-up is for `sob-2hx` (delayed sends through Oban) + to re-run this exact scenario with a real scheduler as its acceptance + test. That would be a mirrored bead pair, which per the umbrella's rules + needs the other tracker read first. Not filed here. +2. **Should `Demo.*` graduate into + `lib/statifier_persistence/testing/`?** ADR-0003 decision 5 set the + precedent for shipping test-side surface in `lib/` when a downstream + adapter needs it. A restart *scenario* has no downstream consumer yet, so + the plan keeps it in `test/support/`. Revisit when a second storage + adapter or a host wants to run it. +3. **Does anything about the demo warrant an ADR?** The plan assumes not - + it validates ADR-0004 rather than deciding anything. If the implementer + hits a real API gap (Phase 5 asks for the finding to be named explicitly), + that finding may deserve a bead, and possibly an ADR amendment. +4. **Is a changelog fragment wanted anyway?** `changelog.d/README.md` says + no for tests and docs. Recorded because a reviewer who considers + `docs/restart-demo.md` user-facing documentation of a capability might + disagree; adding one is a one-line change. + +## References + +- Bead: `sp-4an.4` (child of the `sp-4an` charter; ordered after `sp-4an.2` + and `sp-4an.3`, which are done). +- This repo: `docs/adr/0003-storage-adapter-behaviour-and-the-identity-guard.md` + (decisions 1, 2, 4, 5), `docs/adr/0004-run-lifecycle-executor-seam-and-serialization.md` + (decisions 3, 4, 5, 6 and the 2026-08-22 lock amendment), + `docs/adr/0005-ecto-in-package-and-postgres-test-harness.md`. +- Prior plans whose phase/gate discipline this one mirrors: + `docs/plans/260822-sp-4an.2.1-run-lifecycle-executor-seam-stepper.md`, + `docs/plans/260822-sp-4an.3.1-ecto-storage-adapter.md`. +- Upstream: `deps/statifier/docs/persistence.md` (the resume recipe and what + it does not restore), `deps/statifier/docs/extending.md` (the handler + behaviour, per-session registration, the process-less `done.invoke` shape, + the idempotency obligation), `deps/statifier/docs/durable-timers.md` + (Route B, the dedup and cancel keys, the liveness check before feeding a + fired timer back), `deps/statifier/docs/adr/0051`, `0052`, `0060`, `0064`. +- Similar implementation to copy from: + `deps/statifier/test/statifier/interpreter_rehydration_test.exs` + (`rehydrate!/2`), `test/statifier_persistence/runs_test.exs:205-210` + (`active_ids/1`), `test/support/recording_executor.ex`, + `test/statifier_persistence/storage/ecto_conformance_test.exs` (the Ecto + setup shape). + +## Deferred Manual Verification + +Manual verification items are deferred during looped (--loop) execution and +surfaced here once, rather than blocking after each phase. Confirm these +before considering the plan fully landed. + +Status note (2026-08-22, takeover worker): the checked items below are the +ones literally executed during implementation - each sabotage mutation was +applied, confirmed red on the intended assertion, and reverted, and the +Phase 4 runs were repeated. The unchecked review items were performed by +the implementer but remain deferred for the operator's own verify walk, +per loop discipline. The in-body Manual Verification blocks mirror this +section. + +### Phase 1 + +- [x] The sabotage mutation for the new test was confirmed red and reverted, + and the one-line note above the test names the mutation that actually + went red. +- [ ] Reading `host.ex` top to bottom, nothing an embedder would do to the + outside world happens outside `executor/1`. + +**Implementation Note**: Use `mix quality --profile loop` between edits; the +full `mix quality` is the phase gate. In looped execution the Automated +Verification block gates advancement and the Manual items are deferred. + +--- + +### Phase 2 + +- [x] Both sabotage mutations confirmed red and reverted, noted above their + tests. +- [ ] The restart test asserts a *different* live worker pid after recovery, + and that the pre-restart pid is dead - not merely that some pid exists. +- [ ] The executor call log assertion is on exact contents, not on a count. +- [ ] Read the test as prose: an outsider can follow "persist, drop, load, + continue, finish" without reading the support modules. + +**Implementation Note**: Same loop/full-gate discipline as Phase 1. + +--- + +### Phase 3 + +- [x] Sabotage mutations confirmed red and reverted (the mismatch-collapse + red on the `{:identity_mismatch, _, _}` pattern; the `position: :skip` + red on the config-sequence divergence). +- [ ] The replay assertion compares full effect structs, not tags or counts + (the session id, the one generated create input, is re-supplied via + `initialize: [session_id: ...]` rather than normalized away). +- [ ] The mismatch test asserts both identities in the error arm, per + ADR-0003 decision 4 (no arm collapsed). +- [ ] `ordinal` is part of the compared structs (`SendDelayed`/`Cancel` + carry it), so an impure effect fold fails the comparison. + +--- + +### Phase 4 + +- [x] The Ecto variant genuinely re-reads from Postgres after the restart - + asserted structurally (the `{:chart_fetched, _}` marker plus the + identity match against the stored run record), and the SQL debug log + shows the post-restart `fetch_run`/`fetch_chart` selects and the + per-step `pg_advisory_xact_lock` + `FOR UPDATE` pair. +- [x] Sandbox cleanup confirmed: two consecutive `mix test` runs green. + +--- diff --git a/docs/restart-demo.md b/docs/restart-demo.md new file mode 100644 index 0000000..d9dcc65 --- /dev/null +++ b/docs/restart-demo.md @@ -0,0 +1,105 @@ +# Surviving a restart: the demo host + +`test/statifier_persistence/demo/` is a demo embedder that runs a +multi-step chart across a simulated restart with no `Statifier.Session` +process at all: persist mid-run, drop every volatile process and struct, +cold-boot from the run id alone, continue, finish. It exists to validate +this package's whole surface - the identity guard (ADR-0003), the run +lifecycle and executor seam (ADR-0004), and both storage adapters - the +way the charter demands: driven by an embedder-shaped pipeline, not by +unit tests alone. + +Every claim below is asserted by a test; when the prose and the tests +disagree, the tests win. + +- `test/statifier_persistence/demo/restart_demo_test.exs` - the scenario + over `Storage.InMemory`, with the full per-stage assertions. +- `test/statifier_persistence/demo/restart_demo_ecto_test.exs` - the same + scenario bodies over `Storage.Ecto` and real Postgres. +- `test/support/demo/` - the host the tests drive: `Host`, `Ledger`, + `Runtime`, `EnrichHandler`, `Scenario`. + +## The chart and the kill point + +The chart (`test/support/demo/scenario.ex`) walks +`intake -> enriching -> cooling -> settling -> settled`, with an +`escalated` state as the negative target - reaching it means the host +lost a race it exists to control, and the tests assert it is never +entered. + +The kill point is `enriching`, chosen so the restart happens with the +maximum in flight: + +- a **pending durable timer** - `enriching`'s onentry armed a 900s + `sla-timer`; +- an **in-flight async invocation** - the `myapp:enrich` invoke started a + real worker process. + +At that moment the volatile runtime is stopped - every worker pid and +armed in-memory timer dies with it - and a fresh host is booted from +nothing but the run id. + +## What survives, and what the host must rebuild + +| Piece | Lives in | Survives the restart | +|---|---|---| +| the run's position and status | `StatifierPersistence.Storage` (InMemory / Postgres) | yes | +| the chart blob | `StatifierPersistence.Storage` | yes | +| the host's own timer and invocation rows | `Demo.Ledger` (stands in for the embedder's tables) | yes | +| armed timers, live worker pids | `Demo.Runtime` (supervisor over volatile state) | no - stopped and rebuilt | +| the `%Host{}` struct | plain struct | no - rebuilt by `boot/4` | +| the compiled `%Machine{}` | recompiled from the stored chart blob | no - `Chart.from_binary/1` on boot | +| the handler palette (`invoke_types`, `invoke_handlers`) | per-deployment declaration | no - re-supplied on every step (st-ADR-0064) | + +st-ADR-0060's rule is "resume restores position, not liveness", and the +demo makes each half of that visible: + +1. **Position restores.** `boot/4` re-reads the run record and the chart + blob, recompiles a freshly interned machine, and the next step's + identity guard proves the pair still match. The restart tests assert + the boot really re-read stored bytes rather than reusing a carried + struct. +2. **Timers do not.** Nothing about a pending delayed send survives in + the position - `delay_ms` is relative and no wall-clock instant is + stored. `recover/1` re-arms every open timer from the host's own + durable rows; the `sla-timer` that fires after the restart was armed + before it, and fires from that durable row. +3. **Invocations do not.** The position's `active_invocations` records + *what was invoked*, never a pid. `recover/1` re-establishes each + invocation the engine still considers active and the ledger still + shows open, by re-running the ADR-0051 handler's `start/2` - the + engine is the liveness authority, the host's rows are the payload + source. The tests assert the post-restart worker is a live pid + different from the dead pre-restart one. +4. **Nothing runs twice.** Re-arming and re-establishing are idempotent + on their durable keys (`{run_id, ordinal}` for timers, `invoke_id` + for invocations), and recovery goes through the host's own hands, not + back through the executor seam - so the executor call log after the + restarted run is exactly the straight-through run's, asserted on + exact contents. +5. **The inputs are the host's to record.** `boot/4` restores no input + tape; the recorder carries it. A replay of the recorded events - + plus the one generated create input, the session id - against a + fresh store reproduces the identical configuration sequence and the + identical effect structs, field for field. + +## Where every effect goes + +Nothing the host does to the outside world happens outside the +`StatifierPersistence.Executor` seam: one arity-2 fun +(`Demo.Host.executor/1`) receives every effect, records it, and +dispatches it - durable row first, volatile arm/spawn second. The two +deliberate exceptions are cold-boot obligations, not chart effects: +`boot/4`'s chart-fetched marker and `recover/1`'s re-establishment, +both of which exist precisely because no step emits them. + +## The boundary + +The durable timer store here is the *host's own* (`Demo.Ledger`), and +the mock clock is the demo's one departure from a production host - the +gate must stay fast and deterministic. A production embedder hands the +same `SendDelayed`/`Cancel` effects to a real scheduler instead; that is +statifier_oban's charter (st-ADR-0054). The seam is identical either +way, which is the point: this package stops at the effect vocabulary, +and everything the demo rebuilds by hand is exactly the work a durable +scheduler package takes on. diff --git a/test/statifier_persistence/demo/restart_demo_ecto_test.exs b/test/statifier_persistence/demo/restart_demo_ecto_test.exs new file mode 100644 index 0000000..f3790f8 --- /dev/null +++ b/test/statifier_persistence/demo/restart_demo_ecto_test.exs @@ -0,0 +1,142 @@ +defmodule StatifierPersistence.Demo.RestartDemoEctoTest do + @moduledoc """ + The demo scenarios re-run against `StatifierPersistence.Storage.Ecto` + over real Postgres (the ADR-0005 harness), so the demo proves the loop, + not the `InMemory` adapter (Phase 4 of + `docs/plans/260822-sp-4an.4-restart-demo-host.md`). + + The per-stage assertions live in `StatifierPersistence.Demo.RestartDemoTest`; + this variant drives the identical scenario bodies (`Scenario` names no + storage module) and asserts the same outcomes, plus the one thing only + this variant can prove - that the post-restart boot re-read the chart + and the run from the database, under the default `AdapterLock` + serialization's advisory-plus-row lock on every step. + + `async: false` per the plan: the demo drives several `Storage.new/2` + handles in one test over a single sandbox-checked-out connection. + """ + use ExUnit.Case, async: false + + alias Statifier.Effect.{Cancel, CancelInvoke, DatamodelInit, Invoke, SendDelayed} + alias StatifierPersistence.Demo.{Host, Ledger, Scenario} + alias StatifierPersistence.EctoHosts + alias StatifierPersistence.Run + alias StatifierPersistence.Storage + + @adapter Storage.Ecto + @adapter_opts [persistence: EctoHosts.Default, sandbox: true] + + setup do + # The conformance suite's shape: normalize the opts through + # `Storage.new/2`, then check this test process out its own sandboxed + # connection/transaction. + {:ok, store} = Storage.new(@adapter, @adapter_opts) + :ok = @adapter.isolate(store.opts) + :ok + end + + # The scenario bodies and the mutations are shared with the InMemory + # variant (`RestartDemoTest`); the plan's Testing Strategy table records + # that Phase 4 reuses those notes rather than adding new mutations. The + # per-test notes below name which shared mutation reds each test. + + # sabotage: shared with RestartDemoTest's straight-through test - + # Runs.run_status/2 returning :active for a :done machine state reds the + # `status: :completed` assertion here identically (same scenario body). + # Run and confirmed red on the InMemory variant, reverted. + test "drives the chart straight through over Postgres" do + result = Scenario.straight_through({@adapter, @adapter_opts}) + + assert result.configs == [["enriching"], ["cooling"], ["settling"]] + refute Enum.any?(result.configs, &("escalated" in &1)) + assert %Run{status: :completed} = Host.run(result.host) + + calls = result.ledger |> Ledger.calls() |> Enum.map(fn {effect, _context} -> effect end) + + assert [ + {:datamodel_init, %DatamodelInit{}}, + {:send_delayed, %SendDelayed{send_id: "sla-timer", event: "sla.breach"}}, + {:invoke, %Invoke{type: "myapp:enrich", invoke_id: invoke_id}}, + {:cancel_invoke, %CancelInvoke{invoke_id: invoke_id}}, + {:send_delayed, %SendDelayed{send_id: "reminder-timer", event: "reminder"}}, + {:cancel, %Cancel{send_id: "reminder-timer"}} + ] = calls + end + + # sabotage: shared with RestartDemoTest's restart test - Runs.write_run/6 + # passing position: :skip on the :update path leaves the stored blob in + # intake, so `config_at_kill == ["enriching"]` reds here identically + # (same scenario body). Run and confirmed red on the InMemory variant, + # reverted. + test "resumes from a simulated restart over Postgres" do + result = Scenario.across_restart({@adapter, @adapter_opts}) + + # Kill point, node death, and re-established liveness - same claims + # the InMemory variant pins in full. + assert result.config_at_kill == ["enriching"] + refute result.alive_after_stop + assert is_pid(result.pid_after_recover) + refute result.pid_after_recover == result.pid_before + + # The post-restart boot genuinely re-read from Postgres: the + # `{:chart_fetched, _}` marker is present, and the freshly recompiled + # machine's identity matches the stored run record's - this is the + # plan's "confirm `boot/4` issues a `fetch_run` and a `fetch_chart`" + # check, asserted rather than observed by hand. + assert {:chart_fetched, content_hash} = + result.ledger + |> Ledger.side_effects() + |> Enum.find(&match?({:chart_fetched, _}, &1)) + + assert content_hash == result.host_after_boot.machine.identity.content_hash + assert %Run{content_hash: ^content_hash} = Host.run(result.host_after_boot) + + # The tail finishes with the exact same executor call log - nothing + # re-emitted across the restart, now with every step's write behind + # `Storage.Ecto.lock_run/3`. + assert result.configs == [["cooling"], ["settling"]] + assert %Run{status: :completed} = Host.run(result.host) + + calls = result.ledger |> Ledger.calls() |> Enum.map(fn {effect, _context} -> effect end) + + assert [ + {:datamodel_init, %DatamodelInit{}}, + {:send_delayed, %SendDelayed{send_id: "sla-timer", event: "sla.breach"}}, + {:invoke, %Invoke{type: "myapp:enrich", invoke_id: invoke_id}}, + {:cancel_invoke, %CancelInvoke{invoke_id: invoke_id}}, + {:send_delayed, %SendDelayed{send_id: "reminder-timer", event: "reminder"}}, + {:cancel, %Cancel{send_id: "reminder-timer"}} + ] = calls + end + + # sabotage: shared with RestartDemoTest's replay test - the same + # write_run position: :skip mutation diverges the loaded original + # configs from the replay's returned ones, redding the sequence + # comparison here identically (same scenario body and comparison). Run + # and confirmed red on the InMemory variant, reverted. + test "replays the recorded tape over Postgres" do + result = Scenario.across_restart({@adapter, @adapter_opts}) + + {:ok, replay_store} = Storage.new(@adapter, @adapter_opts) + {:ok, replay_ledger} = Ledger.start_link([]) + replay_run_id = "restart-demo-replay-#{System.unique_integer([:positive])}" + + # Same recorded-inputs discipline as the InMemory variant: the session + # id is the one generated create input, re-supplied to the replay. + {:ok, final_position} = Host.position(result.host) + original_session_id = final_position.datamodel["_sessionid"] + + replay_result = + Scenario.replay(Host.tape(result.host), {replay_store, replay_ledger}, replay_run_id, + initialize: [session_id: original_session_id] + ) + + original_configs = [result.config_at_kill] ++ result.configs ++ [Host.config(result.host)] + assert Enum.drop(replay_result.configs, 1) == original_configs + + original_effects = + result.ledger |> Ledger.calls() |> Enum.map(fn {effect, _context} -> effect end) + + assert replay_result.effects == original_effects + end +end diff --git a/test/statifier_persistence/demo/restart_demo_test.exs b/test/statifier_persistence/demo/restart_demo_test.exs new file mode 100644 index 0000000..23daf8a --- /dev/null +++ b/test/statifier_persistence/demo/restart_demo_test.exs @@ -0,0 +1,259 @@ +defmodule StatifierPersistence.Demo.RestartDemoTest do + use ExUnit.Case, async: true + + alias Statifier.Effect.{Cancel, CancelInvoke, DatamodelInit, Invoke, SendDelayed} + alias Statifier.Event + alias Statifier.Send.Routes + alias StatifierPersistence.Demo.{Host, Ledger, Runtime, Scenario} + alias StatifierPersistence.{Run, Runs, Storage} + alias StatifierPersistence.Storage.InMemory + + # sabotage: Runs.run_status/2 (runs.ex ~465-471) changed from + # `machine_state.status == :done -> :completed` to `-> :active` -> red + # (the `status: :completed` assertion below fails: `Host.run(host).status` + # comes back `:active`). Reverted and confirmed green. + test "drives the chart straight through, with no restart, to :completed" do + result = Scenario.straight_through({InMemory, []}) + + # The configuration path after each non-terminal step, in order: + # intake -> enriching -> cooling -> settling. `escalated` - the + # negative target reached only if the demo lost the race it exists to + # control - is never among them. + assert result.configs == [["enriching"], ["cooling"], ["settling"]] + refute Enum.any?(result.configs, &("escalated" in &1)) + + assert %Run{status: :completed} = Host.run(result.host) + + # The exact executor call log: the create-time `:datamodel_init` + # baseline, one send_delayed per armed timer, one invoke, one + # cancel_invoke on the invocation's exit, and one cancel of the + # reminder-timer the still-armed sla-timer's fire drives - nothing + # executed twice, and `:done` never reaches the executor at all (the + # lifecycle consumes it before the seam). + calls = result.ledger |> Ledger.calls() |> Enum.map(fn {effect, _context} -> effect end) + + assert [ + {:datamodel_init, %DatamodelInit{}}, + {:send_delayed, %SendDelayed{send_id: "sla-timer", event: "sla.breach"}}, + {:invoke, %Invoke{type: "myapp:enrich", invoke_id: invoke_id}}, + {:cancel_invoke, %CancelInvoke{invoke_id: invoke_id}}, + {:send_delayed, %SendDelayed{send_id: "reminder-timer", event: "reminder"}}, + {:cancel, %Cancel{send_id: "reminder-timer"}} + ] = calls + end + + # sabotage: Runs.write_run/6 (runs.ex ~512-525) changed so the `:update` + # write path passes `position: :skip` unconditionally -> red. Every + # post-create step (`step_tail/6` reloads the position from storage on + # every call) then persists nothing, so the stored blob never leaves + # `intake`: this test's very first assertion, `config_at_kill == + # ["enriching"]`, fails immediately (left: `["intake"]`). Confirmed by + # actually running the mutation - it also fails the straight-through + # test the same way, but this test's own kill-point assertion is + # sufficient to make it red on its own. Reverted and confirmed green. + test "resumes from a simulated restart with no duplicate side effects" do + result = Scenario.across_restart({InMemory, []}) + + # --- at the kill point --- + assert result.config_at_kill == ["enriching"] + assert [%{send_id: "sla-timer"}] = result.open_timers_at_kill + assert [%{invoke_id: "enrich", type: "myapp:enrich"}] = result.open_invocations_at_kill + assert result.active_invocations_at_kill == 1 + + assert is_pid(result.pid_before) + assert result.alive_before_stop + assert MapSet.size(result.armed_before_stop) == 1 + + # --- after Runtime.stop/1 --- + refute result.alive_after_stop + assert MapSet.size(result.armed_after_boot) == 0 + + # --- after boot/4 --- + # The machine was rebuilt from stored bytes, not carried over - + # `%Machine{}` compares by value so `!==` against the pre-restart + # struct would prove nothing (two compilations of the same source are + # equal terms). Assert the path instead: `boot/4` recorded that it + # re-fetched the chart, and the freshly compiled machine's identity + # matches the stored run record's. + assert {:chart_fetched, content_hash} = + result.ledger + |> Ledger.side_effects() + |> Enum.find(&match?({:chart_fetched, _}, &1)) + + assert content_hash == result.host_after_boot.machine.identity.content_hash + assert %Run{content_hash: ^content_hash} = Host.run(result.host_after_boot) + + # --- after recover/1 --- + assert is_pid(result.pid_after_recover) + assert result.alive_after_recover + refute result.pid_after_recover == result.pid_before + assert [%{send_id: "sla-timer"}] = result.open_timers_after_recover + + # --- the tail: finish_invocation -> tick -> ack --- + assert result.configs == [["cooling"], ["settling"]] + refute Enum.any?(result.configs, &("escalated" in &1)) + assert %Run{status: :completed} = Host.run(result.host) + + # The reminder-timer row (armed after the restart) is dropped once its + # cancel is driven by the sla-timer's fire - no row of any kind is left + # open once the run settles. + assert Ledger.open_timers(result.ledger, result.run_id) == [] + + # --- no duplicate side effects across the restart --- + # The host's idempotency ledger, on exact contents: one row per key + # even though `recover/1` re-ran `handler.start/2` (the re-arm and the + # re-establishment hit existing keys and appended nothing), plus the + # one `{:chart_fetched, _}` marker the post-restart `boot/4` wrote. + run_id = result.host.run_id + + assert [ + {:arm_timer, {^run_id, sla_ordinal}}, + {:record_invocation, {^run_id, "enrich"}}, + {:chart_fetched, ^content_hash}, + {:arm_timer, {^run_id, reminder_ordinal}} + ] = Ledger.side_effects(result.ledger) + + refute sla_ordinal == reminder_ordinal + + # The executor call log, on exact contents - identical to the + # straight-through run's. The restart added no executor call at all: + # `recover/1` re-establishes liveness host-side, through the handler, + # never back through the seam, so neither the `:invoke` nor the + # sla-timer's `:send_delayed` was re-emitted - st-ADR-0060's "resume + # restores position, not liveness" stated as an assertion. + calls = result.ledger |> Ledger.calls() |> Enum.map(fn {effect, _context} -> effect end) + + assert [ + {:datamodel_init, %DatamodelInit{}}, + {:send_delayed, %SendDelayed{send_id: "sla-timer", event: "sla.breach"}}, + {:invoke, %Invoke{type: "myapp:enrich", invoke_id: invoke_id}}, + {:cancel_invoke, %CancelInvoke{invoke_id: invoke_id}}, + {:send_delayed, %SendDelayed{send_id: "reminder-timer", event: "reminder"}}, + {:cancel, %Cancel{send_id: "reminder-timer"}} + ] = calls + end + + # A chart identical to `Scenario.source/0` plus one extra, unreachable + # state - the same `chart_a`/`chart_b` device + # `StatifierPersistence.Testing.Charts` uses (charts.ex:8-15) to change + # `Statifier.Machine.Identity.of_source/2`'s content hash without + # changing anything about how the chart runs. + @wrong_revision_source Scenario.source() + |> String.replace( + ~r{\s*\z}, + " \n\n" + ) + + # sabotage: Storage.load_run_position/3 (storage.ex ~384-395) changed so + # `precheck_identity/2`'s `{:error, {:identity_mismatch, expected, + # actual}}` arm collapses to `{:error, :chart_not_found}` -> red (the + # pattern match on `{:error, {:identity_mismatch, expected, actual}}` + # below no longer matches the returned `{:error, :chart_not_found}`). + # Reverted and confirmed green. + test "refuses to step a stored run against a different chart revision" do + run_id = "restart-demo-wrong-rev-#{System.unique_integer([:positive])}" + {:ok, store} = Storage.new(InMemory, []) + {:ok, ledger} = Ledger.start_link([]) + {:ok, runtime} = Runtime.start_link([]) + + {:ok, host} = + Host.start_run(store, ledger, runtime, run_id, Scenario.source(), + invoke_handlers: Scenario.handlers(), + invoke_types: Scenario.invoke_types() + ) + + host = Host.submit(host, "submit") + assert Host.config(host) == ["enriching"] + {:ok, position_before} = Host.position(host) + + {:ok, wrong_machine} = Statifier.compile(@wrong_revision_source) + refute wrong_machine.identity.content_hash == host.machine.identity.content_hash + + result = + Runs.step(store, run_id, wrong_machine, Event.external("sla.breach"), + executor: fn _effect, _context -> :ok end, + invoke_types: Scenario.invoke_types(), + routes: Routes.new() + ) + + # Both identities named, per ADR-0003 decision 4 - no arm collapsed to + # a generic reason that drops which chart was stored and which was + # supplied. + assert {:error, {:identity_mismatch, expected, actual}} = result + assert expected == host.machine.identity + assert actual == wrong_machine.identity + + {:ok, position_after} = Host.position(host) + assert position_after == position_before + end + + # A mutation hitting both sides identically (reordering + # `Runs.execute_effects/3`, for one) cannot sabotage this test - the + # restarted run and the replay would still walk identical paths. The + # mutation below is the one that can, because it is asymmetric between + # them: the restarted run crosses a real restart and reloads the stored + # blob on every post-restart step, while `Scenario.replay/3` never + # crosses a restart and never reloads anything but what it itself wrote. + # + # sabotage: Runs.write_run/6 (runs.ex ~512-525) changed so the `:update` + # write path passes `position: :skip` unconditionally -> red. Both sides + # stall (no step's result is ever stored), but asymmetrically: the + # original's *loaded* configs stay `["intake"]` throughout, while the + # replay's configs come off each step's *returned* state, whose first + # element still advances to `["enriching"]` - the sequences diverge and + # the `Enum.drop(replay_result.configs, 1) == original_configs` + # assertion goes red (confirmed by running the mutation; the earlier + # tests in this file go red on their own assertions too). Reverted and + # confirmed green. + test "replaying the recorded tape against a fresh store reproduces the same path" do + result = Scenario.across_restart({InMemory, []}) + + {:ok, replay_store} = Storage.new(InMemory, []) + {:ok, replay_ledger} = Ledger.start_link([]) + replay_run_id = "restart-demo-replay-#{System.unique_integer([:positive])}" + + # The session id is the one non-deterministic input the original + # create took (`MachineState.new/2` generates it; it lands in the + # `:datamodel_init` effect's `_sessionid`/`_ioprocessors` system + # variables). It is part of the recorded inputs, so the replay + # re-supplies it - read here off the finished run's durable position. + {:ok, final_position} = Host.position(result.host) + original_session_id = final_position.datamodel["_sessionid"] + + replay_result = + Scenario.replay(Host.tape(result.host), {replay_store, replay_ledger}, replay_run_id, + initialize: [session_id: original_session_id] + ) + + # `result.config_at_kill` is the config after tape event 1 (`submit`); + # `result.configs` are the configs after tape events 2 and 3 + # (`finish_invocation`, the fired `sla-timer`); `Host.config/1` on the + # finished host is the config after tape event 4 (`ack`) - empty, + # because reaching the top-level `` exits every state. Dropping + # `replay_result.configs`' element zero (the pre-tape config right + # after `Runs.create/4`) lines the two sequences up one-for-one. + original_configs = [result.config_at_kill] ++ result.configs ++ [Host.config(result.host)] + assert Enum.drop(replay_result.configs, 1) == original_configs + + # And pinned literally, so equal-but-both-wrong sequences (a mutation + # that stalls the restarted run and the replay identically) still go + # red rather than sliding through the comparison above. + assert replay_result.configs == + [["intake"], ["enriching"], ["cooling"], ["settling"], []] + + # Struct for struct, not tag for tag or count for count. This holds + # because the deterministic fold state each effect carries - + # `ordinal`, `macrostep`/`microstep`/`round`, `send_id` - comes from + # the counters and the chart's own document-order ids, none of which + # depend on `run_id`; the session id, the one generated input, was + # re-supplied above. Only the executor `context` differs between the + # two runs (it carries `run_id`, and the restarted run and the replay + # use different ones) - which is exactly why this compares the effect + # payloads and not the `{effect, context}` pairs the ledger actually + # recorded. + original_effects = + result.ledger |> Ledger.calls() |> Enum.map(fn {effect, _context} -> effect end) + + assert replay_result.effects == original_effects + end +end diff --git a/test/support/demo/enrich_handler.ex b/test/support/demo/enrich_handler.ex new file mode 100644 index 0000000..a5a5d13 --- /dev/null +++ b/test/support/demo/enrich_handler.ex @@ -0,0 +1,37 @@ +defmodule StatifierPersistence.Demo.EnrichHandler do + @moduledoc """ + Serves `` for the restart demo + (`docs/plans/260822-sp-4an.4-restart-demo-host.md`, Phase 1), following + `deps/statifier/docs/extending.md`'s own `MyApp.EnrichHandler` example. + + `@behaviour Statifier.Invoke.Handler`'s planning callbacks are pure: they + return instructions, they never touch the ledger or the runtime + themselves. `perform/2` is deliberately **not** implemented here - `ctx` + carries `session_id`, `invoke_types`, and `invoke_handlers` only + (`Statifier.Invoke.Handler.ctx/0`), never the ledger or the runtime a + performer would need. `StatifierPersistence.Demo.Host` performs the + planned instructions itself, which is exactly what a process-less host + does and is where the idempotency-on-`invoke_id` obligation + (`docs/extending.md`'s "At-least-once") is actually honored. + """ + + @behaviour Statifier.Invoke.Handler + + alias Statifier.Effect.Invoke + + @impl Statifier.Invoke.Handler + def start(%Invoke{invoke_id: invoke_id, params: params}, _ctx) do + {:ok, [{:handler, __MODULE__, {:start, invoke_id, params}}]} + end + + @impl Statifier.Invoke.Handler + def cancel(invoke_id, _ctx) do + {:ok, [{:handler, __MODULE__, {:cancel, invoke_id}}]} + end + + @impl Statifier.Invoke.Handler + def forward(_invoke_id, _event, _ctx) do + # This handler's jobs take no autoforwarded events. + {:ok, []} + end +end diff --git a/test/support/demo/host.ex b/test/support/demo/host.ex new file mode 100644 index 0000000..1e11be3 --- /dev/null +++ b/test/support/demo/host.ex @@ -0,0 +1,470 @@ +defmodule StatifierPersistence.Demo.Host do + @moduledoc """ + A demo embedder driving a chart with no `Statifier.Session` process at + all (`docs/plans/260822-sp-4an.4-restart-demo-host.md`, Phase 1). + + Every effect a stepped run produces reaches the outside world through + `executor/1`'s arity-2 fun and nowhere else - reading this module top to + bottom, nothing it does to `StatifierPersistence.Demo.Ledger` or + `StatifierPersistence.Demo.Runtime` happens outside that one function and + the private helpers it calls, with two deliberate cold-boot exceptions + that are host obligations rather than chart effects: `boot/4`'s + `{:chart_fetched, _}` observability marker, and `recover/1`'s + re-establishment of timers and invocations from durable state (st-ADR-0060 + leaves both to the host precisely because no step emits them). That is + what makes this host shaped like a real embedder rather than a test + harness with effects inlined. + + A `%Host{}` carries no `Statifier.MachineState.t()` of its own between + calls - the durable position lives in `store`, loaded fresh by + `StatifierPersistence.Runs` on every `submit/3`/`finish_invocation/4`/ + `tick/2`, exactly as a stateless embedder process would. `machine` is the + one piece of compiled, in-memory state a host does keep around: a + `Statifier.Machine.t()` is a pure compiled artifact, not a position, and + re-deriving it from the stored chart is `boot/4`'s job, not every step's. + """ + + alias Statifier.{Chart, Event, Machine, MachineState} + alias Statifier.Effect.{Cancel, CancelInvoke, Invoke, Send, SendDelayed} + alias Statifier.Evaluator.SystemVariables + alias Statifier.Send.Routes + alias StatifierPersistence.Demo.{Ledger, Runtime} + alias StatifierPersistence.{Run, Runs, Storage} + + @enforce_keys [:store, :ledger, :runtime, :machine, :run_id] + defstruct [ + :store, + :ledger, + :runtime, + :machine, + :run_id, + :run, + invoke_handlers: %{}, + invoke_types: nil, + tape: [] + ] + + @type t :: %__MODULE__{ + store: Storage.t(), + ledger: Ledger.t(), + runtime: Runtime.t(), + machine: Machine.t(), + run_id: Runs.run_id(), + run: Run.t() | nil, + invoke_handlers: %{String.t() => module()}, + invoke_types: Statifier.Invoke.Types.t() | nil, + tape: [Event.t()] + } + + @doc """ + Cold-boots from `run_id` alone: `Storage.fetch_run/2` for the content + hash, `Storage.fetch_chart/2` for the stored blob, `Chart.from_binary/1` + to recompile a **freshly interned** `%Machine{}` - never the pre-restart + struct, which is what makes the identity guard on the next step exercise + something real (`Statifier.Chart.to_binary/1`'s moduledoc; the plan's Key + Discoveries). + + Records a `{:chart_fetched, content_hash}` marker on the ledger's own + side-effect log every time it runs, so a test can assert that a + post-restart boot really re-read the chart rather than reusing a carried + struct. + + `invoke_handlers:`/`invoke_types:` are **not** restored here - like + `routes`/`invoke_types` on a decoded `MachineState` (st-ADR-0064), the + handler palette is a per-deployment declaration, not durable state, and a + caller re-supplies it (`%{host | invoke_handlers: ..., invoke_types: ...}`) + before driving the rebuilt host. + """ + @spec boot(Storage.t(), Ledger.t(), Runtime.t(), Runs.run_id()) :: + {:ok, t()} | {:error, term()} + def boot(%Storage{} = store, ledger, runtime, run_id) do + with {:ok, run_record} <- Storage.fetch_run(store, run_id), + {:ok, chart_record} <- Storage.fetch_chart(store, run_record.content_hash), + {:ok, machine} <- Chart.from_binary(chart_record.chart_blob) do + :ok = Ledger.record_side_effect(ledger, {:chart_fetched, run_record.content_hash}) + + {:ok, + %__MODULE__{ + store: store, + ledger: ledger, + runtime: runtime, + machine: machine, + run_id: run_id, + run: Run.from_record(run_record) + }} + end + end + + @doc """ + The very-first-boot path, for a `run_id` with no stored run yet: compiles + `source` fresh, `Storage.save_chart/3`s its `Chart.to_binary/1` blob, then + `Runs.create/4` through this host's own `executor/1` - so even the + creation step's effects (an onentry ``, an onentry ``) + cross the same seam every later step does. + + `opts` accepts `invoke_handlers:` (default `%{}`) and `invoke_types:` + (default `nil`), the palette this run's every future step is driven with. + """ + @spec start_run(Storage.t(), Ledger.t(), Runtime.t(), Runs.run_id(), String.t(), keyword()) :: + {:ok, t()} | {:error, term()} + def start_run(%Storage{} = store, ledger, runtime, run_id, source, opts \\ []) + when is_binary(source) do + invoke_handlers = Keyword.get(opts, :invoke_handlers, %{}) + invoke_types = Keyword.get(opts, :invoke_types) + + with {:ok, machine} <- Statifier.compile(source), + {:ok, chart_blob} <- Chart.to_binary(machine), + :ok <- Storage.save_chart(store, machine, chart_blob) do + host = %__MODULE__{ + store: store, + ledger: ledger, + runtime: runtime, + machine: machine, + run_id: run_id, + run: nil, + invoke_handlers: invoke_handlers, + invoke_types: invoke_types, + tape: [] + } + + case Runs.create(store, run_id, machine, + executor: executor(host), + invoke_types: invoke_types + ) do + {:ok, run, _machine_state} -> {:ok, %{host | run: run}} + {:error, _reason} = error -> error + end + end + end + + @doc """ + The seam: an arity-2 fun `StatifierPersistence.Runs` calls once per + effect. Every call is recorded onto the ledger's executor call log first, + unconditionally, then dispatched per the effect table below - so the log + is complete even for an effect the dispatch itself goes on to fail. + + | effect | host action | + |---|---| + | `{:send_delayed, _}` | `Ledger.arm_timer/3` (durable, idempotent on `{run_id, ordinal}`), then `Runtime.arm/3` | + | `{:cancel, _}` | `Ledger.cancel_timer/3` by `{run_id, send_id}`, then `Runtime.disarm/2` for each row it removed | + | `{:invoke, _}` | look `type` up in `invoke_handlers`; `handler.start/2`; perform the instructions (`Ledger.record_invocation/3` + `Runtime.start_worker/3`) | + | `{:cancel_invoke, _}` | the ledger's own recorded `type` for `invoke_id` finds the handler; `handler.cancel/2`; perform (`Ledger.close_invocation/3` + `Runtime.stop_worker/2`) | + | `{:send, _}` | `Ledger.record_side_effect/2` only - nothing external to reach in a demo | + | everything else (`:log`, `:datamodel_change`, `:datamodel_init`, `:trace`, `:autoforward`) | `:ok`, observational | + + An `{:invoke, _}` or `{:cancel_invoke, _}` whose type resolves to no + registered handler returns `{:error, {:no_handler, type}}` - unreachable + on this chart's happy path, but the seam stays honest rather than + silently succeeding. + """ + @spec executor(t()) :: StatifierPersistence.Executor.t() + def executor(%__MODULE__{} = host) do + fn effect, context -> + :ok = Ledger.record_call(host.ledger, effect, context) + handle_effect(host, effect) + end + end + + @spec handle_effect(t(), Statifier.Effect.t()) :: :ok | {:error, term()} + defp handle_effect(host, {:send_delayed, %SendDelayed{} = payload}) do + due_at_ms = Runtime.now_ms(host.runtime) + payload.delay_ms + + row = %{ + send_id: payload.send_id, + event: payload.event, + data: payload.data, + due_at_ms: due_at_ms, + ordinal: payload.ordinal + } + + :ok = Ledger.arm_timer(host.ledger, host.run_id, row) + :ok = Runtime.arm(host.runtime, payload.ordinal, due_at_ms) + :ok + end + + defp handle_effect(host, {:cancel, %Cancel{send_id: send_id}}) do + host.ledger + |> Ledger.cancel_timer(host.run_id, send_id) + |> Enum.each(&Runtime.disarm(host.runtime, &1)) + + :ok + end + + defp handle_effect(host, {:invoke, %Invoke{type: type} = invoke}) do + case Map.fetch(host.invoke_handlers, type) do + {:ok, handler} -> + {:ok, instructions} = handler.start(invoke, handler_ctx(host)) + Enum.each(instructions, &perform_instruction(host, type, &1)) + :ok + + :error -> + {:error, {:no_handler, type}} + end + end + + defp handle_effect(host, {:cancel_invoke, %CancelInvoke{invoke_id: invoke_id}}) do + type = invocation_type(host, invoke_id) + + case Map.fetch(host.invoke_handlers, type) do + {:ok, handler} -> + {:ok, instructions} = handler.cancel(invoke_id, handler_ctx(host)) + Enum.each(instructions, &perform_instruction(host, type, &1)) + :ok + + :error -> + {:error, {:no_handler, type}} + end + end + + defp handle_effect(host, {:send, %Send{} = payload}) do + Ledger.record_side_effect(host.ledger, {:send, payload}) + end + + defp handle_effect(_host, _observational_effect), do: :ok + + @spec handler_ctx(t()) :: Statifier.Invoke.Handler.ctx() + defp handler_ctx(host) do + %{ + session_id: host.run_id, + invoke_types: host.invoke_types, + invoke_handlers: host.invoke_handlers + } + end + + # The recorded type for an open invocation, so `{:cancel_invoke, _}` - + # which carries no `type` field itself - can still find its handler. + @spec invocation_type(t(), String.t()) :: String.t() | nil + defp invocation_type(host, invoke_id) do + host.ledger + |> Ledger.open_invocations(host.run_id) + |> Enum.find_value(fn row -> row.invoke_id == invoke_id and row.type end) + end + + @spec perform_instruction(t(), String.t() | nil, Statifier.Invoke.Handler.instruction()) :: :ok + defp perform_instruction(host, type, {:handler, _module, {:start, invoke_id, params}}) do + :ok = + Ledger.record_invocation(host.ledger, host.run_id, %{ + invoke_id: invoke_id, + type: type, + params: params + }) + + _worker_pid = Runtime.start_worker(host.runtime, invoke_id, params) + :ok + end + + defp perform_instruction(host, _type, {:handler, _module, {:cancel, invoke_id}}) do + :ok = Ledger.close_invocation(host.ledger, host.run_id, invoke_id) + Runtime.stop_worker(host.runtime, invoke_id) + end + + @doc """ + Appends `Event.external(event_name, event_opts)` to `tape`, then + `Runs.step/5`. Handles all three result arms: `{:discarded, run}` is + recorded onto `run`, not raised - an event delivered to a run that went + terminal on a prior step is exactly what a durable host must tolerate. + """ + @spec submit(t(), String.t(), keyword()) :: t() + def submit(%__MODULE__{} = host, event_name, event_opts \\ []) do + submit_event(host, Event.external(event_name, event_opts)) + end + + @doc """ + Builds the `done.invoke.` event in the exact shape + `deps/statifier/lib/statifier/session.ex`'s own construction site builds + it (the shape `docs/extending.md` documents for a process-less host to + match) and submits it. `run_id` stands in for `session_id` - this host has + no session, only a run. + """ + @spec finish_invocation(t(), String.t(), term(), keyword()) :: t() + def finish_invocation(%__MODULE__{} = host, invoke_id, donedata, _opts \\ []) do + event = + Event.external("done.invoke." <> invoke_id, + data: donedata, + invokeid: invoke_id, + origin: SystemVariables.scxml_location(host.run_id), + origintype: SystemVariables.scxml_event_processor() + ) + + submit_event(host, event) + end + + @doc """ + Advances the mock clock by `delta_ms`. For each ordinal `Runtime.due/2` + reports, loads that timer's row from the ledger, submits its event, and + drops the durable row - after the step returns, so a step that discards + (a terminal run) still drops the row without having driven the chart at + all (`deps/statifier/docs/durable-timers.md:286`'s liveness rule; `Runs` + itself checks run status before any position decode). + """ + @spec tick(t(), non_neg_integer()) :: t() + def tick(%__MODULE__{} = host, delta_ms) do + host.runtime + |> Runtime.due(delta_ms) + |> Enum.reduce(host, fn ordinal, host -> fire_timer(host, ordinal) end) + end + + @spec fire_timer(t(), Runtime.ordinal()) :: t() + defp fire_timer(host, ordinal) do + host = + case timer_row(host, ordinal) do + nil -> host + row -> submit_event(host, Event.external(row.event, sendid: row.send_id, data: row.data)) + end + + :ok = Ledger.drop_timer(host.ledger, host.run_id, ordinal) + host + end + + @spec timer_row(t(), Runtime.ordinal()) :: Ledger.timer_row() | nil + defp timer_row(host, ordinal) do + host.ledger + |> Ledger.open_timers(host.run_id) + |> Enum.find(&(&1.ordinal == ordinal)) + end + + @spec submit_event(t(), Event.t()) :: t() + defp submit_event(host, event) do + host = %{host | tape: host.tape ++ [event]} + + case Runs.step(host.store, host.run_id, host.machine, event, + executor: executor(host), + invoke_types: host.invoke_types, + routes: Routes.new() + ) do + {:ok, run, _machine_state} -> %{host | run: run} + {:discarded, run} -> %{host | run: run} + end + end + + @doc """ + Re-establishes liveness after a cold `boot/4`, per st-ADR-0060 decision + 7's "resume restores position, not liveness". Performs no step and emits + no event - it only repopulates `runtime`/`ledger` state the position + never carried: + + 1. **Timers.** Every `Ledger.open_timers/2` row for this run is + `Runtime.arm/3`ed again, unconditionally - the engine is not + consulted, because nothing about a pending delayed send survives the + position (the plan's Key Discoveries). + 2. **Invocations.** For each `{_key, invoke_id}` the durable position's + `active_invocations` still names *and* the ledger still shows `:open`, + `handler.start/2` is re-run and its instructions performed - the + engine is the liveness authority (which ids are still active), the + ledger is the payload source (the `type`/`params` that id started + with); an id either side has dropped is left alone. Re-running + `Ledger.record_invocation/3` here is idempotent on `invoke_id`, so + re-establishing a still-open invocation records no second + side-effect entry. + 3. Returns the rebuilt `%Host{}`. `invoke_handlers:`/`invoke_types:` must + already be set on `host` (a caller re-supplies them after `boot/4`, + same as any other per-deployment declaration). + """ + @spec recover(t()) :: t() + def recover(%__MODULE__{} = host) do + host.ledger + |> Ledger.open_timers(host.run_id) + |> Enum.each(fn row -> Runtime.arm(host.runtime, row.ordinal, row.due_at_ms) end) + + {:ok, machine_state} = position(host) + + open_invoke_ids = + host.ledger |> Ledger.open_invocations(host.run_id) |> MapSet.new(& &1.invoke_id) + + machine_state.active_invocations + |> Map.values() + |> Enum.filter(&MapSet.member?(open_invoke_ids, &1)) + |> Enum.each(&reestablish_invocation(host, &1)) + + host + end + + @spec reestablish_invocation(t(), String.t()) :: :ok + defp reestablish_invocation(host, invoke_id) do + type = invocation_type(host, invoke_id) + params = invocation_params(host, invoke_id) + + case Map.fetch(host.invoke_handlers, type) do + {:ok, handler} -> + # state_index/invoke_index/macrostep/microstep/round are the + # interpreter's own bookkeeping for a freshly-planned `` - + # meaningless for one re-established outside a step, but + # `%Invoke{}`'s `@enforce_keys` still requires a value for each. + invoke = %Invoke{ + invoke_id: invoke_id, + type: type, + params: params, + state_index: 0, + invoke_index: 0, + macrostep: 0, + microstep: 0, + round: 0 + } + + {:ok, instructions} = handler.start(invoke, handler_ctx(host)) + Enum.each(instructions, &perform_instruction(host, type, &1)) + :ok + + :error -> + {:error, {:no_handler, type}} + end + end + + @spec invocation_params(t(), String.t()) :: term() + defp invocation_params(host, invoke_id) do + host.ledger + |> Ledger.open_invocations(host.run_id) + |> Enum.find_value(fn row -> row.invoke_id == invoke_id and row.params end) + end + + @doc """ + Simulates the node coming back up knowing only `run_id`: `Runtime.stop/1` + the old (dead-by-assumption) runtime, discard the old `%Host{}`, start a + fresh `Runtime`, `boot/4`, `recover/1`. Returns a new struct rather than + mutating one, so a test that keeps using the pre-restart binding is a + test that fails. + + `invoke_handlers:`/`invoke_types:` are carried forward from `host` onto + the rebuilt struct, same per-deployment declaration `boot/4`'s own + moduledoc describes a caller re-supplying. + """ + @spec restart(t()) :: {:ok, t()} | {:error, term()} + def restart(%__MODULE__{} = host) do + :ok = Runtime.stop(host.runtime) + {:ok, runtime} = Runtime.start_link([]) + + with {:ok, rebooted} <- boot(host.store, host.ledger, runtime, host.run_id) do + rebooted = %{ + rebooted + | invoke_handlers: host.invoke_handlers, + invoke_types: host.invoke_types + } + + {:ok, recover(rebooted)} + end + end + + @doc "The last-observed run record - `nil` before `start_run/6`/`boot/4`." + @spec run(t()) :: Run.t() | nil + def run(%__MODULE__{run: run}), do: run + + @doc "The current durable position, reloaded from `store` (never cached on the struct)." + @spec position(t()) :: {:ok, MachineState.t()} | {:error, term()} + def position(%__MODULE__{} = host), + do: Storage.load_run_position(host.store, host.run_id, host.machine) + + @doc "The active leaf states as sorted string ids - `runs_test.exs`'s `active_ids/1` shape, off the durable position." + @spec config(t()) :: [String.t()] + def config(%__MODULE__{} = host) do + {:ok, machine_state} = position(host) + + machine_state + |> MachineState.active_leaf_states() + |> Enum.map(&Machine.id(machine_state.machine, &1)) + |> Enum.sort() + end + + @doc "Every event submitted so far, oldest first - the input tape a replay drives." + @spec tape(t()) :: [Event.t()] + def tape(%__MODULE__{tape: tape}), do: tape +end diff --git a/test/support/demo/ledger.ex b/test/support/demo/ledger.ex new file mode 100644 index 0000000..3fb4133 --- /dev/null +++ b/test/support/demo/ledger.ex @@ -0,0 +1,211 @@ +defmodule StatifierPersistence.Demo.Ledger do + @moduledoc """ + The demo host's own durable store: an `Agent` standing in for the + embedder's own database tables (`docs/plans/260822-sp-4an.4-restart-demo-host.md`, + Phase 1). Unlike `StatifierPersistence.Storage`, this survives a + simulated restart *by construction* - it is never stopped alongside + `StatifierPersistence.Demo.Runtime`, only the volatile layer is. + + Started per test with `start_supervised!/1` and carried on the + `StatifierPersistence.Demo.Host` struct by pid, so demo tests stay + `async: true` - there is no global name to collide on. + + Four tables: + + - `timers` - pending durable sends, keyed by `{run_id, ordinal}` + (st-ADR-0059's dedup key: the counter triple and the content position + alone cannot tell two `` iterations of the same `` apart, + only `ordinal` can). + - `invocations` - open/closed async invocations, keyed by + `{run_id, invoke_id}`. + - `side_effects` - the append-only idempotency ledger: `arm_timer/3` and + `record_invocation/3` each append here only the first time their key is + seen, which is what lets a test assert "no duplicate side effects" + independently of the executor call log. + - `executor_calls` - every `{effect, context}` pair the demo host's + executor saw, in call order - the log the bead's success criteria ask + to assert exactly. + """ + + use Agent + + @type run_id :: String.t() + @type ordinal :: pos_integer() + @type timer_key :: {run_id(), ordinal()} + @type invocation_key :: {run_id(), String.t()} + @type timer_row :: %{ + send_id: String.t() | nil, + event: String.t(), + data: term(), + due_at_ms: non_neg_integer(), + ordinal: ordinal() + } + @type invocation_row :: %{ + invoke_id: String.t(), + type: String.t() | nil, + params: term(), + status: :open | :done + } + @type side_effect_key :: {atom(), term()} + + @type t :: pid() + + @type state :: %{ + timers: %{timer_key() => timer_row()}, + invocations: %{invocation_key() => invocation_row()}, + side_effects: [side_effect_key()], + executor_calls: [{Statifier.Effect.t(), StatifierPersistence.Executor.context()}] + } + + @doc "Starts a fresh, empty ledger. `opts` is unused; `start_supervised!/1`'s own signature." + @spec start_link(term()) :: Agent.on_start() + def start_link(_opts \\ []) do + Agent.start_link(fn -> + %{timers: %{}, invocations: %{}, side_effects: [], executor_calls: []} + end) + end + + @doc """ + Records a pending durable timer, idempotent on `{run_id, timer.ordinal}`. + A second call under the same key (a re-driven step, or `recover/1` + re-arming) leaves the stored row and the `side_effects` log untouched. + """ + @spec arm_timer(t(), run_id(), timer_row()) :: :ok + def arm_timer(ledger, run_id, %{ordinal: ordinal} = row) do + key = {run_id, ordinal} + + Agent.update(ledger, fn state -> + if Map.has_key?(state.timers, key) do + state + else + state + |> put_in([:timers, key], row) + |> append_side_effect({:arm_timer, key}) + end + end) + end + + @doc """ + Drops every timer row under `run_id` whose `send_id` matches, returning + the removed rows' ordinals. `` names a send id, not an + ordinal, and more than one armed row can share a send id (two iterations + of the same authored ``), so this can remove more than one row. + """ + @spec cancel_timer(t(), run_id(), String.t() | nil) :: [ordinal()] + def cancel_timer(ledger, run_id, send_id) do + Agent.get_and_update(ledger, fn state -> + {removed, kept} = + Enum.split_with(state.timers, fn {{r, _ordinal}, row} -> + r == run_id and row.send_id == send_id + end) + + ordinals = removed |> Enum.map(fn {{_r, ordinal}, _row} -> ordinal end) |> Enum.sort() + {ordinals, %{state | timers: Map.new(kept)}} + end) + end + + @doc "Unconditionally drops the timer row for `{run_id, ordinal}`, after it fired or was consumed." + @spec drop_timer(t(), run_id(), ordinal()) :: :ok + def drop_timer(ledger, run_id, ordinal) do + Agent.update(ledger, fn state -> + update_in(state.timers, &Map.delete(&1, {run_id, ordinal})) + end) + end + + @doc "Every open timer row for `run_id`, oldest ordinal first." + @spec open_timers(t(), run_id()) :: [timer_row()] + def open_timers(ledger, run_id) do + Agent.get(ledger, fn state -> + state.timers + |> Enum.filter(fn {{r, _ordinal}, _row} -> r == run_id end) + |> Enum.sort_by(fn {{_r, ordinal}, _row} -> ordinal end) + |> Enum.map(fn {_key, row} -> row end) + end) + end + + @doc """ + Records an invocation as `:open`, idempotent on `{run_id, invoke_id}`. A + second call under the same key (`recover/1` re-running `handler.start/2`) + leaves the stored row and the `side_effects` log untouched - the ledger, + not the handler, is what makes re-establishment idempotent. + """ + @spec record_invocation(t(), run_id(), %{invoke_id: String.t(), type: term(), params: term()}) :: + :ok + def record_invocation(ledger, run_id, %{invoke_id: invoke_id} = attrs) do + key = {run_id, invoke_id} + + Agent.update(ledger, fn state -> + if Map.has_key?(state.invocations, key) do + state + else + row = Map.put(attrs, :status, :open) + + state + |> put_in([:invocations, key], row) + |> append_side_effect({:record_invocation, key}) + end + end) + end + + @doc "Marks the invocation under `{run_id, invoke_id}` `:done`. A no-op when no such row exists." + @spec close_invocation(t(), run_id(), String.t()) :: :ok + def close_invocation(ledger, run_id, invoke_id) do + key = {run_id, invoke_id} + + Agent.update(ledger, fn state -> + update_in(state.invocations[key], fn + nil -> nil + row -> %{row | status: :done} + end) + end) + end + + @doc "Every `:open` invocation row for `run_id`, sorted by `invoke_id`." + @spec open_invocations(t(), run_id()) :: [invocation_row()] + def open_invocations(ledger, run_id) do + Agent.get(ledger, fn state -> + state.invocations + |> Enum.filter(fn {{r, _id}, row} -> r == run_id and row.status == :open end) + |> Enum.sort_by(fn {{_r, invoke_id}, _row} -> invoke_id end) + |> Enum.map(fn {_key, row} -> row end) + end) + end + + @doc "Appends `{effect, context}` to the executor call log, in call order." + @spec record_call(t(), Statifier.Effect.t(), StatifierPersistence.Executor.context()) :: :ok + def record_call(ledger, effect, context) do + Agent.update(ledger, fn state -> + %{state | executor_calls: [{effect, context} | state.executor_calls]} + end) + end + + @doc "Every recorded `{effect, context}` call, oldest first." + @spec calls(t()) :: [{Statifier.Effect.t(), StatifierPersistence.Executor.context()}] + def calls(ledger) do + Agent.get(ledger, fn state -> Enum.reverse(state.executor_calls) end) + end + + @doc """ + Appends `key` to the idempotency ledger, but only the first time it is + seen - the mechanism `arm_timer/3` and `record_invocation/3` share. + """ + @spec record_side_effect(t(), side_effect_key()) :: :ok + def record_side_effect(ledger, key) do + Agent.update(ledger, &append_side_effect(&1, key)) + end + + @doc "Every recorded side-effect key, oldest first, one entry per genuinely new key." + @spec side_effects(t()) :: [side_effect_key()] + def side_effects(ledger) do + Agent.get(ledger, fn state -> Enum.reverse(state.side_effects) end) + end + + @spec append_side_effect(state(), side_effect_key()) :: state() + defp append_side_effect(state, key) do + if key in state.side_effects do + state + else + %{state | side_effects: [key | state.side_effects]} + end + end +end diff --git a/test/support/demo/runtime.ex b/test/support/demo/runtime.ex new file mode 100644 index 0000000..73e6146 --- /dev/null +++ b/test/support/demo/runtime.ex @@ -0,0 +1,173 @@ +defmodule StatifierPersistence.Demo.Runtime do + @moduledoc """ + The demo host's volatile layer (Phase 1 of + `docs/plans/260822-sp-4an.4-restart-demo-host.md`): a `Supervisor` over + two `Agent` children plus a monotonic mock clock. Stopping it is the + simulated node death - everything it owns, including the invoke worker + processes it spawned, dies with it, and nothing here is durable. + + - `Timers` - the mock clock (`now_ms`) and the set of armed timer keys, + each mapped to the `due_at_ms` it fires at. `StatifierPersistence.Demo.Ledger` + holds the timer's payload (event/data/send_id); this agent holds only + scheduling state, mirroring the split the plan draws between "what + survives" (the ledger) and "what does not" (this). + - `Workers` - one real spawned `Agent` per live invocation, keyed by + `invoke_id`. Each worker is a genuine process so "the pid died with the + node" and "a different pid serves it now" (post-restart) are both + directly observable rather than asserted by convention. + + No wall-clock timer anywhere: the mock clock only advances when a test + calls `due/2`, which is what keeps the demo fast and deterministic. + """ + + alias StatifierPersistence.Demo.Runtime.{Timers, Workers} + + @type ordinal :: pos_integer() + + @type t :: %{sup: pid(), timers: pid(), workers: pid()} + + @doc "Starts a fresh runtime: a supervisor over one `Timers` agent and one `Workers` agent." + @spec start_link(term()) :: {:ok, t()} | {:error, term()} + def start_link(_opts \\ []) do + with {:ok, sup} <- Supervisor.start_link([Timers, Workers], strategy: :one_for_one) do + children = Supervisor.which_children(sup) + + {:ok, + %{sup: sup, timers: child_pid(children, Timers), workers: child_pid(children, Workers)}} + end + end + + @spec child_pid([tuple()], module()) :: pid() + defp child_pid(children, module) do + Enum.find_value(children, fn {id, pid, _type, _mods} -> id == module and pid end) + end + + @doc """ + Simulates the node dying: every worker process this runtime spawned is + stopped first (they are not supervisor children - only `Timers` and + `Workers` are - so stopping the supervisor alone would leave them + running), then the supervisor itself is stopped. + """ + @spec stop(t()) :: :ok + def stop(%{sup: sup, workers: workers}) do + workers + |> Workers.pids() + |> Enum.each(fn pid -> if Process.alive?(pid), do: Agent.stop(pid) end) + + Supervisor.stop(sup) + end + + @doc "Arms `ordinal` to fire at `due_at_ms` (absolute mock-clock milliseconds)." + @spec arm(t(), ordinal(), non_neg_integer()) :: :ok + def arm(%{timers: timers}, ordinal, due_at_ms), do: Timers.arm(timers, ordinal, due_at_ms) + + @doc "Disarms `ordinal`, whether or not it was armed." + @spec disarm(t(), ordinal()) :: :ok + def disarm(%{timers: timers}, ordinal), do: Timers.disarm(timers, ordinal) + + @doc "The armed ordinals right now, with no clock advance." + @spec armed(t()) :: MapSet.t(ordinal()) + def armed(%{timers: timers}), do: Timers.armed(timers) + + @doc """ + Advances the mock clock by `delta_ms` and returns every armed ordinal + whose `due_at_ms` is now at or before the new clock time, disarming each + one it returns (a fired timer is no longer armed). + """ + @spec due(t(), non_neg_integer()) :: [ordinal()] + def due(%{timers: timers}, delta_ms), do: Timers.advance(timers, delta_ms) + + @doc "The mock clock's current time, in milliseconds." + @spec now_ms(t()) :: non_neg_integer() + def now_ms(%{timers: timers}), do: Timers.now_ms(timers) + + @doc "Spawns a fresh worker process for `invoke_id`, replacing any prior one under the same id." + @spec start_worker(t(), String.t(), term()) :: pid() + def start_worker(%{workers: workers}, invoke_id, payload), + do: Workers.start(workers, invoke_id, payload) + + @doc "The live worker pid for `invoke_id`, or `nil` if none is running." + @spec worker(t(), String.t()) :: pid() | nil + def worker(%{workers: workers}, invoke_id), do: Workers.get(workers, invoke_id) + + @doc "Stops the worker for `invoke_id`, if one is running." + @spec stop_worker(t(), String.t()) :: :ok + def stop_worker(%{workers: workers}, invoke_id), do: Workers.stop(workers, invoke_id) + + defmodule Timers do + @moduledoc false + use Agent + + @spec start_link(term()) :: Agent.on_start() + def start_link(_opts \\ []) do + Agent.start_link(fn -> %{now_ms: 0, armed: %{}} end) + end + + @spec arm(pid(), pos_integer(), non_neg_integer()) :: :ok + def arm(timers, ordinal, due_at_ms) do + Agent.update(timers, fn state -> put_in(state.armed[ordinal], due_at_ms) end) + end + + @spec disarm(pid(), pos_integer()) :: :ok + def disarm(timers, ordinal) do + Agent.update(timers, fn state -> update_in(state.armed, &Map.delete(&1, ordinal)) end) + end + + @spec armed(pid()) :: MapSet.t(pos_integer()) + def armed(timers) do + Agent.get(timers, fn state -> state.armed |> Map.keys() |> MapSet.new() end) + end + + @spec advance(pid(), non_neg_integer()) :: [pos_integer()] + def advance(timers, delta_ms) do + Agent.get_and_update(timers, fn state -> + now_ms = state.now_ms + delta_ms + + {due, armed} = + Enum.split_with(state.armed, fn {_ordinal, due_at_ms} -> due_at_ms <= now_ms end) + + ordinals = due |> Enum.map(fn {ordinal, _due_at_ms} -> ordinal end) |> Enum.sort() + {ordinals, %{now_ms: now_ms, armed: Map.new(armed)}} + end) + end + + @spec now_ms(pid()) :: non_neg_integer() + def now_ms(timers), do: Agent.get(timers, & &1.now_ms) + end + + defmodule Workers do + @moduledoc false + use Agent + + @spec start_link(term()) :: Agent.on_start() + def start_link(_opts \\ []) do + Agent.start_link(fn -> %{} end) + end + + @spec start(pid(), String.t(), term()) :: pid() + def start(workers, invoke_id, payload) do + {:ok, worker_pid} = Agent.start_link(fn -> payload end) + Agent.update(workers, &Map.put(&1, invoke_id, worker_pid)) + worker_pid + end + + @spec get(pid(), String.t()) :: pid() | nil + def get(workers, invoke_id), do: Agent.get(workers, &Map.get(&1, invoke_id)) + + @spec stop(pid(), String.t()) :: :ok + def stop(workers, invoke_id) do + Agent.get_and_update(workers, fn state -> + {worker_pid, state} = Map.pop(state, invoke_id) + stop_worker_pid(worker_pid) + {:ok, state} + end) + end + + @spec stop_worker_pid(pid() | nil) :: :ok + defp stop_worker_pid(nil), do: :ok + defp stop_worker_pid(pid), do: if(Process.alive?(pid), do: Agent.stop(pid), else: :ok) + + @spec pids(pid()) :: [pid()] + def pids(workers), do: Agent.get(workers, &Map.values(&1)) + end +end diff --git a/test/support/demo/scenario.ex b/test/support/demo/scenario.ex new file mode 100644 index 0000000..b1f6f92 --- /dev/null +++ b/test/support/demo/scenario.ex @@ -0,0 +1,338 @@ +defmodule StatifierPersistence.Demo.Scenario do + @moduledoc """ + The demo chart and the scenario body that drives it + (`docs/plans/260822-sp-4an.4-restart-demo-host.md`), parameterized on a + `{adapter, opts}` storage pair so the same body runs unchanged against + `StatifierPersistence.Storage.InMemory` (Phase 1-3) and + `StatifierPersistence.Storage.Ecto` (Phase 4). + + The chart, probe-verified end to end before this plan was written (see + the plan's Key Discoveries): `intake` submits into `enriching`, which + arms a 900s `sla-timer` and starts an async `myapp:enrich` invocation in + parallel; `done.invoke.enrich` moves to `cooling`, which arms a 3600s + `reminder-timer`; the still-armed `sla-timer` (never explicitly + cancelled - only `` exit-cancels, never a plain ``) fires + `sla.breach` from inside `cooling`, cancelling `reminder-timer` on the + way to `settling`; `ack` reaches the top-level final `settled`. `escalated` + is the negative target: reaching it means the demo lost the race it + exists to control. + """ + + alias Statifier.Event + alias Statifier.Invoke.Types, as: InvokeTypes + alias Statifier.Send.Routes + alias StatifierPersistence.Demo.{EnrichHandler, Host, Ledger, Runtime} + alias StatifierPersistence.{Runs, Storage} + + @chart_source """ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + """ + + @doc "The chart's raw SCXML source." + @spec source() :: String.t() + def source, do: @chart_source + + @doc "The chart, compiled once. Raises on a compile failure - the source is probe-verified." + @spec machine!() :: Statifier.Machine.t() + def machine! do + {:ok, machine} = Statifier.compile(@chart_source) + machine + end + + @doc "The registered-types snapshot every step in this scenario stamps." + @spec invoke_types() :: InvokeTypes.t() + def invoke_types, do: InvokeTypes.new(types: ["myapp:enrich"]) + + @doc "The handler palette every step in this scenario stamps." + @spec handlers() :: %{String.t() => module()} + def handlers, do: %{"myapp:enrich" => EnrichHandler} + + @doc """ + Drives the chart straight through, with no restart: `submit` -> + `finish_invocation` -> a `tick` long enough for the still-armed + `sla-timer` to fire -> `ack`. Returns a map with the final `Host.t()`, + the ledger and runtime it ran against (for a caller that wants to inspect + them directly), and the leaf-id configuration observed after each of the + three non-terminal steps, in order. + + `{adapter, opts}` is handed straight to `Storage.new/2`, unchanged - this + function names no storage module itself, which is what lets Phase 4 run + it again against `Storage.Ecto` with no edit here. + """ + @spec straight_through({module(), keyword()}) :: %{ + host: Host.t(), + ledger: Ledger.t(), + runtime: Runtime.t(), + configs: [[String.t()]] + } + def straight_through({adapter, opts}) do + run_id = unique_run_id() + {:ok, store} = Storage.new(adapter, opts) + {:ok, ledger} = Ledger.start_link([]) + {:ok, runtime} = Runtime.start_link([]) + + {:ok, host} = + Host.start_run(store, ledger, runtime, run_id, @chart_source, + invoke_handlers: handlers(), + invoke_types: invoke_types() + ) + + host = Host.submit(host, "submit") + after_submit = Host.config(host) + + host = Host.finish_invocation(host, "enrich", %{"score" => 7}) + after_finish = Host.config(host) + + # Long enough for the still-armed 900s sla-timer to fire; short of the + # 3600s reminder-timer cooling armed, which must never fire. + host = Host.tick(host, :timer.minutes(20)) + after_tick = Host.config(host) + + host = Host.submit(host, "ack") + + %{ + host: host, + ledger: ledger, + runtime: runtime, + configs: [after_submit, after_finish, after_tick] + } + end + + @doc """ + Drives the chart to the kill point (`enriching`, `sla-timer` pending, + `enrich` in flight), stops the volatile runtime to simulate the node + dying, cold-boots a fresh `Host.t()` from `run_id` alone, `recover/1`s + it, then drives the rest of the chart to `:completed` exactly as + `straight_through/1` does after its own kill point. + + Returns a map carrying every value the restart test needs to assert at + each stage: the host and its pre-restart worker pid at the kill point, + the rebuilt host after `boot/4` and again after `recover/1`, the final + host, the ledger (shared across the restart - it is the durable layer), + and the leaf-id configuration observed after each non-terminal step. + + `{adapter, opts}` is the same pass-through `straight_through/1` takes. + """ + @spec across_restart({module(), keyword()}) :: %{ + host_at_kill: Host.t(), + config_at_kill: [String.t()], + open_timers_at_kill: [Ledger.timer_row()], + open_invocations_at_kill: [Ledger.invocation_row()], + active_invocations_at_kill: non_neg_integer(), + pid_before: pid(), + alive_before_stop: boolean(), + alive_after_stop: boolean(), + armed_before_stop: MapSet.t(pos_integer()), + host_after_boot: Host.t(), + armed_after_boot: MapSet.t(pos_integer()), + host_after_recover: Host.t(), + pid_after_recover: pid() | nil, + alive_after_recover: boolean(), + open_timers_after_recover: [Ledger.timer_row()], + host: Host.t(), + ledger: Ledger.t(), + run_id: String.t(), + configs: [[String.t()]] + } + def across_restart({adapter, opts}) do + run_id = unique_run_id() + {:ok, store} = Storage.new(adapter, opts) + {:ok, ledger} = Ledger.start_link([]) + {:ok, runtime} = Runtime.start_link([]) + + {:ok, host} = + Host.start_run(store, ledger, runtime, run_id, @chart_source, + invoke_handlers: handlers(), + invoke_types: invoke_types() + ) + + host_at_kill = Host.submit(host, "submit") + + # `Host.position/1`/`Host.config/1` and the ledger's `open_*` readers + # always reflect the *current* store/ledger state, never a cached + # snapshot (`Host.position/1`'s own moduledoc) - so every kill-point + # observation the test wants to make must be captured here, before the + # scenario drives any further step, not read back off `host_at_kill` + # once this function has returned. + config_at_kill = Host.config(host_at_kill) + open_timers_at_kill = Ledger.open_timers(ledger, run_id) + open_invocations_at_kill = Ledger.open_invocations(ledger, run_id) + {:ok, machine_state_at_kill} = Host.position(host_at_kill) + active_invocations_at_kill = map_size(machine_state_at_kill.active_invocations) + + pid_before = Runtime.worker(host_at_kill.runtime, "enrich") + alive_before_stop = Process.alive?(pid_before) + armed_before_stop = Runtime.armed(host_at_kill.runtime) + + # The node dies: every volatile pid this runtime owns, timers and + # workers alike, goes with it. `Process.alive?(pid_before)` after this + # point is captured here too - the scenario runs to completion inside + # one call, so a caller that checked it against the returned pid only + # after `across_restart/1` returns would always see the post-restart + # world, same reason `config_at_kill` above is captured rather than + # re-derived from `host_at_kill`. + :ok = Runtime.stop(host_at_kill.runtime) + alive_after_stop = Process.alive?(pid_before) + + {:ok, new_runtime} = Runtime.start_link([]) + {:ok, host_after_boot} = Host.boot(store, ledger, new_runtime, run_id) + + # The input tape is the recorder's own log, carried across the restart + # by this scenario (the recorder never died - only the node did), not + # restored by `boot/4`: recording inputs durably is a host concern + # outside the position's scope, same as the handler palette re-supplied + # below. + host_after_boot = %{ + host_after_boot + | invoke_handlers: handlers(), + invoke_types: invoke_types(), + tape: Host.tape(host_at_kill) + } + + # Captured before `recover/1` runs, on the same fresh runtime pid + # `host_after_recover` below goes on to arm - proving `boot/4` alone + # restores nothing volatile. + armed_after_boot = Runtime.armed(host_after_boot.runtime) + + host_after_recover = Host.recover(host_after_boot) + pid_after_recover = Runtime.worker(host_after_recover.runtime, "enrich") + # Captured now, before `finish_invocation/4` below cancel-stops this + # very worker as part of driving the tail - same reason every other + # "as of this moment" field here is captured eagerly. + alive_after_recover = pid_after_recover != nil and Process.alive?(pid_after_recover) + open_timers_after_recover = Ledger.open_timers(ledger, run_id) + + host = Host.finish_invocation(host_after_recover, "enrich", %{"score" => 7}) + after_finish = Host.config(host) + + # Long enough for the still-armed 900s sla-timer to fire; short of the + # 3600s reminder-timer cooling armed after the restart, which must + # never fire. + host = Host.tick(host, :timer.minutes(20)) + after_tick = Host.config(host) + + host = Host.submit(host, "ack") + + %{ + host_at_kill: host_at_kill, + config_at_kill: config_at_kill, + open_timers_at_kill: open_timers_at_kill, + open_invocations_at_kill: open_invocations_at_kill, + active_invocations_at_kill: active_invocations_at_kill, + pid_before: pid_before, + alive_before_stop: alive_before_stop, + alive_after_stop: alive_after_stop, + armed_before_stop: armed_before_stop, + host_after_boot: host_after_boot, + armed_after_boot: armed_after_boot, + host_after_recover: host_after_recover, + pid_after_recover: pid_after_recover, + alive_after_recover: alive_after_recover, + open_timers_after_recover: open_timers_after_recover, + host: host, + ledger: ledger, + run_id: run_id, + configs: [after_finish, after_tick] + } + end + + @doc """ + Replays `tape` (`Host.tape/1`'s events, oldest first) against a fresh + `{store, ledger}` pair and a fresh `run_id` - with **no `Demo.Runtime` + at all**: no timers, no workers, nothing but the recorded inputs driven + straight through `Runs.create/4` then one `Runs.step/5` per event. The + executor here only records onto `ledger`; it dispatches nothing, because + a replay proves the loop is deterministic, not that the demo host's + side-effecting dispatch is (Phase 1-2 already prove that). + + Returns the leaf-id configuration observed after `create/4` and after + every step, in order (so `length(configs) == length(tape) + 1`), plus + the exact effects the executor saw, in call order. + + `opts` accepts `initialize:` (passed through to `Runs.create/4`'s own + `initialize:` option). The one non-deterministic input a create takes is + the session id `MachineState.new/2` otherwise generates fresh - it is + stamped into the `:datamodel_init` effect's `_sessionid` and + `_ioprocessors` system variables - so a replay that must reproduce the + original effects byte for byte re-supplies the recorded one via + `initialize: [session_id: ...]`, the same way it re-supplies the + recorded events. + """ + @spec replay([Event.t()], {Storage.t(), Ledger.t()}, Runs.run_id(), keyword()) :: %{ + configs: [[String.t()]], + effects: [Statifier.Effect.t()] + } + def replay(tape, {%Storage{} = store, ledger}, run_id, opts \\ []) do + machine = machine!() + + executor = fn effect, context -> + :ok = Ledger.record_call(ledger, effect, context) + :ok + end + + {:ok, _run, machine_state} = + Runs.create(store, run_id, machine, + executor: executor, + invoke_types: invoke_types(), + initialize: Keyword.get(opts, :initialize, []) + ) + + {_final_state, configs} = + Enum.reduce(tape, {machine_state, [leaf_config(machine_state)]}, fn event, + {_state, configs} -> + {:ok, _run, machine_state} = + Runs.step(store, run_id, machine, event, + executor: executor, + invoke_types: invoke_types(), + routes: Routes.new() + ) + + {machine_state, configs ++ [leaf_config(machine_state)]} + end) + + effects = ledger |> Ledger.calls() |> Enum.map(fn {effect, _context} -> effect end) + + %{configs: configs, effects: effects} + end + + @spec leaf_config(Statifier.MachineState.t()) :: [String.t()] + defp leaf_config(machine_state) do + machine_state + |> Statifier.MachineState.active_leaf_states() + |> Enum.map(&Statifier.Machine.id(machine_state.machine, &1)) + |> Enum.sort() + end + + @spec unique_run_id() :: String.t() + defp unique_run_id, do: "restart-demo-" <> Integer.to_string(System.unique_integer([:positive])) +end