diff --git a/.gitignore b/.gitignore index 74365c534..ba06dc741 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,7 @@ /target /target-shared /.claude/runtime/ +.claude/**/runtime/ .cargo/config.toml __pycache__/ *.pyc diff --git a/docs/concepts/typed-outcome-ledger-shared-connection.md b/docs/concepts/typed-outcome-ledger-shared-connection.md new file mode 100644 index 000000000..2026437c5 --- /dev/null +++ b/docs/concepts/typed-outcome-ledger-shared-connection.md @@ -0,0 +1,144 @@ +--- +title: "Typed-Outcome Ledger Shared Connection (one connection per file, not per handler)" +description: > + Why the typed-outcome ledger uses a single, process-wide serialized SQLite + connection per ledger file instead of one connection per CapabilityHandler. + Explains the concurrent-writer burst that produced systemic + "typed outcome persistence failed: database is locked" errors (issue #4483), + why a per-handler Mutex could not prevent it, how a path-keyed shared + connection removes the race by construction, and why serializing the single + ledger writer is an acceptable trade. +last_updated: 2026-07-23 +review_schedule: as-needed +owner: simard +doc_type: explanation +status: design — not yet implemented +related: + - ../reference/typed-outcome-ledger-connection-registry-api.md + - ../howto/diagnose-typed-outcome-database-is-locked.md + - ../reference/ooda-capability-api.md +--- + +# Typed-Outcome Ledger Shared Connection + +> **Spec-first (retcon) document.** This explains the *target* design for issue +> **#4483**. The fix is **not yet landed** — today each `CapabilityHandler::open` +> still builds its own `Mutex`. The documentation and implementation +> land in the **same pull request**; flip `status:` to `implemented` when that PR +> merges. + +## The invariant + +There is exactly **one** typed-outcome ledger file per state root: + +``` +/typed-ooda/outcomes.sqlite3 +``` + +It is the durable audit trail of terminal OODA outcomes and the effect outbox. +The invariant this design protects: **an outcome the system reported as recorded +is durably present in that file, and a write never silently fails.** + +## What went wrong (issue #4483) + +Within a single goal session, several `CapabilityHandler` instances open that +*same file*: + +- the **startup outbox worker** that drains pending effects on session start + (`OutboxWorker::drain_pending` in + `src/ooda_actions/advance_goal/typed_goal_session.rs`), and +- the **route executor** that records terminal outcomes and enqueues effects. + +Concurrently running goal sessions add more openers of the same file. + +Before the fix, each `open` created an **independent** `Connection` and wrapped it +in a **per-handler** `Mutex`: + +```rust +// pre-fix +connection: Mutex, // one per handler +// ... +connection: Mutex::new(connection), // independent connection per open() +``` + +A `Mutex` serializes access **inside one handler**. It does nothing +between handlers: two handlers are two independent connections to the same file. +When both attempt a write transaction at once, SQLite's file-level locking lets +only one writer proceed and the other gets `SQLITE_BUSY` — surfaced as +`"database is locked"`. + +`open` sets a `busy_timeout`, which retries for a few seconds. But when many +writers converge in a **burst** — several goals reaching a terminal at the same +moment, each with its startup drain firing — the timeout is exhausted and the +busy error escapes. It is mapped through `persistence(..)` to +`CapabilityErrorCode::PersistenceFailed`, which **aborts the record**. The +terminal outcome that the loop believed it had persisted is dropped from the +audit trail. That is the "systemic typed-outcome PersistenceFailed" of #4483. + +Concretely, the failure signature is several goals (say `` … ``, +six distinct goals reaching a terminal in the same window) each logging +`typed outcome persistence failed: ... database is locked` within the same +few seconds — a contention *burst*, not a steady leak. + +## Why a bigger `busy_timeout` is not the fix + +Raising the timeout only widens the window before the error surfaces; under a +true burst of independent connections the collision is structural, and a longer +timeout trades a lost outcome for a stalled goal session. The problem is *having +multiple independent writers to one file at all*, not how long each one waits. + +## The fix: one connection per file, shared + +The design collapses "one connection per handler" into **one connection per +ledger file, per process**, held behind a process-global, path-keyed registry: + +``` +OnceLock>>>> +``` + +Every `CapabilityHandler` opened against a given file **clones the same** +`Arc>`. The `Mutex` that used to serialize one handler now +serializes **every** writer to that file across the whole process. There is only +ever one connection issuing writes, so the cross-connection `SQLITE_BUSY` race +cannot occur — it is removed by construction, not merely retried away. + +Opening a new file also applies durable pragmas once — WAL journaling, +`busy_timeout = 5000`, and `foreign_keys = ON` — and a bounded +`with_busy_retry` wraps writes as defense-in-depth against *external* processes +touching the file. See the +[connection registry API reference](../reference/typed-outcome-ledger-connection-registry-api.md) +for the exact contract. + +## Why serializing the writer is acceptable + +The typed-outcome ledger is a **low-frequency, small-write** audit trail: +terminal outcomes and outbox effect rows, written at OODA cycle boundaries — not +a hot data-plane. Serializing its single writer costs nothing meaningful in +throughput, and it buys a hard correctness guarantee: no dropped outcomes under +contention. WAL additionally keeps **readers** from blocking the writer, so +liveness/claim reads stay responsive while a write holds the connection. + +## Why path-keyed, not one global connection + +A single global connection would force *every* ledger file in the process to +serialize against one lock — breaking test isolation (each test uses its own +temp-dir ledger) and any future multi-tenant separation. Keying the registry by +**canonical path** means only handlers for the *same* file share a connection; +distinct files remain fully independent. This preserves the existing test +suite's parallelism and keeps tenants' ledgers isolated. + +## What this does *not* change + +- The public `CapabilityHandler::open` / `with_engineer_liveness` API. +- The schema and its `UNIQUE(outcome_id)` / `UNIQUE(session_id, cycle_id)` + invariants and foreign keys. +- The fail-visible contract: a write either commits or the caller sees + `PersistenceFailed`. Sharing a connection changes *how many* writers exist, not + *whether* failures are surfaced. +- Authorization, replay, and effect-lease semantics from the + [OODA capability API](../reference/ooda-capability-api.md). + +## See also + +- Reference: [Typed-outcome ledger connection registry API](../reference/typed-outcome-ledger-connection-registry-api.md). +- Runbook: [Diagnose "typed outcome persistence failed: database is locked"](../howto/diagnose-typed-outcome-database-is-locked.md). diff --git a/docs/howto/diagnose-typed-outcome-database-is-locked.md b/docs/howto/diagnose-typed-outcome-database-is-locked.md new file mode 100644 index 000000000..bad4d03ec --- /dev/null +++ b/docs/howto/diagnose-typed-outcome-database-is-locked.md @@ -0,0 +1,156 @@ +--- +title: Diagnose "typed outcome persistence failed: database is locked" +description: > + Operator runbook for the systemic typed-outcome PersistenceFailed burst fixed + under issue #4483. Recognise the concurrent-writer "database is locked" + signature in the typed-ooda/outcomes.sqlite3 ledger, confirm the shared + path-keyed connection registry, WAL/busy_timeout/foreign_keys pragmas, and + bounded busy-retry are in effect, run the concurrency regression test, and + localise a recurrence to an external writer or a bypassed open() path. +last_updated: 2026-07-23 +review_schedule: as-needed +owner: simard +doc_type: how-to +status: design — not yet implemented +related: + - ../concepts/typed-outcome-ledger-shared-connection.md + - ../reference/typed-outcome-ledger-connection-registry-api.md + - diagnose-and-recover-ooda-step-failures.md +--- + +# Diagnose "typed outcome persistence failed: database is locked" + +> **Spec-first (retcon) runbook.** This documents operating the fix for issue +> **#4483**, which is **not yet landed**. The documentation and implementation +> ship in the **same pull request**; flip `status:` to `implemented` on merge. +> Until then, "in effect" checks below describe the state you are verifying once +> the fix is present. + +## Symptom + +One or more terminal outcomes fail to persist, and the logs show: + +``` +typed outcome persistence failed: ... database is locked +``` + +(from `persistence(..)` → `CapabilityErrorCode::PersistenceFailed`). The +distinguishing signature of issue #4483 is a **burst**: several distinct goals +reaching a terminal in the same few-second window each emit the error, rather +than a single steady failure. The affected file is the typed-outcome ledger: + +``` +/typed-ooda/outcomes.sqlite3 +``` + +## Why it happens (one line) + +Multiple `CapabilityHandler` instances opened against that one file used to hold +**independent** connections; concurrent writers collided at SQLite's file lock +and one got `SQLITE_BUSY`. See +[the concept doc](../concepts/typed-outcome-ledger-shared-connection.md) for the +full explanation. + +## Step 1 — Recognise the burst signature + +Confirm it is the concurrency burst and not an unrelated I/O error: + +```bash +# Adjust the log source to your deployment. +journalctl --user -u 'simard*' --since '15 min ago' \ + | grep -E 'typed outcome persistence failed.*database is locked' +``` + +Look for **several distinct goal / session identifiers** clustered within the +same few seconds. A lone occurrence spread over minutes is more likely an +external writer (Step 4), disk pressure, or a permissions problem. + +## Step 2 — Confirm the shared connection registry is in effect + +The fix makes every handler for one file share a single connection. Verify the +registry and the shared field type exist: + +```bash +cd +grep -n 'OnceLock>\|SharedConn' src/typed_ooda/ledger.rs +grep -n 'fn apply_pragmas\|fn with_busy_retry\|fn is_sqlite_busy' src/typed_ooda/ledger.rs +``` + +**Pre-fix (bug present)** you will instead see: + +``` +connection: Mutex, // per-handler, independent connections +connection: Mutex::new(connection), +``` + +If you see the pre-fix form, the burst is expected under load — the fix has not +landed on this build. + +## Step 3 — Confirm the durability pragmas and retry + +```bash +grep -n 'journal_mode.*WAL\|WAL' src/typed_ooda/ledger.rs +grep -n 'busy_timeout' src/typed_ooda/ledger.rs # PRAGMA busy_timeout=5000 (== 5s) +grep -n 'foreign_keys' src/typed_ooda/ledger.rs # foreign_keys = ON +``` + +Then confirm WAL is actually active on a live ledger: + +```bash +sqlite3 "/typed-ooda/outcomes.sqlite3" 'PRAGMA journal_mode;' +# expect: wal +``` + +A `-wal` / `-shm` sidecar file next to `outcomes.sqlite3` is the on-disk sign +WAL is in use. + +## Step 4 — Localise a recurrence + +If the burst still appears **after** the fix is in effect, it must come from +outside the in-process shared connection: + +1. **An external process** (a stray CLI, a manual `sqlite3` write session, a + backup tool holding a write lock) is touching the ledger. Check: + ```bash + fuser -v "/typed-ooda/outcomes.sqlite3" 2>&1 || \ + lsof "/typed-ooda/outcomes.sqlite3" + ``` + Close the external writer; `with_busy_retry` should absorb brief overlaps. +2. **A bypassed `open` path** — some code constructed a raw `Connection` to the + ledger instead of going through `CapabilityHandler::open`, escaping the + registry. Search for it: + ```bash + grep -rn 'Connection::open' src/ | grep -i 'outcomes.sqlite3\|typed-ooda\|ledger' + ``` + All ledger access must flow through `CapabilityHandler::open`. +3. **Disk / filesystem** — a network filesystem with weak locking (NFS) can + break SQLite locking regardless of the registry. The ledger must live on a + local filesystem. + +## Step 5 — Run the concurrency regression test + +The fix ships with a regression that reproduces the #4483 burst — many handlers, +one ledger path, concurrent writes, asserting zero `database is locked` +failures: + +```bash +cargo test -p typed_ooda -- --nocapture 2>&1 | grep -iE 'lock|busy|persist' +# or target the specific test module for the ledger registry +cargo test --test typed_ooda_contracts 2>&1 | tail -20 +``` + +Green means the shared connection, pragmas, and retry are holding. If you can +reproduce a burst in production but the test is green, you are almost certainly +in a Step 4 case (external writer or bypassed `open`). + +## Escalation + +If none of the above localises it, capture: + +- the clustered log lines (Step 1) with goal/session IDs and timestamps, +- `PRAGMA journal_mode;` output and the presence/absence of `-wal`/`-shm`, +- `lsof` / `fuser` output for the ledger file, + +and attach them to a new issue referencing #4483 and the +[connection registry reference](../reference/typed-outcome-ledger-connection-registry-api.md). diff --git a/docs/index.md b/docs/index.md index fbce969e7..74291c5f9 100644 --- a/docs/index.md +++ b/docs/index.md @@ -27,6 +27,7 @@ Terminal sessions and repo-grounded engineer runs now bridge through one explici - [Typed-capability OODA architecture](./architecture/typed-ooda-loop.md) - Semantic/typed boundary, actor-session authority, durable terminals, effect outbox, and the explicit remaining migration boundary. - [OODA capability API](./reference/ooda-capability-api.md) - Terminal schemas, authorization, replay, effect leases, current limitations, errors, and policy configuration. - [Typed OODA goal-session deterministic rails](./reference/typed-ooda-goal-session-rails.md) — the two thin rail fixes that unblocked live OODA goals (#4076): propagating `AMPLIHACK_AGENT_BINARY` to the goal-session `recipe-runner-rs` subprocess (no silent `claude` fallback) and normalizing bare goal repo names to `rysweet/` at spawn admission, plus the additive Act-loop failure-detail log. +- [Typed-outcome ledger connection registry API](./reference/typed-outcome-ledger-connection-registry-api.md) — the path-keyed shared-connection registry (WAL + `busy_timeout` + `foreign_keys` pragmas, bounded busy-retry) that makes every handler for `typed-ooda/outcomes.sqlite3` share one serialized SQLite connection, closing the systemic "database is locked" `PersistenceFailed` burst (#4483). See the [why](./concepts/typed-outcome-ledger-shared-connection.md) and the [operator runbook](./howto/diagnose-typed-outcome-database-is-locked.md). - [Tutorial: Complete a typed OODA cycle](./tutorials/complete-a-typed-ooda-cycle.md) - Deterministic action, no-action, replay, and conflict examples. - [Tutorial: Run your first local session](./tutorials/run-your-first-local-session.md) - Exercise the local runtime through the primary CLI. - [Simard installer reference](./reference/simard-installer.md) - Shipped deployment contract for the binary, the owned `~/.local/bin/simard` PATH entrypoint and stale-orphan reconciliation, prompt assets, user systemd units, the post-deploy version-parity gate, rollback artifacts, and dry-run controls. diff --git a/docs/reference/typed-outcome-ledger-connection-registry-api.md b/docs/reference/typed-outcome-ledger-connection-registry-api.md new file mode 100644 index 000000000..9497e2f52 --- /dev/null +++ b/docs/reference/typed-outcome-ledger-connection-registry-api.md @@ -0,0 +1,329 @@ +--- +title: "Reference: Typed-Outcome Ledger Connection Registry API" +description: > + The process-global, path-keyed connection registry that makes every + CapabilityHandler opened against the same typed-outcome ledger file + (typed-ooda/outcomes.sqlite3) share a single serialized SQLite connection. + Covers the registry data structure, apply_pragmas (WAL + busy_timeout + + foreign_keys), the with_busy_retry / is_sqlite_busy bounded backoff, the + unchanged CapabilityHandler::open surface, lock ordering, error semantics, + security notes, and the concurrency regression contract that closes issue + #4483 (systemic typed-outcome PersistenceFailed under concurrent writers). +last_updated: 2026-07-23 +review_schedule: as-needed +owner: simard +doc_type: reference +status: design — not yet implemented +related: + - ../concepts/typed-outcome-ledger-shared-connection.md + - ../howto/diagnose-typed-outcome-database-is-locked.md + - ooda-capability-api.md + - engineer-claim-release-api.md + - typed-ooda-goal-session-rails.md +--- + +# Reference: Typed-Outcome Ledger Connection Registry API + +> **Spec-first (retcon) document.** This reference describes the *target* design +> for issue **#4483**. At the time of writing the fix is **not yet landed**: +> `CapabilityHandler` still holds a private `Mutex` and each `open` +> builds an independent connection. This document is the implementation +> specification. **The documentation and the implementation land in the same +> pull request** — when that PR merges, flip `status:` to `implemented`. Until +> then, treat present-tense statements below as the contract to build against, +> not as shipped behaviour. +> +> **Snippet disclaimer.** The Rust snippets below illustrate the *contract* +> (types, ordering, error mapping). They are not copied verbatim from a +> not-yet-written source; the final `ledger.rs` may differ in naming and layout +> as long as the observable contract in this document holds. + +## Problem this API closes + +The typed-outcome ledger is a single SQLite file: + +``` +/typed-ooda/outcomes.sqlite3 +``` + +(`typed_ooda::LEDGER_RELATIVE_PATH`, resolved by `typed_ooda::ledger_path`.) + +Multiple `CapabilityHandler` instances are opened against that **same file** +within one process during a single goal session — at minimum: + +- the goal-session **startup outbox worker** that drains pending effects + (`OutboxWorker::drain_pending`, wired in + `src/ooda_actions/advance_goal/typed_goal_session.rs`), and +- the **route executor** that records terminal outcomes and enqueues effects + (`typed_ooda::route` / `typed_ooda::executor`). + +Before this fix each `open` created its **own** `Connection` and wrapped it in a +**per-handler** `Mutex`. That mutex serializes access *within one handler* but +does nothing across handlers: two handlers hold two independent connections to +the same file. Concurrent write transactions from those distinct connections +collide at the SQLite file-lock layer and one returns `SQLITE_BUSY` +("database is locked"). `open` sets a `busy_timeout`, but under a burst of +simultaneous writers the timeout can still be exhausted. The busy error is +mapped through `persistence(..)` to `CapabilityErrorCode::PersistenceFailed`, +which aborts the record and **drops a durable terminal outcome from the audit +trail** — the systemic failure reported in issue #4483. + +The fix collapses "one connection per handler" into **one connection per ledger +file per process**, so the existing inner `Mutex` becomes a process-wide +serialization point and the cross-connection `SQLITE_BUSY` race is eliminated by +construction. + +## Design decisions + +| ID | Decision | +| --- | -------- | +| D1 | A process-global, **path-keyed** connection registry: `OnceLock>>>>`. One shared connection per canonical ledger path. | +| D2 | `CapabilityHandler.connection` changes type from `Mutex` to a shared `Arc>` (`SharedConn`). All handlers for the same file clone the **same** `Arc`. | +| D3 | `apply_pragmas` runs **once per newly created connection**, before `schema::initialize`: `journal_mode = WAL`, `busy_timeout = 5000ms`, `foreign_keys = ON`. | +| D4 | `with_busy_retry` wraps write transactions in a **bounded** retry that classifies `SQLITE_BUSY` / `SQLITE_LOCKED` via `is_sqlite_busy` and retries with short backoff before surfacing `PersistenceFailed`. Fail-visible: exhausted retries still return the error. | +| D5 | The startup outbox drain and the route executor now share one connection, so startup recovery no longer contends with the executor for the same file. | +| D6 | **First-init serialization**: the first opener of a path initializes the schema while holding the registry mutex; concurrent openers of the same path block, then clone the ready `Arc`. | +| D7 | **Lock ordering**: hold the registry mutex only to look up / insert and clone the `Arc`, then **release it before** locking the inner connection mutex. No nested acquisition → no deadlock. | +| D8 | The registry is **path-keyed**, not a single global connection. Distinct ledger files (e.g. per-test temp dirs) get distinct connections, preserving test isolation and multi-tenant separation. | + +## The connection registry + +```rust +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::{Arc, Mutex, OnceLock}; +use rusqlite::Connection; + +/// One serialized connection shared by every handler for a given ledger file. +type SharedConn = Arc>; + +/// Process-global registry keyed by canonical ledger path. +static LEDGER_CONNECTIONS: OnceLock>> = OnceLock::new(); + +fn registry() -> &'static Mutex> { + LEDGER_CONNECTIONS.get_or_init(|| Mutex::new(HashMap::new())) +} +``` + +Keying is by a **canonicalized** path so that `./x/outcomes.sqlite3` and an +absolute form of the same file resolve to one entry. If canonicalization fails +(the file does not exist yet), the registry falls back to the path as given — +the directory is created by the caller before `open`, and the first successful +`open` canonicalizes for subsequent lookups. + +## `apply_pragmas` + +```rust +fn apply_pragmas(connection: &Connection) -> CapabilityResult<()> { + // Readers never block the single writer; the single writer is already + // serialized by SharedConn, so WAL contention cannot burst across + // connections. + connection + .pragma_update(None, "journal_mode", "WAL") + .map_err(persistence)?; + // Defense-in-depth for any *external* contention (e.g. another OS process + // touching the file). Equivalent to `PRAGMA busy_timeout = 5000`. + connection + .busy_timeout(Duration::from_secs(5)) + .map_err(persistence)?; + // Enforce the effect_jobs -> terminal_outcomes foreign key at write time. + connection + .pragma_update(None, "foreign_keys", "ON") + .map_err(persistence)?; + Ok(()) +} +``` + +`apply_pragmas` runs exactly once, on the freshly opened `Connection`, **before** +`super::schema::initialize`, and while the registry mutex is held (D6). Handlers +that clone an already-registered `Arc` never re-run it. + +## `with_busy_retry` / `is_sqlite_busy` + +```rust +fn is_sqlite_busy(error: &rusqlite::Error) -> bool { + matches!( + error, + rusqlite::Error::SqliteFailure(e, _) + if e.code == rusqlite::ErrorCode::DatabaseBusy + || e.code == rusqlite::ErrorCode::DatabaseLocked + ) +} + +/// Retry a write closure a bounded number of times on SQLITE_BUSY/LOCKED, +/// with short backoff, then surface the error unchanged. +fn with_busy_retry( + mut op: impl FnMut() -> rusqlite::Result, +) -> rusqlite::Result { + const MAX_ATTEMPTS: u32 = 5; // ~low-hundreds-of-ms worst case + let mut attempt = 0; + loop { + match op() { + Ok(value) => return Ok(value), + Err(error) if is_sqlite_busy(&error) && attempt + 1 < MAX_ATTEMPTS => { + attempt += 1; + std::thread::sleep(backoff(attempt)); + } + Err(error) => return Err(error), // fail-visible + } + } +} +``` + +`with_busy_retry` is defense-in-depth. With `SharedConn` in place the *in-process* +writer is already single-threaded, so busy errors should not originate from this +process; the retry absorbs transient contention from an external process and +guarantees that a genuinely stuck write still ends as `PersistenceFailed` rather +than hanging. The constant (`MAX_ATTEMPTS ≈ 5`) and backoff schedule are +illustrative — reconcile the doc with the final constants once implemented. + +## `CapabilityHandler::open` — unchanged surface + +The public signature does **not** change: + +```rust +impl CapabilityHandler { + pub fn open(path: impl AsRef, policy: CapabilityPolicy) -> CapabilityResult; + pub fn with_engineer_liveness(self, liveness: Box) -> Self; +} +``` + +Only the body changes — it consults the registry instead of building a private +connection: + +```rust +pub fn open(path: impl AsRef, policy: CapabilityPolicy) -> CapabilityResult { + let key = canonical_key(path.as_ref()); + let mut table = registry() + .lock() + .map_err(|_| persistence_message("outcome ledger registry lock is poisoned"))?; + + let shared = match table.get(&key) { + Some(existing) => Arc::clone(existing), // D2/D7: clone + reuse + None => { + let mut connection = Connection::open(path.as_ref()).map_err(persistence)?; + apply_pragmas(&connection)?; // D3, before schema + super::schema::initialize(&mut connection, now_millis()).map_err(persistence)?; // D6 + let shared = Arc::new(Mutex::new(connection)); + table.insert(key, Arc::clone(&shared)); + shared + } + }; + drop(table); // D7: release registry mutex before any inner lock + + Ok(Self { connection: shared, policy, engineer_liveness: None }) +} +``` + +`lock()` — the single accessor of the connection field — is unchanged in +behaviour and still maps a poisoned mutex to a fail-visible `PersistenceFailed`: + +```rust +fn lock(&self) -> CapabilityResult> { + self.connection + .lock() + .map_err(|_| persistence_message("outcome ledger lock is poisoned")) +} +``` + +Because `connection` is now `Arc>`, `self.connection.lock()` +still yields `MutexGuard<'_, Connection>` and every existing call site +(`release_engineer_claim`, `record_action`, `claim_next_effect`, +`recover_expired_effects`, …) compiles unchanged. + +## Shared startup-recovery path (D5) + +`typed_goal_session.rs` opens the handler once and builds the startup +`OutboxWorker` from it: + +```rust +let handler = CapabilityHandler::open(&ledger_path, policy)? + .with_engineer_liveness(Box::new(WorktreeEngineerLiveness { .. })); + +let startup_worker = OutboxWorker::new( + &handler, &effects, "goal-session-startup-worker", Duration::from_secs(300), +); +if let Err(error) = startup_worker.drain_pending(32) { + eprintln!("[simard] typed OODA outbox startup recovery incomplete: {error}"); +} +``` + +The subsequent `route.execute(.., &handler, ..)` uses the **same** handler and +therefore the same `SharedConn`. Even if a second handler were opened against the +same `ledger_path`, D1–D2 guarantee it clones the same `Arc` — the startup drain +and the executor can never hold two competing connections to the file. + +## Error semantics + +| Situation | Result | +| --- | --- | +| Registry mutex poisoned | `PersistenceFailed` — `"outcome ledger registry lock is poisoned"` | +| Connection mutex poisoned | `PersistenceFailed` — `"outcome ledger lock is poisoned"` (unchanged) | +| `Connection::open` / pragma / `schema::initialize` fails | `PersistenceFailed` via `persistence(..)` | +| Write hits `SQLITE_BUSY`/`SQLITE_LOCKED`, retries exhausted | `PersistenceFailed` (fail-visible; never swallowed) | +| Duplicate terminal outcome | Existing `UNIQUE(outcome_id)` / `UNIQUE(session_id, cycle_id)` conflict handling is unchanged; the shared connection does not relax it | + +All ledger errors continue to funnel through `persistence` / +`persistence_message` into `CapabilityErrorCode::PersistenceFailed`. This fix +adds **no** new error code and preserves the fail-visible contract: an outcome is +either durably recorded or the caller sees an explicit error. + +## Security notes + +- **S — least exposure.** Registry keys are canonical ledger **file paths** only. + No record payloads, credentials, session identities, or actor scopes are held + in the registry. Its blast radius is a `PathBuf → Arc>` map. +- **S — audit-trail integrity.** The durable-terminal uniqueness invariants + (`UNIQUE(outcome_id)`, `UNIQUE(session_id, cycle_id)`) are enforced by the + schema, not by connection multiplicity. Sharing one connection only *serializes* + writers; it cannot cause a double-insert, and `with_busy_retry` re-runs the same + guarded transaction, so a retried insert of an already-committed outcome still + conflicts deterministically rather than duplicating. +- **S — referential integrity.** `foreign_keys = ON` (D3) keeps the + `effect_jobs.outcome_id → terminal_outcomes(outcome_id)` foreign key enforced, + so an effect can never be enqueued for a non-existent outcome. +- **S — fail-visible, never fail-open.** Poisoned locks and exhausted retries map + to `PersistenceFailed`. The ledger never silently degrades to "recorded" when a + write did not commit. +- **S — fail-closed liveness preserved.** The engineer-liveness reclaim gate + (`with_engineer_liveness`, `EngineerLiveness`) is untouched; with no provider a + claim is still treated as live and a duplicate spawn stays rejected. +- **S — no new external surface.** The registry is in-process only + (`static OnceLock`). It exposes no IPC, no network, and no filesystem paths + beyond the ledger file the process already owns. +- **S — test / tenant isolation.** Because keys are per-path (D8), distinct + ledgers never share a connection; one test or tenant cannot serialize against, + observe, or corrupt another's file. + +## Concurrency regression contract + +The fix is guarded by a regression test that reproduces issue #4483's burst: + +- Open **N** `CapabilityHandler`s against **one** ledger path (mirroring the + goal-session startup-worker + executor + concurrent sessions). +- Drive concurrent terminal-outcome / effect writes across those handlers. +- Assert **zero** `PersistenceFailed` results attributable to + `SQLITE_BUSY`/`database is locked`, and that every outcome that reported success + is durably present (row counts match, uniqueness intact). + +The test must **fail** against the pre-fix "connection per handler" code and +**pass** once the registry, pragmas, and retry land — that is the acceptance +signal for issue #4483. + +## What did not change + +- `CapabilityHandler::open` / `with_engineer_liveness` signatures. +- The `lock()` accessor contract and its poisoned-lock error message. +- The `persistence` / `persistence_message` mapping to `PersistenceFailed`. +- The schema (`terminal_outcomes`, `effect_jobs`, `engineer_claims`, …) and its + uniqueness / foreign-key constraints. +- The `OutboxWorker` API (`new`, `drain_pending`, `recover_startup`). +- Any authorization, replay, or effect-lease semantics from + [OODA capability API](./ooda-capability-api.md). + +## See also + +- Concept: [Typed-outcome ledger shared connection](../concepts/typed-outcome-ledger-shared-connection.md) — *why* one connection per file. +- Runbook: [Diagnose "typed outcome persistence failed: database is locked"](../howto/diagnose-typed-outcome-database-is-locked.md). +- [OODA capability API](./ooda-capability-api.md). +- [Engineer-Claim Release & Reclaim API](./engineer-claim-release-api.md). diff --git a/mkdocs.yml b/mkdocs.yml index c15fdaf6e..e4a68a7b4 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -108,6 +108,7 @@ nav: - Write-Authority Posture: concepts/write-authority-posture.md - Identity-Scoped Cognition (Seed Goals, Observe-Only Act): concepts/identity-scoped-cognition.md - OODA Loop Self-Detection: concepts/ooda-loop-self-detection.md + - Typed-Outcome Ledger Shared Connection: concepts/typed-outcome-ledger-shared-connection.md - Unified RecipeBrain: concepts/unified-recipe-brain.md - Prompt-Driven OODA Brain: concepts/prompt-driven-ooda-brain.md - Prompt-Driven Brain Iteration: concepts/prompt-driven-brain-iteration.md @@ -202,6 +203,7 @@ nav: - Brain Decision Parse Failures: howto/diagnose-brain-decision-parse-failures.md - Decide/Orient Parse Failures: howto/diagnose-decide-orient-parse-failures.md - Diagnose and Recover OODA Step Failures: howto/diagnose-and-recover-ooda-step-failures.md + - Diagnose Typed-Outcome "database is locked": howto/diagnose-typed-outcome-database-is-locked.md - Diagnose Journal E2BIG Spawn Failures: howto/diagnose-journal-e2big-spawn-failures.md - Diagnose Signal-Meeting E2BIG Spawn Failures: howto/diagnose-signal-meeting-e2big.md - Add a Safe Agent/Recipe Spawn Site: howto/add-a-safe-agent-spawn-site.md @@ -290,6 +292,7 @@ nav: - Investigate-Before-Reap API: reference/investigate-stale-engineer-api.md - Tombstoned-Goal Engineer Reaper API: reference/tombstoned-goal-engineer-reaper-api.md - Typed OODA Goal-Session Deterministic Rails: reference/typed-ooda-goal-session-rails.md + - Typed-Outcome Ledger Connection Registry API: reference/typed-outcome-ledger-connection-registry-api.md - Stable Goal-Session Identity API: reference/stable-goal-session-identity-api.md - Standing-Research Novelty-Directive API: reference/standing-research-goal-novelty-directive-api.md - Research-Goal Never-Idle Rail API: reference/research-goal-never-idle-rail-api.md diff --git a/src/ooda_actions/advance_goal/typed_goal_session.rs b/src/ooda_actions/advance_goal/typed_goal_session.rs index 5adaa6635..18267371a 100644 --- a/src/ooda_actions/advance_goal/typed_goal_session.rs +++ b/src/ooda_actions/advance_goal/typed_goal_session.rs @@ -148,7 +148,12 @@ pub(crate) fn run( std::time::Duration::from_secs(300), ); if let Err(error) = startup_worker.drain_pending(32) { - eprintln!("[simard] typed OODA outbox startup recovery incomplete: {error}"); + tracing::error!( + target: "simard::typed_ooda", + goal_id = %goal.id, + error = %error, + "typed OODA outbox startup recovery incomplete" + ); } let execution = match route.execute( repo_root, diff --git a/src/typed_ooda/ledger.rs b/src/typed_ooda/ledger.rs index cf677f2f8..480ce1b6e 100644 --- a/src/typed_ooda/ledger.rs +++ b/src/typed_ooda/ledger.rs @@ -1,8 +1,8 @@ -use std::collections::BTreeSet; +use std::collections::{BTreeSet, HashMap}; use std::io::{self, Read, Write}; use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; -use std::sync::{Mutex, MutexGuard}; +use std::sync::{Arc, Mutex, MutexGuard, OnceLock}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use rusqlite::{Connection, OptionalExtension, Transaction, TransactionBehavior, params}; @@ -12,6 +12,142 @@ use uuid::Uuid; use super::types::*; +/// Maximum attempts (initial try + retries) for acquiring an immediate write +/// transaction when SQLite reports the database busy/locked. Belt-and-suspenders +/// on top of process-level write serialization (one shared connection per DB +/// path) and the 5s per-connection `busy_timeout`, covering cross-process / +/// OS-level contention. On exhaustion the underlying error is surfaced unchanged +/// as a `PersistenceFailed`; it is never swallowed. +const BUSY_RETRY_MAX_ATTEMPTS: u32 = 5; + +/// One shared SQLite connection per typed-outcome DB path. +type SharedConnection = Arc>; + +/// Acquire an `Immediate` write transaction with a bounded, fail-visible retry +/// on `SQLITE_BUSY` / "database is locked". Evaluates to the live `Transaction` +/// on success; on a non-busy error or retry exhaustion it `return`s the +/// `persistence` error from the enclosing method (identical outward behavior to +/// the prior `.map_err(persistence)?`). +macro_rules! begin_immediate { + ($connection:expr) => {{ + let mut __attempt: u32 = 0; + loop { + match $connection.transaction_with_behavior(TransactionBehavior::Immediate) { + Ok(__transaction) => break __transaction, + Err(__error) => { + __attempt += 1; + if __attempt >= BUSY_RETRY_MAX_ATTEMPTS || !is_sqlite_busy(&__error) { + return Err(persistence(__error)); + } + std::thread::sleep(busy_backoff(__attempt)); + } + } + } + }}; +} + +/// Process-wide registry mapping a canonical typed-outcome DB path to its single +/// shared connection. Routing every handle through this registry means all OODA +/// cycles and the outbox startup-recovery path serialize on one WAL + +/// `busy_timeout` connection instead of opening independent connections that +/// contend on the file-level write lock (issue #4483). +fn connection_registry() -> &'static Mutex> { + static REGISTRY: OnceLock>> = OnceLock::new(); + REGISTRY.get_or_init(|| Mutex::new(HashMap::new())) +} + +/// Canonical registry key for a DB path. The DB file may not exist yet, so the +/// parent directory is canonicalized and the file name rejoined; this keeps +/// distinct temp DBs (per-test isolation) on distinct shared connections while +/// still collapsing different spellings of the same file to one entry. +fn canonical_key(path: &Path) -> PathBuf { + let Some(file_name) = path.file_name() else { + return path.to_path_buf(); + }; + // A parent-less path (bare relative file name) is anchored to the current + // working directory so it still canonicalizes to an absolute key, keeping + // the invariant that one file maps to exactly one shared connection. + let directory = path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + match directory.canonicalize() { + Ok(canonical) => canonical.join(file_name), + Err(_) => path.to_path_buf(), + } +} + +/// Return the shared connection for `path`, creating and initializing it (WAL + +/// `busy_timeout` pragmas, then schema migration) exactly once per canonical +/// path. Every subsequent handle for the same path clones the same `Arc`. +fn shared_connection(path: &Path) -> CapabilityResult { + let key = canonical_key(path); + let mut registry = connection_registry() + .lock() + .map_err(|_| persistence_message("typed outcome connection registry is poisoned"))?; + if let Some(existing) = registry.get(&key) { + return Ok(Arc::clone(existing)); + } + let mut connection = Connection::open(path).map_err(persistence)?; + apply_connection_pragmas(&connection)?; + super::schema::initialize(&mut connection, now_millis()).map_err(persistence)?; + let shared: SharedConnection = Arc::new(Mutex::new(connection)); + registry.insert(key, Arc::clone(&shared)); + Ok(shared) +} + +/// Apply the durability + concurrency pragmas every typed-outcome connection +/// must run: foreign keys on, 5s busy timeout, and WAL journaling. Applied +/// unconditionally at connect (defense-in-depth; WAL also persists on disk). +fn apply_connection_pragmas(connection: &Connection) -> CapabilityResult<()> { + connection + .busy_timeout(Duration::from_secs(5)) + .map_err(persistence)?; + connection + .execute_batch("PRAGMA foreign_keys = ON;") + .map_err(persistence)?; + // SQLite reports the *resulting* journal mode without erroring when it + // cannot honor the request (e.g. an unsupported filesystem). Verify WAL was + // actually engaged instead of silently proceeding, since the whole + // shared-connection concurrency fix depends on WAL being active. + let journal_mode: String = connection + .query_row("PRAGMA journal_mode = WAL", [], |row| row.get(0)) + .map_err(persistence)?; + if !journal_mode.eq_ignore_ascii_case("wal") { + return Err(persistence_message(format!( + "typed outcome connection could not enter WAL journal mode (got {journal_mode:?}); \ + the shared-connection concurrency fix requires WAL" + ))); + } + Ok(()) +} + +/// True iff `error` is a SQLite busy/locked condition eligible for retry. +fn is_sqlite_busy(error: &rusqlite::Error) -> bool { + match error { + rusqlite::Error::SqliteFailure(failure, message) => { + matches!( + failure.code, + rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked + ) || message + .as_deref() + .is_some_and(|message| message.contains("database is locked")) + } + _ => false, + } +} + +/// Exponential backoff of `5 * 2^attempt` milliseconds, capped at 100ms. +/// +/// `begin_immediate!` calls this only for `attempt` 1..=4 (it returns at +/// `BUSY_RETRY_MAX_ATTEMPTS = 5` before sleeping), so the effective sleep +/// sequence is 10ms, 20ms, 40ms, 80ms. The 100ms cap guards against overflow +/// and is unreachable at the current `BUSY_RETRY_MAX_ATTEMPTS`. +fn busy_backoff(attempt: u32) -> Duration { + let millis = 5u64.saturating_mul(1u64 << attempt.min(5)).min(100); + Duration::from_millis(millis) +} + #[derive(Clone, Debug, serde::Deserialize, Eq, PartialEq, serde::Serialize)] pub struct EffectKind(String); @@ -144,7 +280,7 @@ pub trait EngineerLiveness: Send + Sync { } pub struct CapabilityHandler { - connection: Mutex, + connection: SharedConnection, policy: CapabilityPolicy, engineer_liveness: Option>, } @@ -253,13 +389,9 @@ impl std::fmt::Debug for CapabilityHandler { impl CapabilityHandler { pub fn open(path: impl AsRef, policy: CapabilityPolicy) -> CapabilityResult { - let mut connection = Connection::open(path.as_ref()).map_err(persistence)?; - connection - .busy_timeout(Duration::from_secs(5)) - .map_err(persistence)?; - super::schema::initialize(&mut connection, now_millis()).map_err(persistence)?; + let connection = shared_connection(path.as_ref())?; Ok(Self { - connection: Mutex::new(connection), + connection, policy, engineer_liveness: None, }) @@ -287,9 +419,7 @@ impl CapabilityHandler { /// `docs/reference/engineer-claim-release-api.md`. pub fn release_engineer_claim(&self, claim_key: &str) -> CapabilityResult<()> { let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); transaction .execute( "DELETE FROM engineer_claims WHERE claim_key = ?1", @@ -369,9 +499,7 @@ impl CapabilityHandler { )?; let binding = ActorBinding::new(actor, cycle_id, goal_id, repository)?; let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); if let Some(existing) = replay_request(&transaction, request_id, "actor_session", &fingerprint)? { @@ -492,9 +620,7 @@ impl CapabilityHandler { &("privileged_approval_v1", effect_id), )?; let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); if let Some(existing) = replay_request( &transaction, request_id, @@ -617,9 +743,7 @@ impl CapabilityHandler { let fingerprint = fingerprint(actor, &self.policy.revision, &request)?; let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); if let Some(existing) = replay_request( &transaction, &request.identity.request_id, @@ -781,9 +905,7 @@ impl CapabilityHandler { } let fingerprint = fingerprint(actor, &self.policy.revision, &request)?; let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); if let Some(existing) = replay_request( &transaction, &request.identity.request_id, @@ -869,9 +991,7 @@ impl CapabilityHandler { self.validate_process_execution(actor, request)?; let fingerprint = fingerprint(actor, &self.policy.revision, &("process_exec_v1", request))?; let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); if let Some(existing) = replay_request( &transaction, &request.identity.request_id, @@ -952,9 +1072,7 @@ impl CapabilityHandler { ) -> CapabilityResult<()> { let result_json = serde_json::to_vec(record).map_err(serialization)?; let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); let changed = transaction .execute( "UPDATE process_executions SET status=?2, result_json=?3 @@ -1138,9 +1256,7 @@ impl CapabilityHandler { &("claim_next_effect_v1", worker, lease_millis), )?; let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); if let Some(existing) = replay_request(&transaction, request_id, "effect_claim", &fingerprint)? { @@ -1207,9 +1323,7 @@ impl CapabilityHandler { ), )?; let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); if let Some(existing) = replay_request(&transaction, request_id, "effect_claim", &fingerprint)? { @@ -1251,9 +1365,7 @@ impl CapabilityHandler { let actor = AuthenticatedToolContext::new("effect-recovery", "system", std::iter::empty()); let fingerprint = fingerprint(&actor, &self.policy.revision, &("recover_effects_v1", now))?; let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); if let Some(existing) = replay_request(&transaction, request_id, "effect_recovery", &fingerprint)? { @@ -1313,9 +1425,7 @@ impl CapabilityHandler { ), )?; let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); if replay_request::( &transaction, request_id, @@ -1372,9 +1482,7 @@ impl CapabilityHandler { ), )?; let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); if replay_request::( &transaction, request_id, @@ -1467,9 +1575,7 @@ impl CapabilityHandler { ), )?; let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); if replay_request::( &transaction, request_id, @@ -1529,9 +1635,7 @@ impl CapabilityHandler { }; let result_json = serde_json::to_vec(result).map_err(serialization)?; let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); if replay_request::(&transaction, request_id, "effect_finish", &fingerprint)? .is_some() { @@ -1577,9 +1681,7 @@ impl CapabilityHandler { fingerprint: String, ) -> CapabilityResult { let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; + let transaction = begin_immediate!(connection); if let Some(existing) = replay_request(&transaction, &identity.request_id, "terminal", &fingerprint)? { diff --git a/tests/typed_ooda_outcome_lock_regression.rs b/tests/typed_ooda_outcome_lock_regression.rs new file mode 100644 index 000000000..ec83e8045 --- /dev/null +++ b/tests/typed_ooda_outcome_lock_regression.rs @@ -0,0 +1,216 @@ +//! Regression: concurrent typed-OODA outcome persistence + outbox startup +//! recovery must never surface `database is locked` / `SQLITE_BUSY`. +//! +//! Reproduces issue #4483: on a daemon restart, multiple concurrent OODA cycles +//! (each a distinct goal opening its own `CapabilityHandler` on the SAME +//! typed-outcome SQLite file) contend with the outbox startup-recovery path +//! (`drain_pending`). Every handle to the ledger DB must funnel through a single +//! shared, WAL + busy_timeout connection so opens and writes serialize instead +//! of colliding on the file-level write lock. +//! +//! TDD status: RED until every typed-outcome DB handle is routed through the +//! shared connection factory (serialized access + bounded busy retry). Before +//! that fix, each `CapabilityHandler::open` creates an independent connection; +//! when many cycles start at once (as on restart) their first-init + +//! `PRAGMA journal_mode = WAL` acquisitions pile up past the per-connection 5s +//! `busy_timeout`, and at least one `open` / `record_*` / `drain_pending` call +//! returns `typed outcome persistence failed: database is locked`. +//! +//! After the fix there is exactly one shared connection per path, so the +//! concurrent-open count is irrelevant: initialization happens once, every +//! write serializes through the shared mutex, and the test is fast and green. + +use std::path::Path; +use std::sync::{Arc, Barrier}; +use std::thread; +use std::time::Duration; + +use simard::typed_ooda::{ + AuthenticatedToolContext, CapabilityGrant, CapabilityHandler, CapabilityPolicy, + EffectExecutionError, EffectExecutor, EffectJob, EffectResult, OpaqueBytes, OutboxWorker, + RecordNoActionRequest, RepositoryRef, TerminalRequestIdentity, +}; + +/// Number of concurrent OODA cycles (distinct goals) opening the ledger at once. +/// +/// The journal evidence was a six-goal post-restart burst, but six independent +/// connections are masked by the per-connection 5s `busy_timeout`. To surface +/// the latent file-lock race deterministically we scale the simultaneous-open +/// count well past the point where cumulative first-init serialization exceeds +/// that timeout. With the shared-connection factory in place there is exactly +/// one connection, so this count is irrelevant and the test stays fast + green. +const CONCURRENT_GOALS: usize = 112; + +/// Terminal outcomes each cycle persists back-to-back. First-init contention +/// dominates the race, so a small write count keeps the post-fix run fast. +const WRITES_PER_GOAL: usize = 4; + +/// Number of concurrent startup-recovery workers draining the outbox while the +/// cycles persist outcomes. Added to the simultaneous-open pressure. +const RECOVERY_WORKERS: usize = 16; + +/// Iterations of `drain_pending` each recovery worker performs. +const RECOVERY_PASSES: usize = 4; + +struct NoopEffects; + +impl EffectExecutor for NoopEffects { + fn execute(&self, _job: &EffectJob) -> Result { + Ok(EffectResult::Succeeded { + evidence: Vec::new(), + }) + } +} + +fn is_lock_error(message: &str) -> bool { + let lowered = message.to_ascii_lowercase(); + lowered.contains("database is locked") || lowered.contains("sqlite_busy") +} + +fn open_handler(path: &Path) -> Result { + // Opening also runs schema initialization; on a fresh file this includes + // `PRAGMA journal_mode = WAL` and the migration transaction, which is the + // real contention point across racing connections. + CapabilityHandler::open(path, CapabilityPolicy::new("policy-v1")) + .map_err(|error| error.to_string()) +} + +/// Pre-build the `(actor, request)` payloads for one goal so that, once the +/// starting barrier releases, each thread's first observable action is +/// `open()` — maximizing the number of connections simultaneously inside the +/// first-init window where the lock race lives. +fn build_no_action_requests(goal: usize) -> Vec<(AuthenticatedToolContext, RecordNoActionRequest)> { + (0..WRITES_PER_GOAL) + .map(|iteration| { + let session_id = format!("session-goal-{goal}"); + let cycle_id = format!("cycle-{goal}-{iteration}"); + let goal_id = format!("goal-{goal}"); + let request_id = format!("request-{goal}-{iteration}"); + + let actor = AuthenticatedToolContext::new( + "goal-session-actor", + session_id.clone(), + [CapabilityGrant::RecordNoAction], + ) + .scoped_to_repository(RepositoryRef::new("rysweet", "Simard")) + .bound_to_cycle_goal(cycle_id.clone(), goal_id.clone()); + + let request = RecordNoActionRequest { + identity: TerminalRequestIdentity::new(request_id, session_id, cycle_id, goal_id), + reason: OpaqueBytes::from(b"no protocol action this cycle".to_vec()), + raw_semantic: OpaqueBytes::from(vec![0x00, 0xff, b'N']), + evidence: Vec::new(), + }; + (actor, request) + }) + .collect() +} + +#[test] +fn concurrent_cycles_and_startup_recovery_never_lock_the_outcome_ledger() { + let dir = tempfile::tempdir().expect("tempdir"); + let ledger_path = dir.path().join("outcomes.sqlite3"); + + // Every worker opens its OWN handler onto the shared file and they all race + // from the same starting line — reproducing the post-restart burst where + // distinct goals and the outbox recovery path open connections at once. + let barrier = Arc::new(Barrier::new(CONCURRENT_GOALS + RECOVERY_WORKERS)); + let mut handles = Vec::new(); + + // Writer threads: each distinct goal is its own OODA cycle with its own + // handler/connection onto the shared ledger file. + for goal in 0..CONCURRENT_GOALS { + let path = ledger_path.clone(); + let barrier = Arc::clone(&barrier); + handles.push(thread::spawn(move || -> Vec { + // Build payloads BEFORE synchronizing so the post-barrier hot path + // is only `open()` + writes. + let requests = build_no_action_requests(goal); + let mut lock_errors = Vec::new(); + + barrier.wait(); + let handler = match open_handler(&path) { + Ok(handler) => handler, + Err(message) => { + if is_lock_error(&message) { + lock_errors.push(message); + } + return lock_errors; + } + }; + for (actor, request) in requests { + if let Err(error) = handler.record_no_action(&actor, request) { + let message = error.to_string(); + if is_lock_error(&message) { + lock_errors.push(message); + } + } + } + lock_errors + })); + } + + // Startup-recovery threads: the outbox recovery path (`drain_pending`) + // contends with the writers, exactly as it does on daemon restart. + for worker in 0..RECOVERY_WORKERS { + let path = ledger_path.clone(); + let barrier = Arc::clone(&barrier); + handles.push(thread::spawn(move || -> Vec { + let worker_id = format!("startup-recovery-{worker}"); + let mut lock_errors = Vec::new(); + + barrier.wait(); + let handler = match open_handler(&path) { + Ok(handler) => handler, + Err(message) => { + if is_lock_error(&message) { + lock_errors.push(message); + } + return lock_errors; + } + }; + let effects = NoopEffects; + for _ in 0..RECOVERY_PASSES { + let outbox = + OutboxWorker::new(&handler, &effects, &worker_id, Duration::from_secs(300)); + if let Err(error) = outbox.drain_pending(32) { + let message = error.to_string(); + if is_lock_error(&message) { + lock_errors.push(message); + } + } + } + lock_errors + })); + } + + let mut lock_errors: Vec = Vec::new(); + for handle in handles { + lock_errors.extend(handle.join().expect("worker thread must not panic")); + } + + assert!( + lock_errors.is_empty(), + "concurrent outcome persistence + startup recovery must never report a locked \ + database, but observed {} lock failure(s); first few: {:?}", + lock_errors.len(), + lock_errors.iter().take(5).collect::>(), + ); + + // Every distinct (session, cycle) terminal must be durably persisted: no + // write may have been silently dropped in the name of avoiding the lock. + let verifier = open_handler(&ledger_path).expect("open verifier handler"); + for goal in 0..CONCURRENT_GOALS { + let session_id = format!("session-goal-{goal}"); + for iteration in 0..WRITES_PER_GOAL { + let cycle_id = format!("cycle-{goal}-{iteration}"); + let terminal = verifier + .terminal_for_cycle(&session_id, &cycle_id) + .expect("terminal lookup must not fail"); + assert!( + terminal.is_some(), + "expected a durable terminal outcome for {session_id}/{cycle_id}", + ); + } + } +}