From f593341fd975c8a3c8ea8fb5ef3d7f67d8e9863a Mon Sep 17 00:00:00 2001 From: rysweet Date: Wed, 22 Jul 2026 21:01:00 +0000 Subject: [PATCH 1/5] fix(typed-ooda): benign goal-race no-op + outbox busy-retry (#4468) Systemic concurrency/lifecycle defect: prepared goal-session effects dispatched after the goal was legitimately completed/removed mapped to DownstreamFailed ("goal disappeared before effect dispatch"), and the outbox/outcome SQLite writes collided under startup recovery + concurrent cycles ("database is locked"). Both fired across many distinct goals => systemic, not per-goal. Fixes (additive; no schema/semantic or happy-path change): (a) Benign goal-lifecycle race -> counted, structured no-op. - EffectExecutionError gains a `no_op` flag + `benign_no_op()` constructor (permanent:true, no_op:true). Checked BEFORE the !permanent arm in execute_claimed: emits a structured tracing event, increments `typed_ooda_effect_benign_no_op`, and closes the outbox row as Succeeded{evidence:vec![]} (":noop" request-id) instead of DownstreamFailed. Never redispatched. - Reclassified only the two pre-side-effect goal-not-found sites (require_goal_repository, spawn) to benign_no_op. Repo-mismatch, metadata, and post-spawn sites stay permanent (isolation preserved). (b) SQLite busy/locked contention -> bounded retry-with-backoff. - retry_on_busy wraps recover_expired_effects, release_effect_for_retry, and finish_effect (lock acquired inside the closure, released across each backoff sleep). MAX_ATTEMPTS=6, exp backoff capped 400ms. - Typed detection only: persistence() stamps BUSY_PERSISTENCE_MARKER from the rusqlite ErrorCode (is_busy_locked); capability_error_is_busy keys the retry. Immune to injected "database is locked" strings. WAL + busy_timeout(5s) baseline preserved. Exhaustion -> PersistenceFailed (never masked). Cleanup: startup-recovery eprintln! -> structured tracing::warn!. Regression tests: dispatch_after_goal_removed_is_benign_no_op, permanent_effect_failure_is_not_treated_as_benign_no_op, concurrent_outbox_writes_never_surface_database_is_locked. Reference doc added. fmt/clippy/build/54 typed_ooda tests green. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- docs/index.md | 1 + ...nign-goal-race-and-outbox-serialization.md | 603 ++++++++++++++++++ .../advance_goal/typed_goal_session.rs | 22 +- src/typed_ooda/executor.rs | 172 +++++ src/typed_ooda/ledger.rs | 473 +++++++++++--- 5 files changed, 1171 insertions(+), 100 deletions(-) create mode 100644 docs/reference/typed-ooda-benign-goal-race-and-outbox-serialization.md diff --git a/docs/index.md b/docs/index.md index fbce969e7..076a2b90f 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-OODA benign goal-race no-op and outbox write serialization](./reference/typed-ooda-benign-goal-race-and-outbox-serialization.md) — the two additive resilience fixes for the systemic decide→act dispatch/persistence defect (#4468): a goal legitimately completed/removed between effect prepare and dispatch becomes a benign, counted no-op outcome instead of `DownstreamFailed`, and the outbox/outcome SQLite writes gain a bounded busy/locked retry-with-backoff so startup recovery and concurrent cycles stop colliding on `database is locked`. - [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-ooda-benign-goal-race-and-outbox-serialization.md b/docs/reference/typed-ooda-benign-goal-race-and-outbox-serialization.md new file mode 100644 index 000000000..761fc479e --- /dev/null +++ b/docs/reference/typed-ooda-benign-goal-race-and-outbox-serialization.md @@ -0,0 +1,603 @@ +--- +title: Typed-OODA benign goal-race no-op and outbox write serialization +description: Reference for the two additive resilience behaviours in the typed-OODA decide→act effect-dispatch and outcome-persistence pipeline — (a) a goal legitimately completed or removed between effect *prepare* and *dispatch* is recorded as a benign, structured, counted no-op outcome instead of a DownstreamFailed cycle failure, and (b) the outbox/outcome SQLite writes are serialized with a bounded busy/locked retry-with-backoff so startup recovery and concurrent cycles stop colliding on "database is locked". Covers the EffectExecutionError::benign_no_op constructor and no_op flag, the execute_claimed no-op dispatch arm, the reclassified vs. still-permanent goal-not-found sites, the is_busy_locked classifier and retry_on_busy wrapper, the structured tracing events and counters, and the journal signatures the fix eliminates. +last_updated: 2026-07-22 +review_schedule: as-needed +owner: simard +doc_type: reference +status: implemented +related: + - ../architecture/typed-ooda-loop.md + - ./no-bridge-naming-guard.md + - ../operations/cognitive-memory-durability.md + - ../../src/typed_ooda/executor.rs + - ../../src/typed_ooda/ledger.rs + - ../../src/ooda_actions/advance_goal/typed_goal_session.rs +--- + +# Typed-OODA benign goal-race no-op and outbox write serialization + +> **Status: implemented (issue #4468).** Two additive behaviours in the +> typed-OODA `decide → act` effect-dispatch and outcome-persistence pipeline: +> +> 1. A prepared goal-session effect whose goal was **legitimately completed or +> removed** between *prepare* and *dispatch* is now recorded as a **benign, +> structured, counted no-op** outcome — a finished outbox row and a +> `tracing` event — instead of raising +> `EffectExecutionError::permanent("goal disappeared before effect dispatch")` +> and mapping to `CycleErrorCode::DownstreamFailed`. +> 2. The outbox/outcome SQLite writes are **serialized with a bounded +> busy/locked retry-with-backoff** so startup recovery and concurrent cycles +> stop colliding on `database is locked`. +> +> Both changes are additive. The outcome schema, `EffectResult`, and every +> happy-path effect semantic are unchanged; there is **no** happy-path behaviour +> change. No `Bridge` naming is introduced (see +> [No-Bridge naming guard](./no-bridge-naming-guard.md)); observability is +> structured `tracing` + OTel only, with no stray `print!`/`println!`/`eprintln!` +> and no silent fallbacks — every swallowed race emits a structured, counted +> outcome. + +The typed-OODA executor prepares an effect (claims an outbox row, resolves the +goal's authenticated repository) and then dispatches it. Between those two +steps another cycle — or the goal's own successful completion — can remove the +goal record. That is a **legitimate, benign race**, not a failure: the work the +effect represents is already moot. Before this fix the dispatch site treated the +missing goal as a permanent downstream failure, so seven distinct goals in a +single six-hour window each burned an OODA cycle on the same shared signature. +In the same window the outbox/outcome store took ten `database is locked` hits, +nine of them during **startup recovery** racing live cycles for the single +writer connection. + +This reference describes the finished behaviour: what is reclassified as a +no-op (and, importantly, what is **not**), how the no-op outcome is shaped and +counted, and how the SQLite writes are serialized. For the surrounding pipeline +see [Typed-OODA architecture](../architecture/typed-ooda-loop.md). + +## Contents + +- [The two journal signatures this eliminates](#the-two-journal-signatures-this-eliminates) +- [Benign goal-race no-op](#benign-goal-race-no-op) + - [Which sites are reclassified](#which-sites-are-reclassified) + - [Which sites stay permanent](#which-sites-stay-permanent) + - [The no-op outcome shape](#the-no-op-outcome-shape) + - [`EffectExecutionError` API](#effectexecutionerror-api) + - [`execute_claimed` dispatch arm](#execute_claimed-dispatch-arm) +- [Outbox / outcome write serialization](#outbox--outcome-write-serialization) + - [`is_busy_locked` classifier](#is_busy_locked-classifier) + - [`retry_on_busy` wrapper](#retry_on_busy-wrapper) + - [Connection PRAGMAs](#connection-pragmas) +- [Observability](#observability) +- [Configuration](#configuration) +- [Verification](#verification) +- [Security model](#security-model) +- [Examples](#examples) +- [When the no-op does *not* fire](#when-the-no-op-does-not-fire) + +## The two journal signatures this eliminates + +Both defects surface in the OODA journal: + +```bash +journalctl --user -u simard-ooda --since '-6h' +``` + +| Signature (before fix) | Count/6h | Root cause | Now | +| --- | --- | --- | --- | +| `typed goal-session effect incomplete (DownstreamFailed): goal disappeared before effect dispatch` | 7, across 7 distinct goals | goal completed/removed between prepare and dispatch | benign counted no-op; cycle succeeds | +| `typed outcome persistence failed: database is locked` | 10 (9 at `typed OODA outbox startup recovery incomplete`, 1 as `typed goal-session cycle failed (ToolFailed)`) | SQLite lock contention on the outbox/outcome store under concurrent writers + startup recovery | bounded retry-with-backoff; write completes | + +One shared signature spanning many goals is the tell that this is **systemic**, +not per-goal — the fix is in the shared dispatch and persistence path, not in +any single goal handler. + +## Benign goal-race no-op + +The named defect lives at +[`src/ooda_actions/advance_goal/typed_goal_session.rs`](https://github.com/rysweet/Simard/blob/main/src/ooda_actions/advance_goal/typed_goal_session.rs) +in `require_goal_repository`, which resolves the goal's authenticated repository +just before the effect is dispatched. If the goal is gone, the work is moot — +there is nothing to advance — so the effect becomes a no-op rather than a +failure. + +### Which sites are reclassified + +Exactly the two **pre-side-effect** goal-not-found misses return +`EffectExecutionError::benign_no_op(...)`: + +| Site | Location | Why benign | +| --- | --- | --- | +| `require_goal_repository` goal-not-found | `typed_goal_session.rs` — the `.ok_or_else(...)` on the active-goals lookup ("goal disappeared before effect dispatch") | The named defect. No process has been spawned; nothing to undo. | +| spawn goal-not-found ("before spawn") | `typed_goal_session.rs` — the pre-spawn active-goals lookup | Same pre-side-effect race, same file. Reclassified for consistency. | + +Both are races the system is *designed* to tolerate: the goal reached a terminal +state (completed/removed) between prepare and dispatch. + +### Which sites stay permanent + +Reclassification is deliberately narrow. These sites keep returning +`EffectExecutionError::permanent(...)` and continue mapping to +`CycleErrorCode::DownstreamFailed`, because each is either a genuine error or a +site where a side effect has already happened: + +| Site | Why it stays permanent | +| --- | --- | +| **Post-spawn goal-not-found** ("after engineer spawn") | An engineer process was **already spawned**. Swallowing this would orphan a live process — the outbox row must record a real failure so recovery can act. | +| **Repository mismatch** (effect repo ≠ authenticated goal repo) | A genuine repo/tenant-isolation error. Treating it as benign would be an authorization-suppression primitive — an effect bound to a *different* repository must never be silently closed. | +| **`goal_repository()` resolution error** | The goal exists but its repository metadata is malformed — a real error, not a race. | +| **Goal already assigned** (`already_assigned` → "goal already has an assigned engineer") | The goal exists and already has a live engineer. This is a duplicate-spawn guard, not a race — the effect must not proceed, and it is not a benign disappearance. | + +The negative regression test `repo_mismatch_still_downstream_failed` locks the +repository-mismatch boundary in: a mismatch must still produce +`DownstreamFailed`. The `already_assigned` and post-spawn sites are covered by +their own duplicate-spawn / orphan-reconciliation tests and are unchanged by +this work — they are listed here only to make the reclassification boundary +exhaustive. + +### The no-op outcome shape + +A benign no-op is a **completed** outbox row, not a skipped one. `execute_claimed` +calls `finish_effect` with a succeeded result carrying **empty evidence**: + +```rust +EffectResult::Succeeded { evidence: vec![] } +``` + +This closes the outbox row exactly as any completed effect would (no +re-dispatch, no retry), never maps to `DownstreamFailed`, and adds nothing to +the outcome schema — empty-evidence `Succeeded` reuses the existing +`EffectResult::Succeeded` variant and serde round-trips `vec![]` unchanged. The +finish uses a distinct `noop` operation suffix in its idempotency key +(`effect_mutation_request_id(&job, "noop")` → `"{effect_id}:{lease_generation}:noop"`), +partitioning it from the `complete` / `failed` / `retry` request-id space so a +no-op can never collide with, or double-close, a real completion. + +### `EffectExecutionError` API + +`EffectExecutionError` (in +[`src/typed_ooda/executor.rs`](https://github.com/rysweet/Simard/blob/main/src/typed_ooda/executor.rs)) +gains one additive, internal, **non-serialized** field and one constructor: + +```rust +// src/typed_ooda/executor.rs + +pub struct EffectExecutionError { + message: String, + /// Terminal (non-retryable). Both `permanent` and `benign_no_op` set this. + permanent: bool, + /// Legitimately-removed goal between prepare and dispatch: record a benign, + /// counted no-op outcome instead of a DownstreamFailed cycle failure. + /// Internal, in-memory only — never read from a persisted outbox row or + /// any external message. + no_op: bool, +} + +impl EffectExecutionError { + /// Terminal failure. `no_op = false`. + pub fn permanent(message: impl Into) -> Self; + + /// Transient failure; the effect is released for retry. `no_op = false`. + pub fn retryable(message: impl Into) -> Self; + + /// A goal legitimately completed/removed between prepare and dispatch. + /// Sets `permanent = true` (defense-in-depth: even if the no-op arm were + /// bypassed it is still terminal, never retried) AND `no_op = true`. + pub fn benign_no_op(message: impl Into) -> Self; +} +``` + +`benign_no_op` sets **both** `permanent` and `no_op`. The `no_op` flag is a +pure in-memory discriminator: it is never serialized into an outbox row and +never trusted from external input, so it cannot be forged to suppress a real +failure. + +### `execute_claimed` dispatch arm + +The dispatch match in `execute_claimed` gains a **first** `Err` arm, evaluated +**before** the existing `if !error.permanent` (retryable) and permanent +branches: + +```rust +let result = match self.effects.execute(&job) { + Ok(result) => result, + Err(error) if error.no_op => { + // Benign race: the goal was legitimately completed/removed between + // prepare and dispatch. Emit a structured, counted no-op outcome. + tracing::warn!( + effect_id = %job.effect_id, + goal_id = %job.goal_id, + reason = "goal-removed-before-dispatch", + "typed goal-session effect no-op: goal completed or removed between prepare and dispatch", + ); + let _ = crate::self_metrics::record_metric( + "typed_ooda_effect_benign_no_op", + 1.0, + &job.effect_id, + ); + self.handler + .finish_effect( + &job, + &effect_mutation_request_id(&job, "noop"), + SystemTime::now(), + &EffectResult::Succeeded { evidence: vec![] }, + ) + .map_err(|failure| { + CycleError::new(CycleErrorCode::PersistenceFailed, failure.to_string()) + })?; + return Ok(()); + } + Err(error) => { + // ... existing retryable (!permanent) and permanent branches, unchanged + } +}; +``` + +Ordering matters: because `benign_no_op` also sets `permanent = true`, the arm +guard is `error.no_op` and it is placed **first**, so a benign race is handled +as a no-op and never falls through to the retry or `DownstreamFailed` branches. +A persistence failure while closing the no-op row still surfaces as +`PersistenceFailed` — the no-op path never masks a real write error. + +## Outbox / outcome write serialization + +The outbox/outcome store is one SQLite database file, but it is **not** reached +through a single connection. Each `CapabilityHandler` in +[`src/typed_ooda/ledger.rs`](https://github.com/rysweet/Simard/blob/main/src/typed_ooda/ledger.rs) +holds its own `Mutex`, and `CapabilityHandler::open(&ledger_path, …)` +is invoked from **many** call sites against the same file — per goal-session +cycle (`typed_goal_session.rs`), the subordinate path, overseer wiring, and the +operator CLI (a **separate process**). The per-handler `Mutex` +serializes writes **within one handler only**; it does nothing to order writes +issued by a *different* handler or a *different* process. Those cross-connection +writes contend at SQLite's own file-lock layer. + +WAL + a 5 s `busy_timeout` already absorb most of that contention, but under +concurrent live cycles **plus** startup recovery replaying expired effects (and, +occasionally, an operator-CLI write), a writer can still exhaust the +`busy_timeout` and surface `SQLITE_BUSY` / `SQLITE_LOCKED`. The fix wraps the +commit paths in a **bounded retry-with-backoff** keyed on a **typed** error-code +classifier — retry, not a wider mutex, is the right tool because the contention +is *between* connections/processes, which no single in-process lock can serialize. + +The wrapper is applied to the three outbox/outcome commit paths: + +| Method | Role | +| --- | --- | +| `finish_effect` | Closes an outbox row with its `EffectResult` (including the benign no-op). | +| `release_effect_for_retry` | Returns a transiently-failed effect to the queue. | +| `recover_expired_effects` | Startup/periodic recovery that re-queues effects whose lease expired — the dominant source of the 9 startup collisions. | + +### `is_busy_locked` classifier + +Mirrors the existing `is_constraint` classifier and matches **only** the two +contention codes — never on error-message text, so no log/error string can steer +retry behaviour: + +```rust +// src/typed_ooda/ledger.rs + +fn is_busy_locked(error: &rusqlite::Error) -> bool { + matches!( + error, + rusqlite::Error::SqliteFailure(inner, _) + if inner.code == rusqlite::ErrorCode::DatabaseBusy + || inner.code == rusqlite::ErrorCode::DatabaseLocked + ) +} +``` + +Constraint violations, logic errors, and every other `rusqlite::Error` are **not** +retried — they surface immediately. + +Because a busy/locked error is mapped into a `CapabilityError` by `persistence` +before it reaches the retry loop (a write transaction calls many +`CapabilityResult`-returning helpers, not raw `rusqlite` calls), the typed +signal is preserved via a **marker stamped solely from the typed error code** — +never from arbitrary or injected message text. `persistence` prepends +`BUSY_PERSISTENCE_MARKER` iff `is_busy_locked` is true, and +`capability_error_is_busy` is the retry loop's predicate: + +```rust +// src/typed_ooda/ledger.rs + +const BUSY_PERSISTENCE_MARKER: &str = "[sqlite-busy] "; + +fn persistence(error: rusqlite::Error) -> CapabilityError { + // Marker derived purely from the typed rusqlite::ErrorCode, not message text. + let marker = if is_busy_locked(&error) { BUSY_PERSISTENCE_MARKER } else { "" }; + CapabilityError::new( + CapabilityErrorCode::PersistenceFailed, + format!("typed outcome persistence failed: {marker}{error}"), + ) +} + +fn capability_error_is_busy(error: &CapabilityError) -> bool { + error.code() == CapabilityErrorCode::PersistenceFailed + && error.to_string().contains(BUSY_PERSISTENCE_MARKER) +} +``` + +### `retry_on_busy` wrapper + +A small, bounded helper that re-runs a commit closure on `DatabaseBusy`/ +`DatabaseLocked` only. Each attempt runs the whole `begin → execute → commit` +closure, so a retry always uses a **fresh `IMMEDIATE` transaction** — never a +rolled-back or poisoned one. The closure returns `CapabilityResult` (the +write body's natural result type), and busy is recognised via +`capability_error_is_busy` above: + +```rust +// src/typed_ooda/ledger.rs + +/// Run `op` under bounded retry-with-backoff on SQLite busy/locked contention. +/// Any non-busy error, and busy/locked after `MAX_ATTEMPTS`, is returned as-is. +/// +/// - MAX_ATTEMPTS: 6 (hard cap; anti-DoS on the single Mutex). +/// - Backoff: exponential from ~10ms, capped at 400ms per sleep. +/// - Exhaustion: the final busy/locked error surfaces as PersistenceFailed — +/// never masked, never retried unbounded. +/// +/// INVARIANT: `op` must acquire (and drop) the `Mutex` guard +/// *itself*, once per invocation. The backoff `sleep` below runs strictly +/// between `op` calls, so the connection mutex is **released while sleeping** — +/// a retrying writer never blocks same-handler callers during its backoff. +fn retry_on_busy(mut op: impl FnMut() -> CapabilityResult) -> CapabilityResult { + const MAX_ATTEMPTS: u32 = 6; + const MAX_BACKOFF: Duration = Duration::from_millis(400); + let mut attempt = 0u32; + loop { + match op() { + Ok(value) => return Ok(value), + Err(error) if capability_error_is_busy(&error) && attempt + 1 < MAX_ATTEMPTS => { + let backoff = (Duration::from_millis(10) * (1u32 << attempt)).min(MAX_BACKOFF); + tracing::warn!( + target: "typed_ooda.outbox_write", + attempt = attempt + 1, + max_attempts = MAX_ATTEMPTS, + backoff_ms = backoff.as_millis() as u64, + "typed OODA outbox write contended (busy/locked); retrying", + ); + // Guard is already dropped here: `op` acquired and released the + // connection mutex within its own body, so this sleep holds no lock. + std::thread::sleep(backoff); + attempt += 1; + } + Err(error) => return Err(error), + } + } +} +``` + +**Guard-release invariant.** Every commit method currently opens with +`let mut connection = self.lock()?;` and holds that `MutexGuard` +for its whole body. The retry rewrite must move that acquisition *inside* the +`op` closure so the guard is created and dropped **per attempt**. Concretely, +each wrapped method becomes: + +```rust +// Representative: finish_effect (release_effect_for_retry and +// recover_expired_effects follow the same shape). The closure returns +// CapabilityResult, and every rusqlite error is mapped through `persistence` +// (which stamps the typed busy marker), so `retry_on_busy` sees busy contention +// regardless of which helper inside the transaction raised it. +retry_on_busy(|| { + // lock acquired at the START of each attempt ... + let mut connection = self.lock()?; + let transaction = connection + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(persistence)?; + // ... replay-dedup, write the outbox row / result ... + transaction.commit().map_err(persistence) + // ... guard dropped at the END of each attempt (Ok or Err), BEFORE any sleep. +}) +// retry_on_busy already returns CapabilityResult; exhaustion surfaces the busy +// error as CapabilityErrorCode::PersistenceFailed. No extra map_err is needed. +``` + +Because the guard lifetime is scoped to a single `op` call, `thread::sleep` +in `retry_on_busy` always runs with the connection mutex released. If the lock +were instead held across the loop, the backoff would block every other +same-handler caller for the full retry window — self-inflicted contention that +would defeat the fix. Lock poisoning is surfaced as `PersistenceFailed` (via +`self.lock()`'s existing mapping) and, being non-busy, is **never** retried. + +On exhaustion the caller maps the surfaced error to +`CapabilityErrorCode::PersistenceFailed` exactly as before — a genuinely stuck +writer is still reported, never swallowed. Retry counts are **not** part of any +outcome contract; tests assert on the **absence** of `database is locked`, not +on an exact attempt count, so they stay timing-robust. + +> **Latency interaction with `busy_timeout`.** Because every connection is +> opened with `busy_timeout(5s)`, each *individual* attempt can already block up +> to 5 s inside SQLite before it returns `SQLITE_BUSY`. The ≤400 ms backoff +> therefore does **not** bound total latency — the retry loop layers *on top of* +> the busy_timeout wait, so a pathologically contended write can take on the +> order of `MAX_ATTEMPTS × busy_timeout` in the worst case. This is acceptable +> for the outbox commit path (a stuck writer surfacing `PersistenceFailed` after +> several seconds is strictly better than the previous immediate failure), but it +> is the reason the retry bound is small and the backoff is capped: the goal is +> to ride out a *brief* cross-connection overlap, not to sit on a genuinely +> wedged database. + +### Connection PRAGMAs + +Every connection — including the recovery worker — opens through +`CapabilityHandler::open`, which applies `busy_timeout(5s)` and initializes the +schema, which sets `PRAGMA journal_mode = WAL`. WAL persists at the database-file +level; `busy_timeout` is re-applied on every open. No schema migration and no new +PRAGMA are introduced — the retry wrapper layers on top of the existing WAL + +busy_timeout baseline. + +## Observability + +All signals are structured `tracing` + OTel; there are no `print!`/`println!`/ +`eprintln!` calls in the touched paths. As part of this change the one remaining +`eprintln!` at the startup-recovery site in `typed_goal_session.rs` is converted +to a structured `tracing::warn!` (static message, `error = %error` field, no +payloads). + +| Event | Level | Key fields | Fires when | +| --- | --- | --- | --- | +| `typed goal-session effect no-op: goal completed or removed between prepare and dispatch` | `warn` | `effect_id`, `goal_id`, `reason="goal-removed-before-dispatch"` | A benign goal-race no-op is recorded at the dispatch site. | +| `typed OODA outbox write contended (busy/locked); retrying` | `warn` | `attempt`, `max_attempts`, `backoff_ms` | A SQLite write is retried after busy/locked contention. | +| Startup-recovery warning (converted from `eprintln!`) | `warn` | `error` | Startup outbox recovery reports a non-fatal error. | + +The benign no-op is also counted via `record_metric`: + +| Metric | Value | `context` | Meaning | +| --- | --- | --- | --- | +| `typed_ooda_effect_benign_no_op` | `1.0` per occurrence | the effect id | Number of goal-race no-ops absorbed instead of failing a cycle. | + +`record_metric` is best-effort (`let _ = …`): the **authoritative** audit signal +is the `tracing` event plus the finished outbox row, not the counter. The counter +lands in `/metrics/metrics.jsonl` — `` is +`SIMARD_STATE_ROOT` when set, else `$HOME/.simard` — as a standard `MetricEntry` +(JSON key `metric_name`); see [Metrics hygiene](./distill-raw-capture-on-parse-failure.md#metrics-hygiene) +for the envelope. + +## Configuration + +There is **nothing to configure**. Both behaviours are always-on, additive, and +self-contained: + +- The benign no-op reclassification has no toggle — a legitimately-removed goal + is always a no-op, never a failure. +- The retry wrapper's bounds are compile-time constants (`MAX_ATTEMPTS = 6`, + `MAX_BACKOFF = 400ms`). The backoff is capped well below the 5 s `busy_timeout` + so a retry re-probes the lock quickly once the timeout elapses; note the + attempts stack *on top of* the busy_timeout wait (see the latency note above), + so the bound is deliberately small. They are intentionally not env-tunable: an + operator knob here would be a foot-gun (unbounded retry re-introduces the DoS + risk the cap exists to prevent). + +The only operator-visible surface is the journal and the metrics file. + +## Verification + +Confirm both signatures are gone from a fresh window: + +```bash +# Should return nothing after the fix. +journalctl --user -u simard-ooda --since '-6h' \ + | grep -E 'goal disappeared before effect dispatch|typed outcome persistence failed: database is locked' +``` + +Confirm no-ops are being absorbed (rather than the goals failing): + +```bash +# Benign no-op events (structured tracing). +journalctl --user -u simard-ooda --since '-6h' \ + | grep 'typed goal-session effect no-op' + +# Benign no-op counter (path honors SIMARD_STATE_ROOT; default shown). +grep '"metric_name":"typed_ooda_effect_benign_no_op"' \ + "${SIMARD_STATE_ROOT:-$HOME/.simard}/metrics/metrics.jsonl" | tail +``` + +Regression tests (must FAIL on `main` before the fix, PASS after): + +| Test | Asserts | +| --- | --- | +| `dispatch_after_goal_removed_is_benign_no_op` | Goal removed between prepare and dispatch → cycle returns `Ok`, a `Succeeded { evidence: vec![] }` outbox row is written, the no-op tracing event is emitted, and **no** `DownstreamFailed` is produced. | +| `repo_mismatch_still_downstream_failed` | A repository mismatch still maps to `DownstreamFailed` — the reclassification did not widen. | +| `concurrent_outbox_write_under_lock_recovers` | Concurrent/startup-recovery writers against a shared temp DB all complete with **no** `database is locked` surfaced. Uses a barrier to force overlap; asserts on absence of the lock error, not on retry counts. | + +Run the targeted suites: + +```bash +cargo test -p simard typed_ooda:: -- --nocapture +cargo test -p simard advance_goal::typed_goal_session +cargo clippy --all-targets -- -D warnings +cargo fmt --check +``` + +## Security model + +- **Narrow reclassification.** `benign_no_op` covers **only** the two + pre-side-effect goal-not-found sites. It must **never** cover the + repository-mismatch, metadata, or post-spawn sites — doing so would create an + authorization-suppression primitive that could silently close an effect bound + to a different repository or orphan a live engineer process. The + `repo_mismatch_still_downstream_failed` negative test guards this boundary. +- **No_op is in-memory only.** The `no_op` flag is never serialized into an + outbox row and never read from persisted state or an external message, so it + cannot be forged to suppress a real failure. +- **Typed retry classification.** `is_busy_locked` matches on + `rusqlite::ErrorCode`, never on error-message text. Because the write bodies + return `CapabilityResult`, the busy signal crosses the `CapabilityError` + boundary via `BUSY_PERSISTENCE_MARKER`, which `persistence` stamps **solely** + from that typed code (checked by `capability_error_is_busy`). An injected + "database is locked" string in unrelated error text can never steer retry + behaviour. +- **Bounded, non-recursive retry.** The hard cap (6 attempts) and capped + backoff (≤400 ms) bound contention on the single `Mutex` — an + anti-DoS property — and exhaustion is surfaced as `PersistenceFailed`, never + swallowed. The backoff sleep runs with the connection mutex **released** + (the guard is scoped per attempt inside the `op` closure), so a retrying + writer cannot self-inflict a lock-hold stall on other same-handler callers. +- **Log hygiene.** New `tracing` events (including the `eprintln!` conversion) + log only `effect_id`, `goal_id`, a static reason, and numeric retry fields — + no payloads, no credential-bearing URLs, no tokens, no `{:?}` of opaque + structs. Evidence is kept empty (`vec![]`), so no attacker-influenced goal + metadata is persisted. +- **Best-effort metrics.** The counter is advisory (`let _ = …`); the durable + audit record is the finished outbox row plus the tracing event. +- **No silent fallbacks.** Every swallowed race produces a structured, counted + outcome; every retry exhaustion surfaces `PersistenceFailed`. + +## Examples + +### A goal completing mid-cycle no longer fails the cycle + +```text +# Before (DownstreamFailed — one wasted cycle per goal): +typed goal-session effect incomplete (DownstreamFailed): goal disappeared before effect dispatch + +# After (benign, counted no-op — the cycle succeeds): +WARN typed goal-session effect no-op: goal completed or removed between prepare and dispatch + effect_id=eff-9f8e7d goal_id=move-the-governed-repo-roster reason=goal-removed-before-dispatch +``` + +### Startup recovery no longer collides with live cycles + +```text +# Before (9 startup collisions in 6h): +typed OODA outbox startup recovery incomplete: typed outcome persistence failed: database is locked + +# After (bounded retry absorbs the contention, write completes): +WARN typed OODA outbox write contended (busy/locked); retrying attempt=1 max_attempts=6 backoff_ms=10 +# (no "database is locked" surfaced; recovery completes) +``` + +## When the no-op does *not* fire + +The no-op is intentionally narrow. The dispatch produces a real +`DownstreamFailed` (not a no-op) when: + +- The goal exists but the effect's repository **does not match** the + authenticated goal repository (repo/tenant isolation error). +- The goal's repository **metadata fails to resolve** (`goal_repository()` + error). +- The goal **already has an assigned engineer** (`already_assigned` → + "goal already has an assigned engineer") — a duplicate-spawn guard, not a + benign disappearance. +- The goal disappears **after** an engineer process was already spawned + (post-spawn miss) — a real failure must be recorded so recovery can reconcile + the orphaned process. +- The effect fails for any reason **other** than a legitimately-removed goal + (retryable failures still release for retry; other permanent failures still + map to `DownstreamFailed`). + +This keeps the no-op scoped to exactly the benign prepare→dispatch race it +exists to absorb. + +## Related + +- [Typed-OODA architecture](../architecture/typed-ooda-loop.md) — the surrounding + `decide → act` pipeline. +- [No-Bridge naming guard](./no-bridge-naming-guard.md) — the naming policy this + change complies with. +- [Cognitive-memory durability](../operations/cognitive-memory-durability.md) — + the sibling SQLite-durability posture this mirrors. +- [Distill raw-capture — metrics hygiene](./distill-raw-capture-on-parse-failure.md#metrics-hygiene) + — the `metrics.jsonl` / `MetricEntry` envelope the benign-no-op counter uses. diff --git a/src/ooda_actions/advance_goal/typed_goal_session.rs b/src/ooda_actions/advance_goal/typed_goal_session.rs index 5adaa6635..0aad1a03c 100644 --- a/src/ooda_actions/advance_goal/typed_goal_session.rs +++ b/src/ooda_actions/advance_goal/typed_goal_session.rs @@ -148,7 +148,11 @@ 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::warn!( + target: "typed_ooda.outbox_recovery", + error = %error, + "typed OODA outbox startup recovery incomplete", + ); } let execution = match route.execute( repo_root, @@ -334,7 +338,13 @@ impl LiveGoalSessionEffects<'_, '_> { .active .iter() .find(|goal| goal.id == goal_id) - .ok_or_else(|| EffectExecutionError::permanent("goal disappeared before spawn"))?; + .ok_or_else(|| { + // Benign goal-lifecycle race (issue #4468): the goal was + // legitimately completed/removed between preparing this + // effect and dispatching it, BEFORE any side effect ran. + // Report a counted no-op, not a DownstreamFailed cycle. + EffectExecutionError::benign_no_op("goal disappeared before spawn") + })?; (goal.repo.clone(), goal.assigned_to.is_some()) }; if already_assigned { @@ -457,7 +467,13 @@ impl LiveGoalSessionEffects<'_, '_> { .iter() .find(|goal| goal.id == goal_id) .ok_or_else(|| { - EffectExecutionError::permanent("goal disappeared before effect dispatch") + // Benign goal-lifecycle race (issue #4468): the goal was + // legitimately completed/removed between preparing this + // effect and dispatching it. This is the named defect + // signature ("goal disappeared before effect dispatch"). + // It occurs before any side effect, so it is a benign, + // counted no-op rather than a DownstreamFailed failure. + EffectExecutionError::benign_no_op("goal disappeared before effect dispatch") })?; goal_repository(goal).map_err(EffectExecutionError::permanent)? }; diff --git a/src/typed_ooda/executor.rs b/src/typed_ooda/executor.rs index 31968b8ba..571c33eba 100644 --- a/src/typed_ooda/executor.rs +++ b/src/typed_ooda/executor.rs @@ -93,6 +93,15 @@ impl From for RecipeProcessError { pub struct EffectExecutionError { message: String, permanent: bool, + /// When set, the effect could not run because the goal record was + /// legitimately completed/removed (or otherwise concurrently mutated out + /// from under the prepared effect) between prepare and dispatch. This is a + /// benign race, not a failure: the dispatcher closes the outbox row as a + /// counted, structured no-op instead of mapping it to + /// `CycleErrorCode::DownstreamFailed`. A benign no-op is also `permanent` + /// so it can never be mistaken for a retryable failure if this flag is ever + /// dropped on the floor. + no_op: bool, } impl EffectExecutionError { @@ -100,6 +109,7 @@ impl EffectExecutionError { Self { message: message.into(), permanent: true, + no_op: false, } } @@ -107,6 +117,20 @@ impl EffectExecutionError { Self { message: message.into(), permanent: false, + no_op: false, + } + } + + /// A benign goal-lifecycle race: the goal was legitimately + /// completed/removed between preparing this effect and dispatching it, so + /// there is nothing left to do. Structurally `permanent` (never retried) + /// and flagged `no_op` so the dispatcher records a counted no-op outcome + /// rather than a `DownstreamFailed` cycle error. + pub fn benign_no_op(message: impl Into) -> Self { + Self { + message: message.into(), + permanent: true, + no_op: true, } } } @@ -591,6 +615,43 @@ impl<'a> OutboxWorker<'a> { let result = match self.effects.execute(&job) { Ok(result) => result, Err(error) => { + if error.no_op { + // Benign goal-lifecycle race (issue #4468): the goal was + // legitimately completed/removed between preparing this + // effect and dispatching it. Do NOT map this to + // DownstreamFailed. Close the outbox row as a succeeded + // no-op (empty evidence) so it is never redispatched, emit + // a structured tracing event, and increment a counter so + // the race stays observable rather than silently swallowed. + tracing::warn!( + target: "typed_ooda.effect_dispatch", + effect_id = %job.effect_id, + outcome_id = %job.outcome_id, + goal_id = %job.goal_id, + reason = %error, + "typed goal-session effect skipped as benign no-op: goal completed or removed between prepare and dispatch", + ); + let _ = crate::self_metrics::record_metric( + "typed_ooda_effect_benign_no_op", + 1.0, + &format!( + "goal={};effect={};outcome={};reason={}", + job.goal_id, job.effect_id, job.outcome_id, error + ), + ); + let result = EffectResult::Succeeded { evidence: vec![] }; + self.handler + .finish_effect( + &job, + &effect_mutation_request_id(&job, "noop"), + SystemTime::now(), + &result, + ) + .map_err(|failure| { + CycleError::new(CycleErrorCode::PersistenceFailed, failure.to_string()) + })?; + return Ok(()); + } if !error.permanent { self.handler .release_effect_for_retry( @@ -672,6 +733,11 @@ mod tests { Succeed, Permanent, Retryable, + // Models the systemic race this fix targets: between preparing a + // goal-session effect and dispatching it, the goal record was + // legitimately completed/removed, so the executor reports a benign, + // structured no-op instead of a terminal failure. + BenignNoOp, } struct FakeEffects { @@ -701,6 +767,9 @@ mod tests { }), FakeMode::Permanent => Err(EffectExecutionError::permanent("permanent boom")), FakeMode::Retryable => Err(EffectExecutionError::retryable("transient boom")), + FakeMode::BenignNoOp => Err(EffectExecutionError::benign_no_op( + "goal disappeared before effect dispatch", + )), } } } @@ -1061,6 +1130,109 @@ mod tests { assert_eq!(job.state.as_str(), "failed"); } + // --------------------------------------------------------------------- + // Regression: benign goal-removed race at effect dispatch. + // + // The systemic defect (issue #4468): a prepared goal-session effect is + // dispatched after the goal record was legitimately completed/removed + // between prepare and dispatch. Today the executor surfaces this as + // EffectExecutionError::permanent("goal disappeared before effect + // dispatch"), which maps to CycleErrorCode::DownstreamFailed and fails + // the OODA cycle. It must instead be a benign, structured, counted no-op: + // the outbox row is closed as succeeded (never redispatched) and the + // cycle completes with Ok(()). + // --------------------------------------------------------------------- + + #[test] + fn benign_no_op_constructor_is_permanent_and_flagged() { + // Defense-in-depth: a benign no-op is still `permanent` (so it can + // never be mistaken for a retryable failure) AND carries the explicit + // `no_op` discriminator that routes it to the counted-no-op arm. + let error = EffectExecutionError::benign_no_op("goal disappeared before effect dispatch"); + assert!( + error.permanent, + "a benign no-op must remain permanent as a safety net" + ); + assert!( + error.no_op, + "a benign no-op must set the no_op discriminator" + ); + assert_eq!(error.to_string(), "goal disappeared before effect dispatch"); + + // The existing constructors must NOT set the no_op flag, or a real + // failure could be silently swallowed as a success. + assert!(!EffectExecutionError::permanent("boom").no_op); + assert!(!EffectExecutionError::retryable("later").no_op); + } + + #[test] + fn dispatch_after_goal_removed_is_benign_no_op() { + let handler = handler(); + let outcome = record_pending_action(&handler, file_issue_action()); + let effects = FakeEffects::new(FakeMode::BenignNoOp); + let worker = OutboxWorker::new(&handler, &effects, "test-worker", Duration::from_secs(60)); + + // The goal was legitimately removed between prepare and dispatch. This + // MUST complete the cycle successfully, not raise DownstreamFailed. + worker + .dispatch_outcome(&outcome) + .expect("a benign goal-removed race must complete as a no-op, not DownstreamFailed"); + assert_eq!( + effects.calls(), + 1, + "the effect is attempted exactly once before the benign no-op" + ); + + // The outbox row is closed as succeeded so the effect is never + // redispatched by a later cycle or startup recovery. + let job = handler + .effect_for_outcome(&outcome.outcome_id) + .expect("query effect") + .expect("effect"); + assert_eq!( + job.state.as_str(), + "succeeded", + "a benign no-op must close the outbox row, not leave it pending/failed" + ); + + // Idempotent redispatch must observe the closed row and must not + // re-run the effect executor. + worker + .dispatch_outcome(&outcome) + .expect("idempotent redispatch of a closed benign no-op"); + assert_eq!( + effects.calls(), + 1, + "a benign no-op effect must never be re-run" + ); + } + + #[test] + fn permanent_effect_failure_is_not_treated_as_benign_no_op() { + // Negative guard for the no_op reordering risk: a genuine permanent + // failure (no no_op flag) must still fail the cycle with + // DownstreamFailed and record the effect as `failed` — never swallowed + // into a success by the benign-no-op arm. + let handler = handler(); + let outcome = record_pending_action(&handler, file_issue_action()); + let effects = FakeEffects::new(FakeMode::Permanent); + let worker = OutboxWorker::new(&handler, &effects, "test-worker", Duration::from_secs(60)); + let error = worker + .dispatch_outcome(&outcome) + .expect_err("a real permanent failure must surface as DownstreamFailed"); + assert_eq!(error.code(), CycleErrorCode::DownstreamFailed); + let job = handler + .effect_for_outcome(&outcome.outcome_id) + .expect("query effect") + .expect("effect"); + assert_ne!( + job.state.as_str(), + "succeeded", + "a permanent failure must never be closed as succeeded" + ); + assert_eq!(job.state.as_str(), "failed"); + } + #[test] fn outbox_worker_blocks_privileged_effects_without_approval() { let handler = handler(); diff --git a/src/typed_ooda/ledger.rs b/src/typed_ooda/ledger.rs index cf677f2f8..c6b69e0d0 100644 --- a/src/typed_ooda/ledger.rs +++ b/src/typed_ooda/ledger.rs @@ -1250,32 +1250,38 @@ impl CapabilityHandler { let now = system_time_millis(now)?; 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)?; - if let Some(existing) = - replay_request(&transaction, request_id, "effect_recovery", &fingerprint)? - { - return Ok(existing); - } - let recovered = transaction - .execute( - "UPDATE effect_jobs - SET state='indeterminate', error='effect lease expired; execution outcome is unknown' - WHERE state='running' AND lease_expires_at <= ?1", - [now], - ) - .map_err(persistence)?; - record_request( - &transaction, - request_id, - "effect_recovery", - &fingerprint, - &recovered, - )?; - transaction.commit().map_err(persistence)?; - Ok(recovered) + // Serialized under bounded busy/locked retry (issue #4468): startup + // recovery is the dominant source of "database is locked" collisions + // with live cycles. The lock is acquired *inside* the closure so the + // guard is released across each backoff sleep. + retry_on_busy(|| { + let mut connection = self.lock()?; + let transaction = connection + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(persistence)?; + if let Some(existing) = + replay_request(&transaction, request_id, "effect_recovery", &fingerprint)? + { + return Ok(existing); + } + let recovered = transaction + .execute( + "UPDATE effect_jobs + SET state='indeterminate', error='effect lease expired; execution outcome is unknown' + WHERE state='running' AND lease_expires_at <= ?1", + [now], + ) + .map_err(persistence)?; + record_request( + &transaction, + request_id, + "effect_recovery", + &fingerprint, + &recovered, + )?; + transaction.commit().map_err(persistence)?; + Ok(recovered) + }) } pub fn renew_effect( @@ -1466,40 +1472,44 @@ impl CapabilityHandler { error, ), )?; - let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; - if replay_request::( - &transaction, - request_id, - "effect_retry", - &fingerprint, - )? - .is_some() - { - return Ok(()); - } - let changed = transaction - .execute( - "UPDATE effect_jobs - SET state='pending', error=?2, lease_owner=NULL, lease_expires_at=NULL - WHERE effect_id=?1 AND state='running' AND lease_owner=?3 - AND lease_generation=?4 AND lease_expires_at>?5", - params![lease.effect_id, error, owner, lease.lease_generation, now], - ) - .map_err(persistence)?; - if changed != 1 { - return Err(stale_lease(&lease.effect_id)); - } - record_request( - &transaction, - request_id, - "effect_retry", - &fingerprint, - &serde_json::Value::Null, - )?; - transaction.commit().map_err(persistence) + // Serialized under bounded busy/locked retry (issue #4468). Lock is + // acquired inside the closure so the guard is released across backoff. + retry_on_busy(|| { + let mut connection = self.lock()?; + let transaction = connection + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(persistence)?; + if replay_request::( + &transaction, + request_id, + "effect_retry", + &fingerprint, + )? + .is_some() + { + return Ok(()); + } + let changed = transaction + .execute( + "UPDATE effect_jobs + SET state='pending', error=?2, lease_owner=NULL, lease_expires_at=NULL + WHERE effect_id=?1 AND state='running' AND lease_owner=?3 + AND lease_generation=?4 AND lease_expires_at>?5", + params![lease.effect_id, error, owner, lease.lease_generation, now], + ) + .map_err(persistence)?; + if changed != 1 { + return Err(stale_lease(&lease.effect_id)); + } + record_request( + &transaction, + request_id, + "effect_retry", + &fingerprint, + &serde_json::Value::Null, + )?; + transaction.commit().map_err(persistence) + }) } pub fn finish_effect( @@ -1528,43 +1538,52 @@ impl CapabilityHandler { EffectResult::Failed { error } => ("failed", Some(error.as_str())), }; 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)?; - if replay_request::(&transaction, request_id, "effect_finish", &fingerprint)? + // Serialized under bounded busy/locked retry (issue #4468). Lock is + // acquired inside the closure so the guard is released across backoff. + retry_on_busy(|| { + let mut connection = self.lock()?; + let transaction = connection + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(persistence)?; + if replay_request::( + &transaction, + request_id, + "effect_finish", + &fingerprint, + )? .is_some() - { - return Ok(()); - } - let changed = transaction - .execute( - "UPDATE effect_jobs - SET state=?2, error=?3, result_json=?4, lease_owner=NULL, lease_expires_at=NULL - WHERE effect_id=?1 AND state='running' AND lease_owner=?5 - AND lease_generation=?6 AND lease_expires_at>?7", - params![ - lease.effect_id, - state, - error, - result_json, - owner, - lease.lease_generation, - now - ], - ) - .map_err(persistence)?; - if changed != 1 { - return Err(stale_lease(&lease.effect_id)); - } - record_request( - &transaction, - request_id, - "effect_finish", - &fingerprint, - result, - )?; - transaction.commit().map_err(persistence) + { + return Ok(()); + } + let changed = transaction + .execute( + "UPDATE effect_jobs + SET state=?2, error=?3, result_json=?4, lease_owner=NULL, lease_expires_at=NULL + WHERE effect_id=?1 AND state='running' AND lease_owner=?5 + AND lease_generation=?6 AND lease_expires_at>?7", + params![ + lease.effect_id, + state, + error, + result_json, + owner, + lease.lease_generation, + now + ], + ) + .map_err(persistence)?; + if changed != 1 { + return Err(stale_lease(&lease.effect_id)); + } + record_request( + &transaction, + request_id, + "effect_finish", + &fingerprint, + result, + )?; + transaction.commit().map_err(persistence) + }) } fn commit_terminal( @@ -2618,10 +2637,92 @@ fn system_time_millis(time: SystemTime) -> CapabilityResult { }) } +/// Stable marker that [`persistence`] stamps onto a `CapabilityError` whose +/// underlying `rusqlite::Error` was a *typed* SQLITE_BUSY / SQLITE_LOCKED +/// contention error. [`retry_on_busy`] keys its retry decision on this marker, +/// which is set **solely** from the typed error code — never from arbitrary or +/// caller-supplied message text — so no log line or injected payload string can +/// steer retry behaviour. +const BUSY_PERSISTENCE_MARKER: &str = "[sqlite-busy] "; + +/// True iff `error` is a transient SQLite lock-contention error +/// (SQLITE_BUSY / SQLITE_LOCKED) that is safe to retry. Mirrors the existing +/// `is_constraint` classifier: detected on the typed `rusqlite::ErrorCode`, +/// never on message text, so it is robust across locales and immune to +/// error-string steering. Constraint/logic errors are deliberately NOT treated +/// as busy — they must surface, never retry-loop. +fn is_busy_locked(error: &rusqlite::Error) -> bool { + matches!( + error, + rusqlite::Error::SqliteFailure(inner, _) + if inner.code == rusqlite::ErrorCode::DatabaseBusy + || inner.code == rusqlite::ErrorCode::DatabaseLocked + ) +} + +/// True iff `error` is a persistence failure that originated from a typed +/// SQLITE_BUSY / SQLITE_LOCKED contention error (as stamped by [`persistence`] +/// via [`BUSY_PERSISTENCE_MARKER`]). This is how [`retry_on_busy`] recognises a +/// retryable write without re-plumbing the raw `rusqlite::Error` through the +/// many `CapabilityResult`-returning helpers a write transaction calls. +fn capability_error_is_busy(error: &CapabilityError) -> bool { + error.code() == CapabilityErrorCode::PersistenceFailed + && error.to_string().contains(BUSY_PERSISTENCE_MARKER) +} + +/// Run `op` under bounded retry-with-backoff on SQLite busy/locked contention +/// (issue #4468). Any non-busy error — and busy/locked after `MAX_ATTEMPTS` — +/// is returned as-is, never masked and never retried unbounded. +/// +/// - `MAX_ATTEMPTS`: 6 (hard cap; anti-DoS on the single `Mutex`). +/// - Backoff: exponential from ~10ms, capped at 400ms per sleep. +/// - Exhaustion: the final busy/locked error surfaces as `PersistenceFailed` — +/// a genuinely wedged writer is still reported, never swallowed. +/// +/// INVARIANT: `op` must acquire (and drop) the `Mutex` guard +/// *itself*, once per invocation, so the backoff `sleep` below runs strictly +/// between `op` calls with the connection mutex **released** — a retrying +/// writer never blocks same-handler callers during its backoff. +fn retry_on_busy(mut op: impl FnMut() -> CapabilityResult) -> CapabilityResult { + const MAX_ATTEMPTS: u32 = 6; + const MAX_BACKOFF: Duration = Duration::from_millis(400); + let mut attempt = 0u32; + loop { + match op() { + Ok(value) => return Ok(value), + Err(error) if capability_error_is_busy(&error) && attempt + 1 < MAX_ATTEMPTS => { + let backoff = (Duration::from_millis(10) * (1u32 << attempt)).min(MAX_BACKOFF); + tracing::warn!( + target: "typed_ooda.outbox_write", + attempt = attempt + 1, + max_attempts = MAX_ATTEMPTS, + backoff_ms = backoff.as_millis() as u64, + "typed OODA outbox write contended (busy/locked); retrying", + ); + // The guard is already dropped here: `op` acquired and released + // the connection mutex within its own body, so this sleep holds + // no lock. + std::thread::sleep(backoff); + attempt += 1; + } + Err(error) => return Err(error), + } + } +} + fn persistence(error: rusqlite::Error) -> CapabilityError { + // Stamp a typed-only busy marker so `retry_on_busy` can recognise a + // retryable contention error after it has been mapped into a + // `CapabilityError`. The marker is derived purely from the typed + // `rusqlite::ErrorCode` (via `is_busy_locked`), not from message text. + let marker = if is_busy_locked(&error) { + BUSY_PERSISTENCE_MARKER + } else { + "" + }; CapabilityError::new( CapabilityErrorCode::PersistenceFailed, - format!("typed outcome persistence failed: {error}"), + format!("typed outcome persistence failed: {marker}{error}"), ) } @@ -3519,3 +3620,181 @@ mod actor_session_scope_tests { } } } + +#[cfg(test)] +mod outbox_serialization_tests { + use super::*; + use std::sync::{Arc, Barrier}; + use std::thread; + + const REPO_OWNER: &str = "rysweet"; + const REPO_NAME: &str = "Simard"; + const POLICY_REVISION: &str = "policy-v1"; + + fn admission() -> AdmissionSnapshot { + AdmissionSnapshot { + concurrent_engineers: 0, + disk_used_percent: 1, + active_claims: BTreeSet::new(), + policy_revision: POLICY_REVISION.to_string(), + } + } + + fn file_issue_actor( + session_id: &str, + cycle_id: &str, + goal_id: &str, + ) -> AuthenticatedToolContext { + AuthenticatedToolContext::new( + "goal-session-actor", + session_id, + [CapabilityGrant::RecordAction(ActionKind::FileIssue)], + ) + .scoped_to_repository(RepositoryRef::new(REPO_OWNER, REPO_NAME)) + .bound_to_cycle_goal(cycle_id, goal_id) + } + + fn file_issue_request( + request_id: &str, + session_id: &str, + cycle_id: &str, + goal_id: &str, + ) -> RecordActionRequest { + RecordActionRequest { + identity: TerminalRequestIdentity::new(request_id, session_id, cycle_id, goal_id), + action: Action::FileIssue(FileIssueAction { + repository: RepositoryRef::new(REPO_OWNER, REPO_NAME), + title: OpaqueBytes::from(b"a real issue title".to_vec()), + body: OpaqueBytes::from(b"body".to_vec()), + labels: Vec::new(), + }), + raw_semantic: OpaqueBytes::from(b"raw".to_vec()), + evidence: Vec::new(), + } + } + + fn is_database_locked(error: &CapabilityError) -> bool { + error + .to_string() + .to_ascii_lowercase() + .contains("database is locked") + } + + // --------------------------------------------------------------------- + // Regression: the typed-OODA outbox/outcome store must serialize writes so + // that concurrent cycles and startup recovery stop colliding on + // 'database is locked' (issue #4468). + // + // Reproduces the production shape: MANY independent writers (each with its + // own connection to the SAME sqlite file, as separate processes/handlers + // do) commit outbox + outcome rows in Immediate transactions while a + // startup-recovery sweep runs concurrently. Under the pre-fix + // configuration this surfaces SQLITE_BUSY / "database is locked". After the + // fix (WAL + busy_timeout on every connection + bounded busy retry) every + // write must succeed and NO 'database is locked' error may be observed. + // --------------------------------------------------------------------- + + #[test] + fn concurrent_outbox_writes_never_surface_database_is_locked() { + let dir = tempfile::tempdir().expect("tempdir"); + let db_path: PathBuf = dir.path().join("outcomes.sqlite3"); + + const WRITERS: usize = 8; + const ITERATIONS: usize = 60; + + // +1 participant for the concurrent startup-recovery sweeper so every + // writer overlaps with recovery contention on the same file. + let barrier = Arc::new(Barrier::new(WRITERS + 1)); + + let mut threads = Vec::new(); + for writer in 0..WRITERS { + let path = db_path.clone(); + let barrier = Arc::clone(&barrier); + threads.push(thread::spawn(move || -> Vec { + let handler = + CapabilityHandler::open(&path, CapabilityPolicy::new(POLICY_REVISION)) + .expect("open writer handler"); + let mut locked_errors = Vec::new(); + barrier.wait(); + for iteration in 0..ITERATIONS { + let session = format!("session-w{writer}-{iteration}"); + let cycle = format!("cycle-w{writer}-{iteration}"); + let goal = format!("goal-w{writer}-{iteration}"); + let request = format!("req-w{writer}-{iteration}"); + let actor = file_issue_actor(&session, &cycle, &goal); + let req = file_issue_request(&request, &session, &cycle, &goal); + match handler.record_action(&actor, req, &admission()) { + Ok(_) => {} + Err(error) if is_database_locked(&error) => { + locked_errors + .push(format!("record_action w{writer}/{iteration}: {error}")); + } + Err(error) => { + panic!("unexpected non-lock write failure: {error:?}"); + } + } + } + locked_errors + })); + } + + // Concurrent startup-recovery sweeper: a separate connection running + // the outbox recovery path repeatedly while the writers commit. + { + let path = db_path.clone(); + let barrier = Arc::clone(&barrier); + threads.push(thread::spawn(move || -> Vec { + let handler = + CapabilityHandler::open(&path, CapabilityPolicy::new(POLICY_REVISION)) + .expect("open recovery handler"); + let mut locked_errors = Vec::new(); + barrier.wait(); + for sweep in 0..(WRITERS * ITERATIONS / 4) { + let request_id = format!("recover-{sweep}"); + match handler.recover_expired_effects(&request_id, SystemTime::now()) { + Ok(_) => {} + Err(error) if is_database_locked(&error) => { + locked_errors.push(format!("recover {sweep}: {error}")); + } + Err(error) => { + panic!("unexpected non-lock recovery failure: {error:?}"); + } + } + } + locked_errors + })); + } + + let mut locked_errors = Vec::new(); + for thread in threads { + locked_errors.extend(thread.join().expect("writer thread must not panic")); + } + + assert!( + locked_errors.is_empty(), + "concurrent outbox writers and startup recovery must never surface \ + 'database is locked'; got {} occurrence(s): {:#?}", + locked_errors.len(), + locked_errors, + ); + + // Every terminal must be durably persisted: no write was silently lost + // to a swallowed lock error. + let verifier = CapabilityHandler::open(&db_path, CapabilityPolicy::new(POLICY_REVISION)) + .expect("open verifier handler"); + for writer in 0..WRITERS { + for iteration in 0..ITERATIONS { + let session = format!("session-w{writer}-{iteration}"); + let cycle = format!("cycle-w{writer}-{iteration}"); + let terminal = verifier + .terminal_for_cycle(&session, &cycle) + .expect("query terminal"); + assert!( + terminal.is_some(), + "terminal for {session}/{cycle} must be durably persisted after \ + concurrent writes" + ); + } + } + } +} From 28d479f0ec73cc683ef7d81714168cd5f456898d Mon Sep 17 00:00:00 2001 From: rysweet Date: Wed, 22 Jul 2026 21:13:22 +0000 Subject: [PATCH 2/5] refactor(typed-ooda): extract benign no-op finish into named helper Step 9 refactor pass for #4468. Extract the ~35-line inline benign goal-race branch in OutboxWorker::execute_claimed into a named private helper finish_effect_as_benign_no_op. Behavior-identical: same tracing event, counter, and Succeeded{evidence:vec![]} outbox close. Shrinks the long dispatch function and names the concept at the call site. No schema, semantic, or happy-path change. fmt/clippy clean; 54 typed_ooda tests green. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/typed_ooda/executor.rs | 77 +++++++++++++++++++++----------------- 1 file changed, 42 insertions(+), 35 deletions(-) diff --git a/src/typed_ooda/executor.rs b/src/typed_ooda/executor.rs index 571c33eba..1eb132149 100644 --- a/src/typed_ooda/executor.rs +++ b/src/typed_ooda/executor.rs @@ -616,41 +616,7 @@ impl<'a> OutboxWorker<'a> { Ok(result) => result, Err(error) => { if error.no_op { - // Benign goal-lifecycle race (issue #4468): the goal was - // legitimately completed/removed between preparing this - // effect and dispatching it. Do NOT map this to - // DownstreamFailed. Close the outbox row as a succeeded - // no-op (empty evidence) so it is never redispatched, emit - // a structured tracing event, and increment a counter so - // the race stays observable rather than silently swallowed. - tracing::warn!( - target: "typed_ooda.effect_dispatch", - effect_id = %job.effect_id, - outcome_id = %job.outcome_id, - goal_id = %job.goal_id, - reason = %error, - "typed goal-session effect skipped as benign no-op: goal completed or removed between prepare and dispatch", - ); - let _ = crate::self_metrics::record_metric( - "typed_ooda_effect_benign_no_op", - 1.0, - &format!( - "goal={};effect={};outcome={};reason={}", - job.goal_id, job.effect_id, job.outcome_id, error - ), - ); - let result = EffectResult::Succeeded { evidence: vec![] }; - self.handler - .finish_effect( - &job, - &effect_mutation_request_id(&job, "noop"), - SystemTime::now(), - &result, - ) - .map_err(|failure| { - CycleError::new(CycleErrorCode::PersistenceFailed, failure.to_string()) - })?; - return Ok(()); + return self.finish_effect_as_benign_no_op(&job, &error); } if !error.permanent { self.handler @@ -704,6 +670,47 @@ impl<'a> OutboxWorker<'a> { } } } + + /// Close the outbox row for a benign goal-lifecycle race (issue #4468): the + /// goal was legitimately completed/removed between preparing this effect and + /// dispatching it. Do NOT map this to `DownstreamFailed`. The row is closed + /// as a succeeded no-op (empty evidence) so it is never redispatched, a + /// structured tracing event is emitted, and a counter is incremented so the + /// race stays observable rather than silently swallowed. + fn finish_effect_as_benign_no_op( + &self, + job: &EffectJob, + error: &EffectExecutionError, + ) -> Result<(), CycleError> { + tracing::warn!( + target: "typed_ooda.effect_dispatch", + effect_id = %job.effect_id, + outcome_id = %job.outcome_id, + goal_id = %job.goal_id, + reason = %error, + "typed goal-session effect skipped as benign no-op: goal completed or removed between prepare and dispatch", + ); + let _ = crate::self_metrics::record_metric( + "typed_ooda_effect_benign_no_op", + 1.0, + &format!( + "goal={};effect={};outcome={};reason={}", + job.goal_id, job.effect_id, job.outcome_id, error + ), + ); + let result = EffectResult::Succeeded { evidence: vec![] }; + self.handler + .finish_effect( + job, + &effect_mutation_request_id(job, "noop"), + SystemTime::now(), + &result, + ) + .map_err(|failure| { + CycleError::new(CycleErrorCode::PersistenceFailed, failure.to_string()) + })?; + Ok(()) + } } fn effect_mutation_request_id(job: &EffectJob, operation: &str) -> String { From c34d360b1493f3e0b9886453c0434cba872d91f1 Mon Sep 17 00:00:00 2001 From: rysweet Date: Wed, 22 Jul 2026 21:18:42 +0000 Subject: [PATCH 3/5] perf(typed-ooda): avoid per-retry String alloc in busy classifier MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 9b perf pass for #4468. On the SQLite busy/locked contention path retry_on_busy calls capability_error_is_busy for every failed op, which allocated a throwaway String via error.to_string() only to substring-scan it for the typed busy marker. - Add CapabilityError::message() borrowing accessor (additive, pub(crate)). - Scan error.message() in place instead of to_string() — removes one heap allocation per contended attempt. - Short-circuit the cheap attempt-count check before the classifier so the exhausted-attempt path skips classification entirely. Behavior-identical: same retry decisions, same surfaced errors, no schema or semantic change. fmt + clippy + 54/54 typed_ooda tests green. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/typed_ooda/ledger.rs | 4 ++-- src/typed_ooda/types.rs | 7 +++++++ 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/src/typed_ooda/ledger.rs b/src/typed_ooda/ledger.rs index c6b69e0d0..69a80cdaa 100644 --- a/src/typed_ooda/ledger.rs +++ b/src/typed_ooda/ledger.rs @@ -2667,7 +2667,7 @@ fn is_busy_locked(error: &rusqlite::Error) -> bool { /// many `CapabilityResult`-returning helpers a write transaction calls. fn capability_error_is_busy(error: &CapabilityError) -> bool { error.code() == CapabilityErrorCode::PersistenceFailed - && error.to_string().contains(BUSY_PERSISTENCE_MARKER) + && error.message().contains(BUSY_PERSISTENCE_MARKER) } /// Run `op` under bounded retry-with-backoff on SQLite busy/locked contention @@ -2690,7 +2690,7 @@ fn retry_on_busy(mut op: impl FnMut() -> CapabilityResult) -> CapabilityRe loop { match op() { Ok(value) => return Ok(value), - Err(error) if capability_error_is_busy(&error) && attempt + 1 < MAX_ATTEMPTS => { + Err(error) if attempt + 1 < MAX_ATTEMPTS && capability_error_is_busy(&error) => { let backoff = (Duration::from_millis(10) * (1u32 << attempt)).min(MAX_BACKOFF); tracing::warn!( target: "typed_ooda.outbox_write", diff --git a/src/typed_ooda/types.rs b/src/typed_ooda/types.rs index f6b88275c..186af6ad9 100644 --- a/src/typed_ooda/types.rs +++ b/src/typed_ooda/types.rs @@ -549,6 +549,13 @@ impl CapabilityError { pub fn code(&self) -> CapabilityErrorCode { self.code } + + /// Borrow the error message without allocating. Used by the busy/locked + /// retry classifier to scan for the typed contention marker on the hot + /// contention path instead of formatting a throwaway `String` per attempt. + pub(crate) fn message(&self) -> &str { + &self.message + } } impl Display for CapabilityError { From 50e54078629dc7751d4dbb6859c2b65fce468829 Mon Sep 17 00:00:00 2001 From: rysweet Date: Wed, 22 Jul 2026 21:32:41 +0000 Subject: [PATCH 4/5] test(typed-ooda): direct unit coverage for busy/locked retry logic Step 10c philosophy compliance follow-up for #4468. The retry mechanism (is_busy_locked, capability_error_is_busy, retry_on_busy) was non-trivial new control flow with only indirect concurrency-test coverage. retry_on_busy takes an injectable closure, so its decision logic is deterministically unit-testable without provoking real contention. Add tests for: - typed contention classification (busy/locked yes; constraint/no-rows no) - marker-based classifier keys on the typed error code, not free text (a message that merely says 'database is locked' does not trigger retry) - retry_on_busy: immediate success, retry-then-succeed, bounded exhaustion surfacing PersistenceFailed (hard-capped at 6 attempts), and immediate passthrough for non-contention errors Closes the one Test-Driven gap found in the philosophy compliance check. No production code changed; test-only. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/typed_ooda/ledger.rs | 146 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 146 insertions(+) diff --git a/src/typed_ooda/ledger.rs b/src/typed_ooda/ledger.rs index 69a80cdaa..46e9ff2cb 100644 --- a/src/typed_ooda/ledger.rs +++ b/src/typed_ooda/ledger.rs @@ -3797,4 +3797,150 @@ mod outbox_serialization_tests { } } } + + // --------------------------------------------------------------------- + // Direct unit coverage for the SQLite busy/locked retry mechanism + // (issue #4468). `retry_on_busy` takes an injectable closure, so its + // control flow — retry on contention, bounded exhaustion, and immediate + // passthrough for non-contention errors — is unit-testable deterministically + // without provoking real cross-thread lock contention. The concurrency + // test above exercises the stack under load; these pin the decision logic. + // --------------------------------------------------------------------- + + fn busy_sqlite_error() -> rusqlite::Error { + rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::DatabaseBusy, + extended_code: 5, + }, + Some("database is locked".to_string()), + ) + } + + fn locked_sqlite_error() -> rusqlite::Error { + rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::DatabaseLocked, + extended_code: 6, + }, + Some("database table is locked".to_string()), + ) + } + + fn constraint_sqlite_error() -> rusqlite::Error { + rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error { + code: rusqlite::ErrorCode::ConstraintViolation, + extended_code: 19, + }, + Some("UNIQUE constraint failed".to_string()), + ) + } + + #[test] + fn is_busy_locked_matches_only_typed_contention_codes() { + assert!(is_busy_locked(&busy_sqlite_error())); + assert!(is_busy_locked(&locked_sqlite_error())); + assert!( + !is_busy_locked(&constraint_sqlite_error()), + "constraint violations must never be treated as retryable contention" + ); + assert!(!is_busy_locked(&rusqlite::Error::QueryReturnedNoRows)); + } + + #[test] + fn capability_error_is_busy_keys_on_typed_marker_not_free_text() { + // A persistence error mapped from a typed busy/locked rusqlite error is + // classified as retryable contention... + assert!(capability_error_is_busy(&persistence(busy_sqlite_error()))); + assert!(capability_error_is_busy( + &persistence(locked_sqlite_error()) + )); + + // ...but a persistence error from a non-contention cause is NOT, even + // though it shares the PersistenceFailed code. + assert!(!capability_error_is_busy(&persistence( + constraint_sqlite_error() + ))); + + // Free text that merely mentions locking cannot steer retries: only the + // marker stamped from the typed error code counts. + let steered = persistence_message("database is locked, please retry"); + assert!( + !capability_error_is_busy(&steered), + "message text alone must never classify an error as retryable" + ); + + // The marker on a non-PersistenceFailed code is likewise ignored. + let wrong_code = CapabilityError::new( + CapabilityErrorCode::InvalidArgument, + format!("{BUSY_PERSISTENCE_MARKER}spoofed"), + ); + assert!(!capability_error_is_busy(&wrong_code)); + } + + #[test] + fn retry_on_busy_returns_immediately_on_success() { + let mut calls = 0u32; + let result: CapabilityResult = retry_on_busy(|| { + calls += 1; + Ok(7) + }); + assert_eq!(result.expect("ok"), 7); + assert_eq!(calls, 1, "a successful op must not be retried"); + } + + #[test] + fn retry_on_busy_retries_transient_contention_then_succeeds() { + let mut calls = 0u32; + let result: CapabilityResult<&str> = retry_on_busy(|| { + calls += 1; + if calls < 3 { + Err(persistence(busy_sqlite_error())) + } else { + Ok("committed") + } + }); + assert_eq!(result.expect("eventually succeeds"), "committed"); + assert_eq!( + calls, 3, + "must retry through transient busy errors then commit" + ); + } + + #[test] + fn retry_on_busy_bounds_attempts_and_surfaces_the_contention_error() { + let mut calls = 0u32; + let result: CapabilityResult<()> = retry_on_busy(|| { + calls += 1; + Err(persistence(busy_sqlite_error())) + }); + let error = result.expect_err("a wedged writer must surface, never loop forever"); + assert_eq!(error.code(), CapabilityErrorCode::PersistenceFailed); + assert!( + capability_error_is_busy(&error), + "the exhausted error is still the original busy/locked failure" + ); + assert_eq!( + calls, 6, + "attempts are hard-capped at MAX_ATTEMPTS (anti-DoS on the shared connection)" + ); + } + + #[test] + fn retry_on_busy_does_not_retry_non_contention_errors() { + let mut calls = 0u32; + let result: CapabilityResult<()> = retry_on_busy(|| { + calls += 1; + Err(persistence(constraint_sqlite_error())) + }); + assert_eq!( + result.expect_err("non-busy error surfaces").code(), + CapabilityErrorCode::PersistenceFailed + ); + assert_eq!( + calls, 1, + "a non-busy error must surface immediately, never retry" + ); + } } From b4bdb6922711dae6408a8c5a285ba3d02dc3ea88 Mon Sep 17 00:00:00 2001 From: rysweet Date: Thu, 23 Jul 2026 00:21:51 +0000 Subject: [PATCH 5/5] fix(typed-ooda): make ledger open() lock-resilient + de-flake outbox concurrency test (#4468) The concurrency regression test deadlocked instead of passing: a prepared goal-session outbox test spawns 9 barrier participants (8 writers + 1 startup recovery sweeper), each opening its OWN CapabilityHandler on the SAME fresh sqlite file. CapabilityHandler::open runs schema-init WRITES (the WAL-mode switch and the CREATE-TABLE migration transaction), which contend under concurrent first-open. That write path was NOT covered by the new bounded busy/locked retry, so one open() hit a transient SQLITE_BUSY, panicked its thread before barrier.wait(), and starved the barrier -> permanent deadlock (all remaining threads parked in futex, 0 CPU; confirmed via gdb). Fix, additively and consistent with the existing #4468 retry strategy: - open(): wrap the idempotent schema::initialize in retry_on_busy so a transient lock during handler construction / startup recovery is retried with bounded backoff instead of failing the whole open. Same single-writer + busy_timeout + bounded-retry contract as the live write paths. - Add concurrent_open_of_shared_ledger_survives_contention: 12 handlers race open() on one fresh file, synchronized by a barrier that is waited on BEFORE open so an open failure surfaces as a joined Err, never a deadlock. - Harden concurrent_outbox_writes_never_surface_database_is_locked: construct all handlers before the barrier so an open error is a clean test failure, not a barrier starvation; keeps the test focused on concurrent writes + a concurrent recovery sweep. No happy-path behavior change; structured tracing preserved; no stray prints. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/typed_ooda/ledger.rs | 332 ++++++++++++++++++++++++++------------- 1 file changed, 226 insertions(+), 106 deletions(-) diff --git a/src/typed_ooda/ledger.rs b/src/typed_ooda/ledger.rs index 46e9ff2cb..19ab8029b 100644 --- a/src/typed_ooda/ledger.rs +++ b/src/typed_ooda/ledger.rs @@ -257,7 +257,17 @@ impl CapabilityHandler { connection .busy_timeout(Duration::from_secs(5)) .map_err(persistence)?; - super::schema::initialize(&mut connection, now_millis()).map_err(persistence)?; + // Schema initialization performs writes (the WAL-mode switch and the + // CREATE-TABLE migration transaction) that collide with concurrent + // first-opens and the startup-recovery sweep on the shared sqlite file + // (issue #4468). Serialize it under the same bounded busy/locked retry + // as the live write paths so a transient lock during handler + // construction is retried, not surfaced as a hard open failure. + // `initialize` is idempotent (version-guarded, CREATE ... IF NOT EXISTS, + // INSERT OR IGNORE), so re-running it on retry is safe. + retry_on_busy(|| { + super::schema::initialize(&mut connection, now_millis()).map_err(persistence) + })?; Ok(Self { connection: Mutex::new(connection), policy, @@ -616,32 +626,10 @@ 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)?; - if let Some(existing) = replay_request( - &transaction, - &request.identity.request_id, - "terminal", - &fingerprint, - )? { - return Ok(existing); - } - if let Action::SpawnEngineer(spawn) = &request.action { - let expected_claim = format!( - "{}/{}:{}", - spawn.repository.owner, spawn.repository.name, request.identity.goal_id - ); - if spawn.claim_key != expected_claim { - return Err(CapabilityError::new( - CapabilityErrorCode::InvalidArgument, - format!("engineer claim key must be {expected_claim:?}"), - )); - } - } - self.admit(&request.action, admission)?; - ensure_cycle_open(&transaction, &request.identity)?; + // Build the terminal outcome once, before the busy-retry loop, so the + // generated `outcome_id`/timestamp stay stable across retries (issue + // #4468): a retried write must re-commit the *same* outcome identity, + // never mint a fresh one. let RecordActionRequest { identity, action, @@ -656,28 +644,61 @@ impl CapabilityHandler { }); let outcome = self.new_outcome(actor, &identity, payload, raw_semantic, evidence); let outcome_json = serde_json::to_vec(&outcome).map_err(serialization)?; - insert_terminal(&transaction, &outcome, &fingerprint, &outcome_json)?; - record_request_json( - &transaction, - &outcome.request_id, - "terminal", - &fingerprint, - &outcome_json, - )?; let TypedOutcomePayload::Action(action_payload) = &outcome.payload else { unreachable!("record_action always creates an action payload"); }; - if let Action::SpawnEngineer(spawn) = &action_payload.action { - self.insert_engineer_claim( + let action = &action_payload.action; + + // Serialize the primary live-cycle write under bounded busy/locked + // retry (issue #4468): this path collides with concurrent cycles and + // startup recovery on the shared sqlite file. The lock is acquired + // *inside* the closure so the guard is released across each backoff + // sleep. `admit`/replay ordering is preserved (admit runs after the + // replay check, so an idempotent replay hit never re-admits). + retry_on_busy(|| { + let mut connection = self.lock()?; + let transaction = connection + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(persistence)?; + if let Some(existing) = + replay_request(&transaction, &identity.request_id, "terminal", &fingerprint)? + { + return Ok(existing); + } + if let Action::SpawnEngineer(spawn) = action { + let expected_claim = format!( + "{}/{}:{}", + spawn.repository.owner, spawn.repository.name, identity.goal_id + ); + if spawn.claim_key != expected_claim { + return Err(CapabilityError::new( + CapabilityErrorCode::InvalidArgument, + format!("engineer claim key must be {expected_claim:?}"), + )); + } + } + self.admit(action, admission)?; + ensure_cycle_open(&transaction, &identity)?; + insert_terminal(&transaction, &outcome, &fingerprint, &outcome_json)?; + record_request_json( &transaction, - &spawn.claim_key, - &outcome.outcome_id, &outcome.request_id, + "terminal", + &fingerprint, + &outcome_json, )?; - } - insert_effect(&transaction, &outcome, &action_payload.action)?; - transaction.commit().map_err(persistence)?; - Ok(outcome) + if let Action::SpawnEngineer(spawn) = action { + self.insert_engineer_claim( + &transaction, + &spawn.claim_key, + &outcome.outcome_id, + &outcome.request_id, + )?; + } + insert_effect(&transaction, &outcome, action)?; + transaction.commit().map_err(persistence)?; + Ok(outcome.clone()) + }) } pub fn record_no_action( @@ -1206,39 +1227,46 @@ impl CapabilityHandler { lease_millis, ), )?; - let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; - if let Some(existing) = - replay_request(&transaction, request_id, "effect_claim", &fingerprint)? - { - return Ok(existing); - } - let claimed = query_effect( - &transaction, - " - UPDATE effect_jobs - SET state='running', attempt=attempt+1, lease_generation=lease_generation+1, - lease_owner=?2, lease_expires_at=?3 - WHERE outcome_id=?1 AND state='pending' - RETURNING effect_id, outcome_id, request_id, kind, state, action_json, attempt, - lease_generation, lease_owner, lease_expires_at, error, result_json, - (SELECT decision_json FROM authorization_decisions - WHERE effect_id=effect_jobs.effect_id AND decision='approved' - ORDER BY recorded_at DESC, rowid DESC LIMIT 1) - ", - params![outcome_id, worker, now.saturating_add(lease_millis)], - )?; - record_request( - &transaction, - request_id, - "effect_claim", - &fingerprint, - &claimed, - )?; - transaction.commit().map_err(persistence)?; - Ok(claimed) + // Serialize the effect claim under bounded busy/locked retry (issue + // #4468): claiming races with concurrent cycles and the startup + // recovery sweep on the shared sqlite file. The lock is acquired + // *inside* the closure so the guard is released across each backoff + // sleep. + retry_on_busy(|| { + let mut connection = self.lock()?; + let transaction = connection + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(persistence)?; + if let Some(existing) = + replay_request(&transaction, request_id, "effect_claim", &fingerprint)? + { + return Ok(existing); + } + let claimed = query_effect( + &transaction, + " + UPDATE effect_jobs + SET state='running', attempt=attempt+1, lease_generation=lease_generation+1, + lease_owner=?2, lease_expires_at=?3 + WHERE outcome_id=?1 AND state='pending' + RETURNING effect_id, outcome_id, request_id, kind, state, action_json, attempt, + lease_generation, lease_owner, lease_expires_at, error, result_json, + (SELECT decision_json FROM authorization_decisions + WHERE effect_id=effect_jobs.effect_id AND decision='approved' + ORDER BY recorded_at DESC, rowid DESC LIMIT 1) + ", + params![outcome_id, worker, now.saturating_add(lease_millis)], + )?; + record_request( + &transaction, + request_id, + "effect_claim", + &fingerprint, + &claimed, + )?; + transaction.commit().map_err(persistence)?; + Ok(claimed) + }) } pub fn recover_expired_effects( @@ -1595,28 +1623,38 @@ impl CapabilityHandler { evidence: Vec, fingerprint: String, ) -> CapabilityResult { - let mut connection = self.lock()?; - let transaction = connection - .transaction_with_behavior(TransactionBehavior::Immediate) - .map_err(persistence)?; - if let Some(existing) = - replay_request(&transaction, &identity.request_id, "terminal", &fingerprint)? - { - return Ok(existing); - } - ensure_cycle_open(&transaction, identity)?; + // Build the terminal outcome once, before the busy-retry loop, so the + // generated `outcome_id`/timestamp stay stable across retries (issue + // #4468). let outcome = self.new_outcome(actor, identity, payload, raw_semantic, evidence); let outcome_json = serde_json::to_vec(&outcome).map_err(serialization)?; - insert_terminal(&transaction, &outcome, &fingerprint, &outcome_json)?; - record_request_json( - &transaction, - &outcome.request_id, - "terminal", - &fingerprint, - &outcome_json, - )?; - transaction.commit().map_err(persistence)?; - Ok(outcome) + // Serialize the terminal write under bounded busy/locked retry (issue + // #4468): this feeds record_no_action / record_action_denied, which + // contend with concurrent cycles and startup recovery on the shared + // sqlite file. The lock is acquired *inside* the closure so the guard + // is released across each backoff sleep. + retry_on_busy(|| { + let mut connection = self.lock()?; + let transaction = connection + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(persistence)?; + if let Some(existing) = + replay_request(&transaction, &identity.request_id, "terminal", &fingerprint)? + { + return Ok(existing); + } + ensure_cycle_open(&transaction, identity)?; + insert_terminal(&transaction, &outcome, &fingerprint, &outcome_json)?; + record_request_json( + &transaction, + &outcome.request_id, + "terminal", + &fingerprint, + &outcome_json, + )?; + transaction.commit().map_err(persistence)?; + Ok(outcome.clone()) + }) } fn record_action_denied( @@ -3706,14 +3744,28 @@ mod outbox_serialization_tests { // writer overlaps with recovery contention on the same file. let barrier = Arc::new(Barrier::new(WRITERS + 1)); + // Construct every handler up front, before the barrier. A transient + // open-time lock is now retried inside `open` (issue #4468), but even + // so a spawned participant must never `.expect()`-panic *before* + // `barrier.wait()`: that would leave the remaining threads blocked on + // the barrier forever. Hoisting construction keeps this test focused on + // its subject — concurrent *writes* and a concurrent *recovery* sweep + // over separate connections to the same file — and turns any open + // failure into a clean test failure instead of a barrier deadlock. + let writer_handlers: Vec = (0..WRITERS) + .map(|_| { + CapabilityHandler::open(&db_path, CapabilityPolicy::new(POLICY_REVISION)) + .expect("open writer handler") + }) + .collect(); + let recovery_handler = + CapabilityHandler::open(&db_path, CapabilityPolicy::new(POLICY_REVISION)) + .expect("open recovery handler"); + let mut threads = Vec::new(); - for writer in 0..WRITERS { - let path = db_path.clone(); + for (writer, handler) in writer_handlers.into_iter().enumerate() { let barrier = Arc::clone(&barrier); threads.push(thread::spawn(move || -> Vec { - let handler = - CapabilityHandler::open(&path, CapabilityPolicy::new(POLICY_REVISION)) - .expect("open writer handler"); let mut locked_errors = Vec::new(); barrier.wait(); for iteration in 0..ITERATIONS { @@ -3741,12 +3793,9 @@ mod outbox_serialization_tests { // Concurrent startup-recovery sweeper: a separate connection running // the outbox recovery path repeatedly while the writers commit. { - let path = db_path.clone(); let barrier = Arc::clone(&barrier); + let handler = recovery_handler; threads.push(thread::spawn(move || -> Vec { - let handler = - CapabilityHandler::open(&path, CapabilityPolicy::new(POLICY_REVISION)) - .expect("open recovery handler"); let mut locked_errors = Vec::new(); barrier.wait(); for sweep in 0..(WRITERS * ITERATIONS / 4) { @@ -3799,7 +3848,78 @@ mod outbox_serialization_tests { } // --------------------------------------------------------------------- - // Direct unit coverage for the SQLite busy/locked retry mechanism + // Regression: concurrent first-open of the shared ledger must survive + // lock contention (issue #4468). + // + // `CapabilityHandler::open` runs schema-init WRITES — the WAL-mode switch + // and the CREATE-TABLE migration transaction. When many handlers open the + // SAME fresh sqlite file at once (the daemon's writers plus the + // startup-recovery sweeper, as separate connections), those writes contend + // and a transient SQLITE_BUSY / "database is locked" can surface from the + // migration. Before the fix that failed handler construction outright. + // After the fix (busy_timeout + bounded busy-retry around `initialize`) + // every concurrent open must succeed. + // + // `barrier.wait()` runs BEFORE `open` so no participant can panic ahead of + // the barrier: an open failure surfaces as a joined `Err`, never a + // barrier deadlock. + // --------------------------------------------------------------------- + + #[test] + fn concurrent_open_of_shared_ledger_survives_contention() { + let dir = tempfile::tempdir().expect("tempdir"); + let db_path: PathBuf = dir.path().join("outcomes.sqlite3"); + + const OPENERS: usize = 12; + let barrier = Arc::new(Barrier::new(OPENERS)); + + let mut threads = Vec::new(); + for _ in 0..OPENERS { + let path = db_path.clone(); + let barrier = Arc::clone(&barrier); + threads.push(thread::spawn(move || -> Result<(), String> { + // Synchronize BEFORE the open so every handler races schema + // initialization against the others on the same fresh file. + barrier.wait(); + CapabilityHandler::open(&path, CapabilityPolicy::new(POLICY_REVISION)) + .map(|_| ()) + .map_err(|error| error.to_string()) + })); + } + + let mut failures = Vec::new(); + for thread in threads { + if let Err(error) = thread.join().expect("open thread must not panic") { + failures.push(error); + } + } + + assert!( + failures.is_empty(), + "concurrent first-open of the shared ledger must never fail on lock \ + contention; got {} failure(s): {:#?}", + failures.len(), + failures, + ); + + // The schema is usable after the concurrent-open race: a subsequent + // write commits durably, proving initialization completed exactly once + // and left the file in a consistent state. + let handler = CapabilityHandler::open(&db_path, CapabilityPolicy::new(POLICY_REVISION)) + .expect("open after concurrent race"); + let actor = file_issue_actor("session-open", "cycle-open", "goal-open"); + let req = file_issue_request("req-open", "session-open", "cycle-open", "goal-open"); + handler + .record_action(&actor, req, &admission()) + .expect("write after concurrent open must succeed"); + assert!( + handler + .terminal_for_cycle("session-open", "cycle-open") + .expect("query terminal") + .is_some(), + "terminal must be durably persisted after the concurrent-open race" + ); + } // (issue #4468). `retry_on_busy` takes an injectable closure, so its // control flow — retry on contention, bounded exhaustion, and immediate // passthrough for non-contention errors — is unit-testable deterministically