diff --git a/alkstore-contract-suite/src/properties.rs b/alkstore-contract-suite/src/properties.rs index 8bea08f..93496b3 100644 --- a/alkstore-contract-suite/src/properties.rs +++ b/alkstore-contract-suite/src/properties.rs @@ -897,9 +897,16 @@ pub async fn drop_rollback_leaves_no_ghosts(factory: &dyn StoreFactory) { /// pin in the drop-rollback and commit-visibility rows; this row pins /// the visibility itself.) /// +/// The negative-seed leg (2026-10-11): a staged save seeding an +/// absent checkpoint stores raw — the boundary/offset totality of +/// ADR-023 §2 rides the tx seam too, in-tx read-back included (a +/// negative seed reads raw, never clamped against the absent-consumer +/// phantom zero, and survives the commit as the stored checkpoint). +/// /// Contract stamp: ADR-021 §1 (tx-side read methods, /// read-your-own-writes); ADR-007 (the commit-atomic seam the reads -/// ride). +/// ride); ADR-023 §2 boundary/offset annotation (annotated +/// 2026-10-11 — a negative is stored as the checkpoint). pub async fn in_tx_reads_see_own_writes(factory: &dyn StoreFactory) { let store = factory.open().await.expect("factory opens a store"); @@ -934,8 +941,43 @@ pub async fn in_tx_reads_see_own_writes(factory: &dyn StoreFactory) { ); let checkpoint = tx.get_offset_tx("events", "c").await.unwrap(); assert_eq!(checkpoint, offset, "the tx sees its own save"); + + // The tx form's seed arm (ADR-023 §2, annotated 2026-10-11): a + // staged save on a consumer with no committed checkpoint seeds + // raw — the in-tx read-back is the raw seed, not the absent-consumer + // phantom zero clamped over it, and the commit stores it verbatim. + tx.save_offset_tx("events", "c2", -5).await.unwrap(); + let seeded = tx.get_offset_tx("events", "c2").await.unwrap(); + assert_eq!( + seeded, -5, + "a staged negative seed reads raw in-tx (no phantom-zero clamp)" + ); tx.commit().await.unwrap(); + let committed_seed = store + .stream("events") + .await + .unwrap() + .get_offset("c2") + .await + .unwrap(); + assert_eq!( + committed_seed, -5, + "a negative seed is stored as the checkpoint through the tx seam" + ); + let resumed = store + .stream("events") + .await + .unwrap() + .read_from_consumer("c2", 10) + .await + .unwrap(); + assert_eq!( + resumed.iter().map(|e| e.offset).collect::>(), + vec![offset], + "reads anchored at the negative checkpoint resume forward: the log from the start" + ); + factory.teardown().await.expect("factory teardown"); } diff --git a/alkstore-mem/src/scheduler.rs b/alkstore-mem/src/scheduler.rs index 3b02425..e4b0a2d 100644 --- a/alkstore-mem/src/scheduler.rs +++ b/alkstore-mem/src/scheduler.rs @@ -64,6 +64,15 @@ const SLEEP_RENEW_EVERY_S: u64 = 2; /// The idle sleep when nothing is scheduled (soonest = 0). const IDLE_SLEEP_S: u64 = 60; +/// The upper bound on one sleep slice, in seconds: a schedule whose +/// next boundary is further out sleeps this long, ticks (a no-op), and +/// sleeps again — the tick recomputes the soonest every iteration. +/// The cap keeps `Instant + Duration` arithmetic on the sleep deadline +/// inside every platform's `Instant` range (a far-future boundary's +/// raw wait can exceed a platform's representable span; a bounded +/// duration never does). +const MAX_SLEEP_S: u64 = 60 * 60 * 24 * 30; + /// The bounded catch-up cap (ADR-009 §4 — a fixed contract constant, /// honker parity, not a knob). const MAX_CATCHUP_FIRES: i64 = 64; @@ -327,7 +336,7 @@ async fn keep_awake_renewing(lease: &MemLock, clock: &Clock, stop: &StopToken, s } else { soonest_at.saturating_sub(now).max(1) as u64 }; - let deadline = started + Duration::from_secs(wait_s); + let deadline = started + Duration::from_secs(wait_s.min(MAX_SLEEP_S)); loop { let now = tokio::time::Instant::now(); if stop.is_cancelled() || now >= deadline { diff --git a/alkstore-mem/src/stream.rs b/alkstore-mem/src/stream.rs index 00396a3..55dc1c3 100644 --- a/alkstore-mem/src/stream.rs +++ b/alkstore-mem/src/stream.rs @@ -8,9 +8,13 @@ //! internal and never compared cross-engine — ADR-015 §4: mem's //! fresh-engine count starts at 1, the SQL engines' counters may //! not), keys +//! not), keys //! are carried metadata (a present key must be non-empty — //! `InvalidName`, ADR-015 §1), and the commit boundary fires one -//! wake per event on the stream's engine-internal wake channel. +//! wake per published event on the stream's engine-internal wake +//! channel (the no-handle publish path; a tx commit coalesces to +//! one wake per stream — the feed is plumbing, not contract +//! surface, and receivers drain by cursor). //! - Cursor reads are `offset ASC` (global FIFO, ADR-015 §4) with the //! extent guard at the trait-impl entry (`limit <= 0` → empty //! `Vec`, ADR-023 §2). @@ -104,6 +108,16 @@ pub(crate) fn get_offset_row(s: &StreamState, stream: &str, consumer: &str) -> i .unwrap_or(0) } +/// Whether a checkpoint row exists for the consumer. The absent read is +/// 0 (`get_offset_row`), but the tx merge needs the distinction: a +/// staged save on a consumer with no committed row is a *seed* (stored +/// raw, per ADR-023 §2's boundary/offset annotation), not a candidate +/// against a phantom zero. +pub(crate) fn has_offset_row(s: &StreamState, stream: &str, consumer: &str) -> bool { + s.offsets + .contains_key(&(stream.to_string(), consumer.to_string())) +} + /// `trim_to`: delete `offset <= horizon` (exact boundary); surviving /// events keep their offsets. Returns the deleted count. Wakes nothing. pub(crate) fn trim_events(s: &mut StreamState, stream: &str, horizon: i64) -> i64 { diff --git a/alkstore-mem/src/tx.rs b/alkstore-mem/src/tx.rs index 09eaa60..79b8448 100644 --- a/alkstore-mem/src/tx.rs +++ b/alkstore-mem/src/tx.rs @@ -22,9 +22,11 @@ //! invisible to every other view until its commit — no isolation //! questions can arise across handles). //! - **Commit-merge + wake-at-commit**: `commit` consumes the handle; -//! the merge is one guarded operation (queues → streams, the -//! canonical lock order) and the overlay's pending wakes fire in the -//! same operation, after the merged state is visible — a tx-staged +//! the merge lands queues then streams (the canonical lock order; +//! two sequential guarded sections on a non-awaiting path — each +//! section's committed view is self-consistent at its boundary) and +//! the overlay's pending wakes fire after the merged state is +//! visible — a tx-staged //! notify delivers nothing until the merge, one wake per staged //! notify (per-notify, never coalesced). That discipline is what the //! pg engine had to fix retroactively (`pg-fix-tx-wake`); here it is @@ -58,7 +60,7 @@ use crate::resolution::{ use crate::state::Guard; use crate::state::{EventRow, JobRow, State}; use crate::stream::{ - get_offset_row, insert_event, read_events, save_offset_row, stream_wake_channel, + get_offset_row, has_offset_row, insert_event, read_events, save_offset_row, stream_wake_channel, }; use crate::wakes::WakeBus; @@ -152,14 +154,27 @@ fn merged_read_since( /// The merged checkpoint read (absent everywhere = 0). fn merged_get_offset(overlay: &Overlay, state: &State, stream: &str, consumer: &str) -> i64 { - let committed = { + let (committed, present) = { let s = state.streams.guard(); - get_offset_row(&s, stream, consumer) + ( + get_offset_row(&s, stream, consumer), + has_offset_row(&s, stream, consumer), + ) }; let staged = overlay .saves .get(&(stream.to_string(), consumer.to_string())); - staged.map_or(committed, |s| (*s).max(committed)) + // The in-tx read shows what the commit would produce: a staged + // save against an existing checkpoint rides the monotone guard + // (the staged save only merges when it is ahead, so the read is + // the max), while a staged save *seeding* an absent checkpoint is + // stored raw — a negative seed reads raw in-tx too (ADR-023 §2's + // boundary/offset annotation), not clamped against a phantom zero. + match staged { + Some(s) if present => (*s).max(committed), + Some(s) => *s, + None => committed, + } } fn event_of(row: EventRow, stream: &str) -> StreamEvent { @@ -420,9 +435,9 @@ impl MemTxHandle { } /// The commit-merge: move every staged effect into the committed state -/// under one guarded operation (the canonical queues → streams lock -/// order; each section's committed view is self-consistent at its -/// boundary), then fire the overlay's pending wakes — the +/// (queues then streams, the canonical lock order; two sequential +/// guarded sections on a non-awaiting path) then fire the overlay's +/// pending wakes — the /// wake-at-commit discipline: nothing delivers until the merge lands. fn merge_and_wake(state: Arc, wakes: Arc, mut overlay: Overlay) { { diff --git a/alkstore-mem/src/wakes.rs b/alkstore-mem/src/wakes.rs index edf03d3..8888383 100644 --- a/alkstore-mem/src/wakes.rs +++ b/alkstore-mem/src/wakes.rs @@ -33,11 +33,6 @@ //! one commit's notify on a channel is one wake per subscribed feed, //! never coalesced (no batching boundary exists to coalesce across). -// The substrate routes every wave-6 mechanism's commit-boundary -// wakes; until the mechanism tasks wire it, the foundations-only -// build sees the attach/fire surface as dead. -#![allow(dead_code)] - use std::sync::Arc; use std::sync::Mutex; @@ -89,7 +84,6 @@ impl std::fmt::Debug for WakeBus { } impl WakeBus { - #[allow(dead_code)] pub(crate) fn new() -> Self { Self { registry: Mutex::new(Registry::default()), diff --git a/alkstore-postgres/src/forwarder.rs b/alkstore-postgres/src/forwarder.rs index 9a88c87..dbc1011 100644 --- a/alkstore-postgres/src/forwarder.rs +++ b/alkstore-postgres/src/forwarder.rs @@ -136,7 +136,7 @@ impl ChannelSet { let mut map = self .channels .lock() - .expect("channel registry mutex poisoned"); + .unwrap_or_else(std::sync::PoisonError::into_inner); let is_new = !map.contains_key(channel); *map.entry(channel.to_string()).or_insert(0) += 1; is_new @@ -150,7 +150,7 @@ impl ChannelSet { let mut map = self .channels .lock() - .expect("channel registry mutex poisoned"); + .unwrap_or_else(std::sync::PoisonError::into_inner); let Some(count) = map.get_mut(channel) else { return false; }; @@ -168,7 +168,7 @@ impl ChannelSet { pub(crate) fn snapshot(&self) -> Vec { self.channels .lock() - .expect("channel registry mutex poisoned") + .unwrap_or_else(std::sync::PoisonError::into_inner) .keys() .cloned() .collect() @@ -414,7 +414,7 @@ impl Forwarder { pub(crate) fn subscribe(&self) -> Option> { self.fanout .lock() - .expect("fanout mutex poisoned") + .unwrap_or_else(std::sync::PoisonError::into_inner) .as_ref() .map(|s| s.subscribe()) } @@ -433,8 +433,14 @@ impl Forwarder { /// /// Fails `Database` when no listener connection is currently live /// (mid-reconnect) or the LISTEN itself failed — transient; a - /// retry recovers it (the registry entry stays either way and the - /// re-issue covers the channel on reconnect). + /// retry recovers it (the caller's retry re-adds the reference). + /// The failure disposition is the **call site's**: a failed + /// register leaves no receiver whose drop would decrement the + /// registry, so the failing site must `unregister` the reference + /// (the notify `listen` and stream `spawn_bridge` paths do + /// exactly that) — a phantom entry here would wedge the + /// last-subscriber UNLISTEN rule and keep the channel LISTENing + /// after every receiver is gone. pub(crate) async fn register(&self, channel: &str) -> Result<(), alkstore::Error> { self.channels.add_subscriber(channel); if !self.connected.load(Ordering::Acquire) { @@ -493,7 +499,10 @@ impl Forwarder { // Take the sender even if a receiver holds this Arc: the // receivers' terminal `None` must fire at engine shutdown // whatever lifetime stands on the handle. - *self.fanout.lock().expect("fanout mutex poisoned") = None; + *self + .fanout + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = None; } /// Aim the reconnect arm's connect step at `config` (the failed- @@ -505,7 +514,7 @@ impl Forwarder { *self .reconnect_config .lock() - .expect("reconnect config mutex poisoned") = config; + .unwrap_or_else(std::sync::PoisonError::into_inner) = config; } /// The reconnect config's current content (the failed-connect @@ -514,7 +523,7 @@ impl Forwarder { pub(crate) fn reconnect_config_snapshot(&self) -> tokio_postgres::Config { self.reconnect_config .lock() - .expect("reconnect config mutex poisoned") + .unwrap_or_else(std::sync::PoisonError::into_inner) .clone() } @@ -581,7 +590,10 @@ pub(crate) struct FanoutRelease { impl Drop for FanoutRelease { fn drop(&mut self) { - *self.slot.lock().expect("fanout mutex poisoned") = None; + *self + .slot + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = None; } } @@ -819,7 +831,9 @@ async fn forwarder_loop( let attempt = { let slot = reconnect_config.clone(); move || { - let guard = slot.lock().expect("reconnect config mutex poisoned"); + let guard = slot + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); let config = guard.clone(); drop(guard); async move { config.connect(NoTls).await.ok() } diff --git a/alkstore-postgres/src/lock.rs b/alkstore-postgres/src/lock.rs index 05009fb..799fb3d 100644 --- a/alkstore-postgres/src/lock.rs +++ b/alkstore-postgres/src/lock.rs @@ -61,11 +61,12 @@ fn table(schema: &str, name: &str) -> String { /// the holder column decides, not the insert's row count (the D-4 /// swallow's lesson: never trust a write result for ownership). /// -/// One statement sequence on one pool connection's implicit -/// transaction (auto-committed as a unit on return-to-pool; the -/// expiry-delete + insert + read-back race window is closed by the -/// PK constraint — a concurrent grant of the same name either -/// lands first or reads as foreign). +/// One statement sequence on one pool connection in autocommit mode +/// (each statement commits as it lands; the acquire is **not** a +/// single transaction): the correctness guarantee is the PK +/// constraint, not statement atomicity — a concurrent grant of the +/// same name either lands first (the read-back then shows the foreign +/// owner) or reads as foreign. pub(crate) async fn lock_acquire( client: &tokio_postgres::Client, schema: &str, diff --git a/alkstore-postgres/src/notify.rs b/alkstore-postgres/src/notify.rs index 0b89366..2fa6108 100644 --- a/alkstore-postgres/src/notify.rs +++ b/alkstore-postgres/src/notify.rs @@ -107,7 +107,14 @@ pub(crate) fn listen( if closed.load(Ordering::Acquire) { return Err(database_error("the store is closed")); } - forwarder.register(&channel).await?; + // A failed registration leaves no receiver whose drop would + // decrement the registry — unregister here, mirroring the + // subscribe-failure arm below (no refcount leak past a failed + // listen; a retry re-adds the reference). + if let Err(e) = forwarder.register(&channel).await { + forwarder.unregister(&channel); + return Err(e); + } let broadcast_rx = forwarder.subscribe(); let Some(broadcast_rx) = broadcast_rx else { // The subscribe raced (or arrived past) a shutdown: fail diff --git a/alkstore-postgres/src/scheduler.rs b/alkstore-postgres/src/scheduler.rs index 0760dad..a07255d 100644 --- a/alkstore-postgres/src/scheduler.rs +++ b/alkstore-postgres/src/scheduler.rs @@ -148,6 +148,15 @@ const SLEEP_SLICE_S: u64 = 1; /// The idle sleep when nothing is scheduled (soonest = 0). const IDLE_SLEEP_S: u64 = 60; +/// The upper bound on one sleep deadline, in seconds: a schedule whose +/// next boundary is further out sleeps this long, ticks (a no-op), and +/// sleeps again — the tick recomputes the soonest every iteration. +/// The cap keeps `Instant + Duration` arithmetic on the sleep deadline +/// inside every platform's `Instant` range (a far-future boundary's +/// raw wait can exceed a platform's representable span; a bounded +/// duration never does). +const MAX_SLEEP_S: u64 = 60 * 60 * 24 * 30; + /// The tick's transient-error retry budget: a tick failed by a /// pool/database error (pool checkout, `BEGIN`, a statement, `COMMIT`) /// retries this many times before the loop exits `Err` (the module @@ -784,7 +793,7 @@ async fn keep_awake_renewing(lease: &mut Box, stop: &StopTok // due immediately (fire on the next iteration, don't sleep). (soonest - now_unix()).max(0) as u64 }; - let deadline = started + Duration::from_secs(wait_s); + let deadline = started + Duration::from_secs(wait_s.min(MAX_SLEEP_S)); let slice = Duration::from_secs(SLEEP_SLICE_S); loop { let now = tokio::time::Instant::now(); diff --git a/alkstore-postgres/src/stream.rs b/alkstore-postgres/src/stream.rs index 7bdda25..ef56080 100644 --- a/alkstore-postgres/src/stream.rs +++ b/alkstore-postgres/src/stream.rs @@ -577,8 +577,13 @@ async fn spawn_bridge( ) -> Result> { // Registration first — a failed registration here (never: LISTEN // failures carry the registry entry's recovery, but a shutdown - // race can kill the command channel) fails the whole subscribe. - forwarder.register(&stream).await?; + // race can kill the command channel) fails the whole subscribe, + // unregistered (a failed register leaves no receiver whose drop + // would decrement the registry — the notify path's same rule). + if let Err(e) = forwarder.register(&stream).await { + forwarder.unregister(&stream); + return Err(e); + } let Some(broadcast_rx) = forwarder.subscribe() else { // The subscribe raced (or arrived past) a shutdown: fail // closed — no receiver can ever deliver again. Unregister diff --git a/alkstore-postgres/src/tx.rs b/alkstore-postgres/src/tx.rs index cce854e..774dd2a 100644 --- a/alkstore-postgres/src/tx.rs +++ b/alkstore-postgres/src/tx.rs @@ -437,14 +437,16 @@ impl TxHandle for PgTxHandle { Box::pin(async move { let client = self.client()?; // Monotone upsert (one owner): a first save seeds the - // checkpoint — clamped at 0, the absent-consumer=0 rule - // (ADR-019 §1) — and a save at-or-below the stored one - // updates nothing (ADR-019 §6's silent no-op). + // checkpoint — raw, no clamp (ADR-023 §2's 2026-10-11 + // boundary/offset annotation: the argument is an offset, + // a negative is stored as the checkpoint) — and a save + // at-or-below the stored one updates nothing (ADR-019 §6's + // silent no-op). client .execute( &format!( "INSERT INTO {} AS o (stream, consumer, \"offset\") - VALUES ($1, $2, GREATEST($3::bigint, 0)) + VALUES ($1, $2, $3::bigint) ON CONFLICT (stream, consumer) DO UPDATE SET \"offset\" = EXCLUDED.\"offset\" WHERE EXCLUDED.\"offset\" > o.\"offset\"", diff --git a/alkstore-sqlite/src/lib.rs b/alkstore-sqlite/src/lib.rs index 4c40d83..4822f83 100644 --- a/alkstore-sqlite/src/lib.rs +++ b/alkstore-sqlite/src/lib.rs @@ -11,9 +11,11 @@ //! unsupported is the inherited lineage honesty, stated //! here as the crate's deployment boundary (compile-time engine //! identity is the only engine-discovery mechanism — pick this crate, -//! get this posture). Durability knobs, pool sizing, and the watcher -//! cadence are engine-configuration concerns on [`SqliteOpts`] -//! (ADR-008 §6) — never contract surface. +//! get this posture). The engine's configuration surface is +//! [`SqliteOpts`] (the watcher cadence and the reader-pool size) — +//! the durability settings (`synchronous = NORMAL`, the busy +//! timeout) are bootstrapped constants, not knobs, per +//! deployment.md's posture (ADR-008 §6) — never contract surface. //! //! **Writer parking** (ADR-007's negative consequence, stated //! honestly): the writer slot is a lease, and long transactions park diff --git a/alkstore-sqlite/src/scheduler.rs b/alkstore-sqlite/src/scheduler.rs index 305741d..e445ff5 100644 --- a/alkstore-sqlite/src/scheduler.rs +++ b/alkstore-sqlite/src/scheduler.rs @@ -42,6 +42,15 @@ use crate::substrate::{ scheduler_soonest, scheduler_tick, scheduler_unregister, }; +/// The upper bound on one sleep slice, in seconds: a schedule whose +/// next boundary is further out sleeps this long, ticks (a no-op), and +/// sleeps again — the tick recomputes the soonest every iteration. +/// The cap keeps `Instant + Duration` arithmetic on the sleep deadline +/// inside every platform's `Instant` range (a far-future boundary's +/// raw wait can exceed a platform's representable span; a bounded +/// duration never does). +const MAX_SLEEP_S: u64 = 60 * 60 * 24 * 30; + /// The leadership lock's TTL (seconds): engine-internal detail — /// renewed every loop iteration *and* across the sleeps (a runner that /// cannot renew has lost the lock, ADR-009 §4 — the lock machinery is @@ -164,6 +173,8 @@ pub(crate) fn run_schedules( stop: StopToken, writer: Arc, closed: Arc, + db_path: String, + #[cfg(test)] commit_fault: crate::seam::commit_fault::Flag, ) -> BoxedFuture<'static, Result<()>> { Box::pin(async move { if closed.load(Ordering::Acquire) { @@ -211,7 +222,32 @@ pub(crate) fn run_schedules( } let soonest = { let writer = writer.clone(); - with_writer(writer, scheduler_tick_transactional).await? + let db_path = db_path.clone(); + #[cfg(test)] + let commit_fault = commit_fault.clone(); + crate::seam::blocking(move || { + let conn = writer.acquire().ok_or_else(closed_store_error)?; + #[cfg(test)] + let (out, conn, poisoned) = + scheduler_tick_transactional(conn, commit_fault.clone()); + #[cfg(not(test))] + let (out, conn, poisoned) = scheduler_tick_transactional(conn); + if poisoned { + // A failed COMMIT (or ROLLBACK) leaves the + // transaction's state unknown: the connection + // is dropped, not released — the tx-handle + // commit arm's posture, with the slot + // replenished via a fresh bootstrapped open. + drop(conn); + if let Ok(fresh) = crate::seam::open_writer_connection(&db_path) { + writer.release(fresh); + } + } else { + writer.release(conn); + } + out + }) + .await? }; if stop.is_cancelled() { let _ = lease.release().await; @@ -229,23 +265,47 @@ pub(crate) fn run_schedules( /// One tick under `BEGIN IMMEDIATE`: the fire enqueues + the boundary /// advance commit together, or the whole tick rolls back (crash /// mid-tick ⇒ the boundary refires; two tickers serialize on the -/// writer). Returns the soonest next fire for the sleep (0 = none). -fn scheduler_tick_transactional(conn: &rusqlite::Connection) -> Result { - conn.execute_batch("BEGIN IMMEDIATE") - .map_err(sqlite_error)?; +/// writer). Takes the connection by value and returns it with its +/// disposition: `poisoned` marks the commit-failure/rollback-failure +/// arms — the transaction's state is unknown there (SQLite documented +/// behavior: a failed COMMIT leaves the transaction active), and the +/// caller must drop the connection and replenish the slot rather than +/// release it (a released connection would keep the open transaction, +/// silently swallowing every later write until close). Returns +/// `(result, connection, poisoned)`; the soonest next fire is for the +/// sleep (0 = none). +fn scheduler_tick_transactional( + conn: rusqlite::Connection, + #[cfg(test)] commit_fault: crate::seam::commit_fault::Flag, +) -> (Result, rusqlite::Connection, bool) { + if let Err(begin) = conn.execute_batch("BEGIN IMMEDIATE") { + // A failed BEGIN opened no transaction: the connection is + // clean and releasable. + return (Err(sqlite_error(begin)), conn, false); + } let body = (|| { - let now = crate::resolution::now_unix(conn)?; - scheduler_tick(conn, now).map_err(sqlite_error)?; - scheduler_soonest(conn).map_err(sqlite_error) + let now = crate::resolution::now_unix(&conn)?; + scheduler_tick(&conn, now).map_err(sqlite_error)?; + scheduler_soonest(&conn).map_err(sqlite_error) })(); match body { Ok(soonest) => { - conn.execute_batch("COMMIT").map_err(sqlite_error)?; - Ok(soonest) + #[cfg(test)] + let committed = if crate::seam::commit_fault::take(&commit_fault) { + Err(crate::seam::commit_fault::failure()) + } else { + conn.execute_batch("COMMIT") + }; + #[cfg(not(test))] + let committed = conn.execute_batch("COMMIT"); + match committed { + Ok(()) => (Ok(soonest), conn, false), + Err(commit) => (Err(sqlite_error(commit)), conn, true), + } } Err(e) => { - let _ = conn.execute_batch("ROLLBACK"); - Err(e) + let rolled = conn.execute_batch("ROLLBACK"); + (Err(e), conn, rolled.is_err()) } } } @@ -265,7 +325,7 @@ async fn keep_awake_renewing(lease: &SqliteLockHandle, stop: &StopToken, soonest .unwrap_or(0); (soonest - now).max(1) as u64 }; - let deadline = started + Duration::from_secs(wait_s); + let deadline = started + Duration::from_secs(wait_s.min(MAX_SLEEP_S)); loop { let now = tokio::time::Instant::now(); if stop.is_cancelled() || now >= deadline { diff --git a/alkstore-sqlite/src/store.rs b/alkstore-sqlite/src/store.rs index 2f39efa..59d1e4e 100644 --- a/alkstore-sqlite/src/store.rs +++ b/alkstore-sqlite/src/store.rs @@ -293,7 +293,18 @@ impl Store for SqliteStore { ) -> alkstore::BoxedFuture<'a, alkstore::Result<()>> { let writer = self.writer.clone(); let closed = self.closed.clone(); - crate::scheduler::run_schedules(stop, writer, closed) + let db_path = self.db_path.to_string_lossy().into_owned(); + #[cfg(test)] + let fut = crate::scheduler::run_schedules( + stop, + writer, + closed, + db_path, + self.commit_fault.clone(), + ); + #[cfg(not(test))] + let fut = crate::scheduler::run_schedules(stop, writer, closed, db_path); + fut } } diff --git a/alkstore-sqlite/src/store/scheduler_tests.rs b/alkstore-sqlite/src/store/scheduler_tests.rs index 23b82e3..8ed9ec8 100644 --- a/alkstore-sqlite/src/store/scheduler_tests.rs +++ b/alkstore-sqlite/src/store/scheduler_tests.rs @@ -543,3 +543,59 @@ async fn catch_up_replays_up_to_the_cap_then_skips_forward() { drop(store); cleanup(&dir); } + +/// Acceptance: a failing `COMMIT` inside the scheduler tick's +/// transaction (the same `SQLITE_FULL`-shaped seam injection as the +/// tx-handle arm) surfaces as the opaque `Database`, and the writer +/// slot is replenished through the tick's poison arm — the connection +/// whose transaction state is unknowable (SQLite documented behavior: +/// a failed COMMIT leaves the transaction active) is dropped, the slot +/// refilled via a fresh bootstrapped open, the tx-handle commit arm's +/// posture applied to the loop's tick (the one commit site that was +/// missing it — the 2026-10-11 general review's finding; a released +/// connection would have kept the open transaction and every later +/// `BEGIN IMMEDIATE` on the slot would have errored until close). The +/// store stays fully usable: schedule, notify, and a follow-up runner +/// all proceed. +#[tokio::test(flavor = "multi_thread")] +async fn failed_tick_commit_replenishes_the_writer_slot() { + let dir = temp_dir("tick-commit-err"); + let concrete = open_store(dir.join("store.db").to_str().unwrap(), Default::default()).unwrap(); + + concrete.arm_commit_fault(); + + // The run errors at the first tick's commit (the leadership + // acquire and the tick's BEGIN both preceded it). + let outcome = concrete.run_schedules(StopToken::new()).await; + match outcome { + Err(Error::Database(source)) => { + assert!( + source.to_string().contains("full"), + "the tick commit error carries the SQLITE_FULL-shaped source chain: {source}" + ); + } + other => panic!("a failed tick commit errs opaque Database, got {other:?}"), + } + + let registered = concrete + .schedule( + "after", + "@every 10s", + "work", + json!({"n": 1}), + ScheduleOpts::default(), + ) + .await; + assert!( + registered.is_ok(), + "the slot must be grantable after the failed tick commit (replenished via reopen): \ + {registered:?}" + ); + concrete + .notify("probe", json!({"after": "tick-commit-error"})) + .await + .expect("auto-commit ops work after the failed tick commit"); + + drop(concrete); + cleanup(&dir); +} diff --git a/alkstore-sqlite/src/substrate/queue_ops.rs b/alkstore-sqlite/src/substrate/queue_ops.rs index cc60281..58159ec 100644 --- a/alkstore-sqlite/src/substrate/queue_ops.rs +++ b/alkstore-sqlite/src/substrate/queue_ops.rs @@ -343,7 +343,7 @@ pub(crate) fn scheduler_tick(conn: &Connection, now: i64) -> rusqlite::Result= SCHEDULER_MAX_CATCHUP_FIRES { diff --git a/alkstore/src/error.rs b/alkstore/src/error.rs index e9cd2eb..41b2370 100644 --- a/alkstore/src/error.rs +++ b/alkstore/src/error.rs @@ -54,9 +54,11 @@ pub enum Error { LeadershipLost, /// An operation against a closed receiver/source. /// - /// The `try_recv`/`recv_timeout` form of wake close (never - /// returned by `recv_timeout`'s timeout expiry, which yields - /// `Ok(None)`); never a timeout signal. + /// The closed arms of the wake receivers' non-`recv` forms + /// (`try_recv`/`recv_timeout` — never + /// `recv_timeout`'s timeout expiry, which yields `Ok(None)`) and + /// the event receiver's `try_recv` on a `Closed`-caused stream + /// receiver; never a timeout signal. #[error("the receiver or source is closed")] Closed, /// Payload serialization or deserialization failure. diff --git a/alkstore/src/payload.rs b/alkstore/src/payload.rs index 7157d32..88bfaad 100644 --- a/alkstore/src/payload.rs +++ b/alkstore/src/payload.rs @@ -23,13 +23,60 @@ use crate::Error; /// Fallible per ADR-023 §1: serialization failure is /// [`Error::Codec`](crate::Error::Codec), typed at this helper — no /// silent fallback form exists (an empty-byte store would decode to -/// `Null`, silent data corruption). Unreachable for `serde_json::Value` -/// in practice (its variants all serialize), pinned against an -/// encoding posture change. +/// `Null`, silent data corruption). Failures here are two-fold: a +/// `Value` nested deeper than the depth limit (`PAYLOAD_DEPTH_LIMIT`) +/// is rejected — the stored bytes could never be read back, and the +/// serializer's own unguarded recursion would otherwise abort deep +/// enough — and genuine serialization failure maps through the same +/// typed arm. pub fn encode_payload(value: &serde_json::Value) -> Result, Error> { + if value_depth(value) > PAYLOAD_DEPTH_LIMIT { + return Err(Error::Codec(format!( + "payload nesting exceeds the depth limit ({PAYLOAD_DEPTH_LIMIT})" + ))); + } serde_json::to_vec(value).map_err(|e| Error::Codec(e.to_string())) } +/// The payload nesting cap: aligned with serde_json's deserialization +/// recursion limit (a nested-open at container depth 128 fails to +/// decode; depths through 127 read back). A `Value` nested deeper than +/// the cap produces stored bytes no decoder can read — silent data +/// loss at read time — and deep enough the (unguarded) serializer +/// itself would overflow the stack abortably; the encode side refuses +/// it instead. +const PAYLOAD_DEPTH_LIMIT: usize = 127; + +/// The maximum number of containers on any path through `value` +/// (iterative, not recursive — a recursive walk would itself overflow +/// the stack on exactly the deep inputs this guards against). +fn value_depth(value: &serde_json::Value) -> usize { + let mut max = 0; + let mut stack: Vec<(&serde_json::Value, usize)> = vec![(value, 1)]; + while let Some((current, depth)) = stack.pop() { + match current { + serde_json::Value::Array(items) => { + if depth > max { + max = depth; + } + for item in items { + stack.push((item, depth + 1)); + } + } + serde_json::Value::Object(map) => { + if depth > max { + max = depth; + } + for item in map.values() { + stack.push((item, depth + 1)); + } + } + _ => {} + } + } + max +} + /// Decode stored payload bytes into a concrete target type. /// /// Used by [`Job::payload_as`](crate::Job::payload_as) and diff --git a/alkstore/src/validation.rs b/alkstore/src/validation.rs index e9abc1d..10f26b4 100644 --- a/alkstore/src/validation.rs +++ b/alkstore/src/validation.rs @@ -26,17 +26,19 @@ pub enum NameKind { /// rejected. Shared, /// Consumer-local identifiers — stream-consumer names, `owner` on - /// `try_lock`, `worker_id` on claims. Non-empty only: a leading - /// `__alkstore_` is legal (they are not a shared engine namespace, - /// so there is nothing to collide with). + /// `try_lock`, `worker_id` on claims. Non-empty (whitespace-only + /// rejected like any empty): a leading `__alkstore_` is legal + /// (they are not a shared engine namespace, so there is nothing to + /// collide with). Local, } /// Validate a name per its [`NameKind`]. /// -/// - non-empty (after no trimming — whitespace-only counts as -/// non-empty content but is nonetheless rejected below) → -/// [`Error::InvalidName`](crate::Error::InvalidName) when empty +/// - non-empty → [`Error::InvalidName`](crate::Error::InvalidName) +/// when empty; the emptiness check is the raw string's (`""` +/// rejects) with whitespace-only strings rejected by the same arm +/// (they are never a meaningful name on any engine — see below) /// - `NameKind::Shared` additionally rejects names carrying /// [`RESERVED_PREFIX`] with /// [`Error::ReservedName`](crate::Error::ReservedName) diff --git a/alkstore/src/value_type_tests.rs b/alkstore/src/value_type_tests.rs index 86c2765..8ee1419 100644 --- a/alkstore/src/value_type_tests.rs +++ b/alkstore/src/value_type_tests.rs @@ -87,7 +87,10 @@ fn sample_event(payload: Vec) -> StreamEvent { fn payload_as_round_trips_json_value_through_job() { let v = json!({"id": 9, "nested": {"ok": [true, null, 1.5]}, "s": "ünï"}); let bytes = crate::encode_payload(&v).unwrap_or_else(|e| panic!("encode failed: {e}")); - assert_eq!(bytes, serde_json::to_vec(&v).unwrap_or_default()); + assert_eq!( + bytes, + serde_json::to_vec(&v).unwrap_or_else(|e| panic!("reference encode failed: {e}")) + ); let job = sample_job(bytes); let decoded: Value = job .payload_as() @@ -128,16 +131,54 @@ fn payload_as_undecodable_target_yields_codec() { Err(crate::Error::Codec(msg)) => assert!(!msg.is_empty()), other => panic!("expected Codec, got {other:?}"), } - let r2: crate::Result = - serde_json::from_slice(b"not json").map_err(|e| crate::Error::Codec(e.to_string())); + let r2: crate::Result = { + let undecodable = sample_job(b"not json".to_vec()); + undecodable.payload_as() + }; assert!(matches!(r2, Err(crate::Error::Codec(_)))); } +#[test] +fn encode_payload_rejects_nesting_below_the_decode_limit_as_typed_codec() { + let boundary = nest_value(Value::Null, 127); + assert!( + crate::encode_payload(&boundary).is_ok(), + "the deepest decodable nesting encodes and reads back" + ); + let past = nest_value(Value::Null, 128); + match crate::encode_payload(&past) { + Err(crate::Error::Codec(msg)) => assert!(!msg.is_empty()), + other => panic!("expected Codec on past-limit nesting, got {other:?}"), + } +} + +fn nest_value(inner: Value, depth: usize) -> Value { + let mut v = inner; + for _ in 0..depth { + v = Value::Array(vec![v]); + } + v +} + +#[test] +fn encode_payload_boundary_nesting_round_trips() { + let v = nest_value(Value::from(0u8), 127); + let bytes = crate::encode_payload(&v).unwrap_or_else(|e| panic!("encode failed: {e}")); + let job = sample_job(bytes); + let back: Value = job + .payload_as() + .unwrap_or_else(|e| panic!("boundary decode failed: {e}")); + assert_eq!(back, v); +} + #[test] fn encode_payload_success_carries_exact_serialization() { let v = json!({"k": [1, 2, 3]}); let bytes = crate::encode_payload(&v).unwrap_or_else(|e| panic!("encode failed: {e}")); - assert_eq!(bytes, serde_json::to_vec(&v).unwrap_or_default()); + assert_eq!( + bytes, + serde_json::to_vec(&v).unwrap_or_else(|e| panic!("reference encode failed: {e}")) + ); } #[test] diff --git a/docs/architecture/core-contract.md b/docs/architecture/core-contract.md index c0f30cf..6e1bc89 100644 --- a/docs/architecture/core-contract.md +++ b/docs/architecture/core-contract.md @@ -792,21 +792,33 @@ before the engine specs are called `stable`: with `limit <= 0` read nothing, never the whole queue/stream), so the SQLite `LIMIT -1` dialect artifact is dead on both engines; durations reject (`try_lock`/`renew` with `ttl <= 0` are opaque - `Err`s on both engines); boundaries are total (a negative - `trim_to` horizon deletes nothing, idempotently; the same - totality over non-positive values covers `heartbeat(extend)`'s - relative-deadline argument and `save_offset`'s offset — both - classified 2026-10-11 by the close-out pass, the ADR-023 §2 - annotation; their suite legs landed at the wave-8 gate in - `job_handle_validity_predicate` and - `receiver_close_and_save_arms`, uniformly green on all three - engines). + `Err`s on both engines); boundaries are total (a negative + `trim_to` horizon deletes nothing, idempotently; the same + totality over non-positive values covers `heartbeat(extend)`'s + relative-deadline argument and `save_offset`'s offset — both + classified 2026-10-11 by the close-out pass, the ADR-023 §2 + annotation; their suite legs landed at the wave-8 gate in + `job_handle_validity_predicate` and + `receiver_close_and_save_arms`, uniformly green on all three + engines; the 2026-10-11 general review added the tx-seam form of + the `save_offset` totality — a staged negative seed reads raw + in-tx and stores raw at commit, pinned in + `in_tx_reads_see_own_writes`, again uniformly green on all three + engines). - **`encode_payload` typed failure** ([ADR-023](decisions/023-fourth-review-round.md) §1) — the helper returns `Result, Error::Codec>`; enqueue/publish store and decode the exact serialization of the input `Value` (byte-identical rows cross-engine, the §4 round-trip unchanged), and the failure arm is typed at the helper — no silent empty-byte fallback exists. + The failure set grew once (the 2026-10-11 general review, applied + pre-publish): a `Value` nested + deeper than the encode-side depth limit (127 containers — a + 128th nested open fails to decode, empirically aligned with + serde_json's recursion limit) is a typed `Codec` rejection, not a + serializable-but-undecodable store and not a serializer stack + overflow; the fuzz corpus-replay invariant pins the boundary + (`fuzz/shared/src/payload_codec.rs`). - **Plain-path SQLite open** ([ADR-023](decisions/023-fourth-review-round.md) §3) — the SQLite engine's `open` path argument is a plain filesystem path: a string carrying a `?name=value` suffix is diff --git a/docs/plans/implementation.md b/docs/plans/implementation.md index 67e1899..348567c 100644 --- a/docs/plans/implementation.md +++ b/docs/plans/implementation.md @@ -1,8 +1,9 @@ --- status: draft last_updated: "2026-10-11 (wave 8 implemented + reviewed — the release -gate passed: the class-1 window's dispositions complete, the suite's -registered boundary-legs follow-up landed, all gates re-run green; +gate passed and the pre-release general review (004) followed it, +fixing one high-severity slot-poisoning defect, two contract-observable +divergences, and their suite/test/doc pins; all gates re-run green; implementation complete pending the user's manual `cargo publish`.)" --- @@ -782,8 +783,43 @@ decomposes. Specific gates: session (fresh package listing + exclusion re-run, no leakage); the frontmatter sweep + renumber spot-checks re-run green; all gates re-run green including the fuzz replay, wasm32, and the doc gate; - the release-facing spec set (core-contract.md, overview.md, - queues.md, deployment.md, the architecture README) flipped - `stable`. Implementation complete pending the user's manual - `cargo publish` (the checklist of record is in - `tasks/release-publish-prep.md`). + the release-facing spec set (core-contract.md, overview.md, + queues.md, deployment.md, the architecture README) flipped + `stable`. Implementation complete pending the user's manual + `cargo publish` (the checklist of record is in + `tasks/release-publish-prep.md`). +- **General review, pre-release / wave 8** (2026-10-11, + `docs/reviews/004-pre-release-general-review.md` — the record is + 004 in the directory's sequence; a whole-workspace pass with four + parallel per-crate deep reads plus a cross-cutting sweep, distinct + from the gate) — nine findings fixed inline, sixteen recorded as + follow-up material, two false positives caught on verification: + the scheduler tick's failed `COMMIT`/`ROLLBACK` leaving the SQLite + writer slot wedged with a stuck-open transaction (HIGH — the one + commit site missing the drop-+-replenish arm; fixed with the + commit-fault seam reaching the tick and a replenish test); the wave-8 + gate's pg negative-seed fix having missed the **tx seam** + (`save_offset_tx` still clamped, mem's in-tx merged read clamping + against a phantom zero — fixed, and the tx-form totality pinned by a + new stamped suite leg green on all three columns vs the live + server); the pg `register`-failure refcount leak (both subscription + call sites now unregister on error); pg's fanout/registry + poisoned-mutex `expect`s converted to the mem poison-recovery idiom; + `encode_payload`'s deep-nesting process abort replaced by a typed + `Codec` rejection at a decode-aligned depth limit (127 containers — + the fuzz invariant's old "encodes but can never be read back" pin + corrected, boundary legs added, applied pre-publish within the open + class-1 window); sleep-deadline `Instant + Duration` overflow caps + and a `saturating_add` in the sqlite tick's claim stamping; six + doc-vs-code corrections (core `Error::Closed` scope, validation + wording, the sqlite lib doc's phantom "durability knobs", pg + `lock.rs`'s atomicity claim, mem's merge/wake-granularity wording); + test hygiene in `value_type_tests.rs`; the stale + `#![allow(dead_code)]` on mem's wakes module. Parked follow-up + inventory (no release blocker): sqlite seam panic-safety armor, the + watcher post-death-subscribe hang + initial-stat self-kill, the pg + forwarder TOCTOU/hang/diagnosability arms, the pg BIGINT + server-side stamp overflow, the mem growth/outbox notes, the pg + identifier-truncation doc note, and smaller records. All gates + re-run green after the fixes (build, workspace tests, live pg + column, clippy, fmt, fuzz replay 6/6, wasm32, doc gate). diff --git a/docs/reviews/004-pre-release-general-review.md b/docs/reviews/004-pre-release-general-review.md new file mode 100644 index 0000000..a3058e9 --- /dev/null +++ b/docs/reviews/004-pre-release-general-review.md @@ -0,0 +1,238 @@ +# General review — wave 8 / pre-release (whole workspace) + +- **Reviewer**: opencode (glm-5.3-flash) +- **Scope**: a general review of the full workspace as it stands at + wave-8 completion (2026-10-11, post `review-wave-8`, pre-publish) — + the same lens family as reviews 001/002/003 (correctness, security, + code smells, classic dev issues), applied to all four published + crates plus the contract suite and the fuzz sidecar, rather than a + single wave's delta. Numbering note: reviews 001 (waves 1–2), 002 + (wave 4), and 003 (wave 5) already occupy the first three slots, so + this record is **004** by the directory's sequence. +- **Input**: four parallel deep-read agents (one per crate: core; + sqlite engine + substrate; postgres; mem) plus a cross-cutting + sweep (panics/unsafe/logging/security greps, metadata, README / + CHANGELOG spot-checks, fuzz/shared), followed by session-level + verification of every finding that drove a change — each fixed + item below was re-verified by direct read and, where behavioral, + by a live run. +- **Gates at review time** (all run this session, after the fixes): + `cargo build`; `cargo test --workspace` server-less (green; core 27, + sqlite 194, mem 102, pg lib SKIP-posture, suite 25 on the + sqlite+mem columns); the **full pg column live** vs the harness + server (`pglo-poc`: 127 lib + 25 suite + 9 schema, all green — the + run that re-verified the pg fixes below through the live server); + `cargo clippy --all-targets -- -D warnings`; `cargo fmt --check`; + the fuzz corpus replay (6/6); the wasm32 check + (`cargo check --target wasm32-unknown-unknown -p alkstore-mem`); + the doc gate (`RUSTDOCFLAGS="-D warnings" cargo doc --no-deps`, + zero warnings). Harness env for the record (review 003's op note): + `ALKSTORE_PG_HOST=localhost ALKSTORE_PG_PORT=15432 + ALKSTORE_PG_USER=postgres ALKSTORE_PG_PASSWORD=poc + ALKSTORE_PG_DB=blobs`. + +## Findings fixed inline + +| # | Severity | Where | Finding | Fix | +|---|---|---|---|---| +| 1 | HIGH | `alkstore-sqlite/src/scheduler.rs` | **A failed tick `COMMIT`/`ROLLBACK` left the writer slot wedged with a stuck-open transaction.** The tick ran inside `with_writer`, which releases the connection on both arms; SQLite's documented behavior is that a failed `COMMIT` *leaves the transaction active* — so the released connection carries a live `BEGIN IMMEDIATE` forever: every later auto-commit op silently executes inside an uncommitted transaction (writes vanish at close) and every later `BEGIN IMMEDIATE` errors. Every other commit site has the drop-+-`reopen`-replenish arm (the tx handle's commit, `ConnLease`, the drop paths — pinned by `sqlite-commit-error-arm` tests); the tick was the one site missing it. Also, the tick's error arm swallowed rollback failure (`let _ = ROLLBACK`). | The tick takes the connection by value now and returns a `(result, conn, poisoned)` disposition: `COMMIT` failure or `ROLLBACK` failure marks the connection poisoned — dropped, not released, slot replenished via a fresh bootstrapped open (the established arm's posture). The `cfg(test)` commit-fault seam reaches the tick (the store's flag is threaded into `run_schedules`), and a new test pins the property: `failed_tick_commit_replenishes_the_writer_slot` (`store/scheduler_tests.rs`) injects a `SQLITE_FULL`-shaped `COMMIT` failure and asserts the error chain, slot grantability, and full store usability after. | +| 2 | MED (contract-observable) | `alkstore-postgres/src/tx.rs`, `alkstore-mem/src/tx.rs` | **The wave-8 gate's pg negative-seed fix missed the tx seam.** The gate fixed pg's auto-commit save (raw seed, `CHECK` dropped) against ADR-023 §2's 2026-10-11 annotation ("a negative is stored as the checkpoint") and pinned `save_offset`'s totality through the *direct* form — but `save_offset_tx` still clamped (`GREATEST($3, 0)`), so the same consumer's first save landed −5 through `StreamHandle::save_offset` and 0 through `TxHandle::save_offset_tx` depending on entry point, with the tx path's doc still citing the pre-annotation clamp rationale. The suite's new legs did not pin the tx form, which is why the gate's audit read passed. **Mem had the same family defect in-tx**: `merged_get_offset` reported `max(staged, committed)` against a *phantom zero* for an absent committed checkpoint, so a staged −5 seed read 0 in-tx but committed as −5 (`save_offset_row` seeds raw) — the in-tx read contradicted the merge outcome; SQLite (uncommitted in-tx SQL) was correct. | Both fixed: pg `save_offset_tx` stores `$3::bigint` raw (comment updated to the annotated posture); mem's merged read now distinguishes an absent committed row (new `has_offset_row` probe) — a staged seed reads raw in-tx, an existing checkpoint rides the monotone max as before. A new suite leg pins the tx form end-to-end (staged −5 → raw in-tx read-back → stored −5 at commit → anchored reads resume forward) in `in_tx_reads_see_own_writes`, stamped with the ADR-023 §2 (annotated 2026-10-11) accumulation; green on all three columns against the live pg server. `core-contract.md`'s numeric-domain verification row names the tx-form leg; core-contract's `save_offset_tx` prose was already uniform ("monotone upsert" — the divergence was pg/mem implementation, not contract text). | +| 3 | MED | `alkstore-postgres/src/forwarder.rs`, `src/notify.rs`, `src/stream.rs` | **A failed `register` leaked a phantom refcount.** `register` increments the channel registry *first*, then faults on three paths (mid-reconnect, forwarder-gone, ack dropped). `listen()` (`notify.rs`) propagated the error without unregistering — and since no receiver was created, no future `Drop` would ever decrement: the last-subscriber UNLISTEN rule could never fire and the reconnect re-issue kept LISTENing a channel nobody consumes. The subscribe-side failure arm two lines below already unregisters ("the acked LISTEN must not leak") — the register-side arm was the miss. `spawn_bridge` (`stream.rs`) had the same shape. | Both call sites now unregister on the register error before returning it (mirroring the existing subscribe-failure arm), and `register`'s doc is corrected: the failure disposition is the call site's (no receiver exists to drop the reference), replacing the old "the registry entry stays either way" claim that codified the leak. | +| 4 | MED | `alkstore-postgres/src/forwarder.rs` | **Nine non-test `expect()`s in the fanout/registry/config locks** — a panic while holding any of them poisoned the mutex and converted every later `register`/`subscribe`/`shutdown` into a panic in unrelated tasks, despite the workspace's no-`expect`-outside-tests rule (and the crate's own seam docs). | Converted to the mem crate's established poison-recovery idiom (`unwrap_or_else(std::sync::PoisonError::into_inner)`) — degraded-but-alive continuation matching `alkstore-mem/src/state.rs`'s locked-discipline posture. The one remaining pg `expect` (`stream.rs` wake-runtime build, a `OnceLock` fixed-config invariant) is recorded, not converted: its failure is a bug, not a condition, and converting it would change the function's signature for no path. | +| 5 | MED | `alkstore/src/payload.rs` (+ `fuzz/shared/src/payload_codec.rs`, engine enqueue paths implicitly) | **Deeply-nested payload `Value`s aborted the process at encode time.** The doc claimed "Unreachable for `serde_json::Value` in practice", but serialization's recursion is unguarded: a caller-supplied `Value` nested ~10⁵ levels overflows the stack inside `serde_json::to_vec` — an abort, not an `Err` — and `encode_payload` runs at every enqueue/publish/notify site on every engine, so one bad `Value` from one consumer kills the runtime. Worse for the pre-change state: the fuzz invariant explicitly pinned that a value nested past the decode recursion limit *encodes successfully* and only "decodes loud" later — i.e. storing bytes no reader could ever read back, silent data loss at read time. A scratch probe pinned serde_json's decode boundary empirically: 127 nested containers read back, the 128th nested open fails (`RECURSION_LIMIT` 128 counts the failing open, not the last success). | `encode_payload` now measures nesting iteratively (an explicit-stack walk — a recursive checker would itself overflow on exactly the deep inputs) and rejects anything deeper than **127 containers** with typed `Error::Codec`: exact encode/decode symmetry (everything decodable encodes; nothing undecodable stores) and no reachable abort path. Applied pre-publish, so the class-1 window covers it as a failure-set addition within ADR-023 §1's typed-failure pin — flagged here for the record's honesty. The fuzz invariant's encode expectations were corrected (past-limit *encode* fails typed; a boundary round-trip leg at 127 containers added), and two core tests pin the boundary + no phantom behavior (`value_type_tests.rs`). `payload_as`'s decode-side behavior is unchanged. | +| 6 | MED (platform-conditional) | `alkstore-mem/src/scheduler.rs`, `alkstore-sqlite/src/scheduler.rs`, `alkstore-postgres/src/scheduler.rs` | **`Instant + Duration` panic / unchecked overflow on far-future boundaries.** The three leader loops' sleep deadlines computed `tokio Instant + Duration::from_secs(soonest − now)` with `soonest` a consumer-reachable i64 (a legal `@every 9999999999999999s` interval); on macOS (u64-nano `Instant`, ~584 y) or Windows (~29 k y) the addition panics for waits past the platform's representable span (the mem agent verified the panic class; Linux tolerates it). The sleep is a wake-latency optimization — correctness never rides it. Same family: the sqlite tick's claim stamping did an unchecked `now + s` (`expires_s` consumer i64) — debug panic / release wrap, in the one spot whose neighbors all use `saturating_add`. | All three sleeps cap one slice at `MAX_SLEEP_S` (30 days): a far-future boundary sleeps, ticks (a no-op), and sleeps again — the tick recomputes the soonest every iteration and the renewals continue, so a 30-day-cap is behavior-preserving on any platform. The sqlite claim stamp is `now.saturating_add(s)` matching `resolution.rs`'s own posture. | +| 7 | LOW | `alkstore-mem/src/wakes.rs` | **Stale module-level `#![allow(dead_code)]`** whose own comment said "until the mechanism tasks wire it" — the mechanisms wired the bus in wave 6; the suppression was masking the module's real dead-code signal. | Removed (with the now-covered `#[allow(dead_code)]` on `new()`, which the `Default` impl exercises); compiles clean — the surface is genuinely live. | +| 8 | LOW (test hygiene, core) | `alkstore/src/value_type_tests.rs` | Two test smells: `payload_as_undecodable_target_yields_codec` exercised a `map_err` the test itself wrote instead of the function under test (it proved nothing); two serialization asserts compared against `to_vec(...).unwrap_or_default()` — a failure in the reference encode would diff against an empty `Vec` and misdirect. | The undecodable-bytes case now goes through `Job::payload_as` on a real undecodable job; the reference encodes panic with their cause. | +| 9 | LOW (doc accuracy) | several | Doc-vs-code mismatches found by the read, all public or contract text: core's `Error::Closed` doc scoped the variant to *wake* receivers while the stream receiver's `try_recv` produces it too; `validation.rs`'s shared-kind bullet was self-contradictory about the whitespace rule and `NameKind::Local` omitted it; `alkstore-sqlite`'s lib header promised "durability knobs … on `SqliteOpts`" although `SqliteOpts` carries only the watcher cadence and reader-pool size (`synchronous`/busy-timeout are bootstrapped constants); pg `lock.rs` claimed the acquire sequence "auto-committed as a unit" though the statements commit individually (correctness lives in the PK constraint, which the doc itself cited second); mem's tx module doc said the commit-merge is "one guarded operation" while the merge is two sequential guarded sections (queues → streams) on a non-awaiting path, and mem's stream doc said publishes fire "one wake per event" where the tx commit path coalesces per stream. | All reworded to match the code. The pg `lock.rs` and mem merge statements now state the real mechanism honestly; the sqlite lib doc now matches `SqliteOpts`'s actual fields; the `Error::Closed` doc covers both receiver families. | + +## Findings recorded, not fixed (follow-up material) + +These are real but either need their own design pass, touch lineage +or distributed machinery whose repair exceeds a review fix, or are +conscious postures to record. None blocks the release; each names its +shape so a future session can decompose without re-deriving. + +1. **sqlite seam panic-safety (defense-in-depth)** — + `with_writer`/`with_reader`/`blocking_read` (`alkstore-sqlite/src/seam.rs`, + `substrate/schema.rs`) release their slot/pool connection after the + closure returns only. A panic inside the closure skips the release: + the single writer slot wedges forever (`Writer::acquire` waits on a + condvar with no timeout), and an abandoned reader connection + permanently shrinks the pool (`Readers.outstanding` never decrements + on abandonment). The tx seam already owns this class (`ConnLease`); + the auto-commit seams don't. No *reachable* panic path today (the + library is panic-free by the standing gates, and these closures take + no consumer code), so this is armor, not exposure — but the fix + needs abandonment plumbing on both `Writer` (drop + path-based + replenish) and `Readers` (an outstanding decrement), which is a + deliberate small design task, not an inline patch. +2. **sqlite post-watcher-death subscribe hang** — after the watcher + thread exits (db-file replacement), `subscribe()` still enrolls new + senders with no poller alive (`substrate/watcher.rs:391-396`): a + `listen()` or stream `subscribe()` on the dead store parks forever, + contradicting engine-sqlite.md's "on watcher death every subscriber + closes" posture (enforced only for pre-death subscribers). Needs a + dead-store/dead-watcher sign that new subscription paths consult — + a state-machine change across `store.rs`/`notify.rs`/`stream.rs`. +3. **sqlite watcher initial-stat failure self-kill** — an + identity-`stat` failure at watcher start seeds `(0, 0)` + (`watcher.rs:164-170`); the first *successful* later stat mismatches + it and drives the death path, spuriously closing every subscriber + on recovery from a transient stat hiccup. Rare, but the fix (an + identity-unknown state instead of a false identity) is another + watcher-state change — bundle with item 2 if either is taken up. +4. **pg forwarder register/unregister TOCTOU** — the registry refcount + and the LISTEN/UNLISTEN command emission are not one critical + section (`forwarder.rs:131-176` vs `438-482`): interleaved + register/unregister across tasks can leave a live refcount with the + server-side LISTEN cleared (both queued commands share a + generation, so no stale check saves it) until the next reconnect + re-issue. The fix is a decide-and-queue restructure inside the + forwarder — forwarder surgery that should ride its own session with + the existing test-seam vocabulary. +5. **pg server-side BIGINT overflow on consumer stamps** — the claim, + heartbeat, retry-reschedule, and retention-sweep statements compute + server-side sums over consumer-supplied i64 values + (`queue.rs:386/912/1000/714`, `lock.rs:91`): one row stamped with + `visibility_timeout_s = i64::MAX` (or retention `i64::MIN`) makes + every claim/heartbeat UPDATE fail with "bigint out of range", and + the failure is *sticky* per row. The client-size resolution is + saturated (`resolution.rs`); the row-stamp arithmetic is not. Fix + shape: clamp/validate at the enqueue seam (or `LEAST`/`GREATEST` + guard the server expressions) — but the *which side* question is + contract-adjacent (ADR-023 §2's totality wording) and deserves its + own decision, not an inline patch. +6. **pg forwarder hang-shaped arms** — two narrow ones: a failed + generation-verify/LISTEN batch is treated as connection death and + parks on `poll_task.await` forever even though the session may be + alive (`forwarder.rs:698-702/803-808`); and `close()`/`Drop` can + leak the listener session past shutdown when the serve loop is + awaiting a command on a black-holed connection (the `run_command` + await sits outside the shutdown `select`, `forwarder.rs:781`, no + timeout on this connection). Both want the same treatment family — + timeouts/bounded waits on the listener path — bundle into the + forwarder-hardening session (item 4). +7. **pg forwarder diagnosability** — the reconnect arm's connect + failures are swallowed with `.ok()` and the loop's exit discards the + poll error (`forwarder.rs:825/855-869`): an operator sees nothing + before wake loss. Related cross-cutting posture: the family has no + log facade and 17 library-path `eprintln!` sites carry its + diagnostics (verified secret-free — no DSN/password material ever + printed). A `log`/`tracing` adoption decision is a family-level + call; recorded rather than made here. +8. **sqlite listener-bridge asymmetry** — `listen()` parks one + `spawn_blocking` thread per subscription (`notify.rs:98-104`, tokio + blocking-pool ceiling ~512) while the stream bridge does the + identical job on a dedicated `std::thread` + (`stream.rs:368-379`); past ~500 concurrent listeners the *whole* + engine's seams queue behind parked bridges. Aligning the two + bridges (std-thread shape) is a small contained change deferred + alongside item 2's subscription-path work. +9. **sqlite run_schedules error flattening** — + `lease.renew(...).await.unwrap_or(false)` maps any renew failure + (including the closed-store `Database` error) to `LeadershipLost` + (`scheduler.rs:205-211`), misdirecting the documented respawn recipe + at a dead store. Small, but the right arm set (closed vs lost vs + storage failure) is a contract-adjacent error-classification change + on a stable surface — record, don't rush. +10. **mem growth postures (N-1 family)** — two perf/posture notes to + carry into the mem spec's growth section when next touched: the + subscribe path drains the full backlog into the receiver's pending + deque before the first yield, and pending grows at publish rate for + a stalled consumer (`stream.rs:314-331` — the documented + "drains to the tail then parks" shape, but unbounded on the + consumer side); the claim path is an O(total-live-rows) scan with + full-row clones per claim (`queue.rs:122-181`) — blessed economics + for the ephemeral tier, worth one sentence in `engine-mem.md` if + ever revisited. +11. **mem outbox outcome swallow** — `outbox run_once` drops the ack/ + retry row-refs' booleans and returns `Ok(true)` even when delivery + outlived the visibility window and nothing landed + (`outbox.rs:104-122`). Inherent to the no-heartbeat outbox design + (at-least-once reclaim is the recovery), but a `debug_assert!` or + doc sentence pinning the "processed means delivered-or-reclaimable" + semantic would prevent future confusion. +12. **core `offset()` initial value** — `EventReceiver::offset()` + pins the post-yield value only; the pre-yield return (the seeded + checkpoint vs 0) is unpinned in the contract text. A doc-only fix + once the three engines' pre-yield values are confirmed equal (not + verified this session; left as a named gap). +13. **pg identifier truncation cross-channel collision** — + `LISTEN`/`UNLISTEN` interpolate the channel as a quoted identifier + (truncated at 63 bytes server-side) while `pg_notify` carries it as + a parameter; two distinct names sharing a 63-byte prefix can + silently collapse into one wake channel, with names unbounded by + the validation contract (the no-cap "length-independence" verdict + is deliberate). The honest cheap fix is a deployment-matrix/ + docs note ("names longer than the PG identifier limit may + collide"); deferred here because the matrix was just finalized by + the gate — fold it into whichever doc pass runs next (or into the + publish session's reading). +14. **core error `Display` log-forging** — `InvalidName`/\ + `ReservedName` / `InvalidSpec` interpolate raw consumer strings + into `Display`; control characters can forge log lines in naive + consumers. Typical-idiom risk, low; recorded. +15. **unsafe without `// SAFETY` in the substrate** — the eight + `unsafe { ctx.get_connection() }` UDF blocks + (`substrate/ops.rs:147-236`) carry no justification comments. The + pattern is the standard rusqlite UDF shape and provably + constrained, but AGENTS.md's own carve-out ("inline comments when + a non-obvious safety constraint would otherwise be missed") says + this is where they'd live. Not added this session: the substrate + is lineage-port code under ADR-012 §3's diff-fidelity posture — + comments must ride a PROVENANCE delta. Recorded for whenever the + substrate is next touched. +16. **Cosmetic metadata asymmetry** — the three engines' dev-dep tokio + features disagree (`rt-multi-thread+macros+time` / `test-util` / + `rt+macros`); harmless (the contract suite carries the test + runtime) but an easy-to-trip asymmetry when copying test code. + Recorded, not normalized mid-release. + +## False positives caught on verification (for the record) + +- **mem `WakeBus::attach` check-then-act race** (reported as a + finding): disproven by read — the `closed` check happens *under* the + registry guard, and `close()`'s `store(true)` is sequenced before + its guarded `clear()`, so the guard's acquire/release supplies the + happens-before edge; no interleaving leaks a subscription. No fix + needed; noted here so a future audit doesn't re-derive it. +- The "tx-path negative seed divergence" was real (fixed, item 2), but + the analogous claim *against SQLite's tx form* was not — its + in-tx SQL is already raw-seed-and-monotone; the divergence was pg's + clamp and mem's phantom-zero merge read only. + +## What was checked and clean (summary) + +- **Security**: SQL injection — every consumer-supplied string rides + bound parameters across all engines (the `format!`-built SQL + interpolates only engine-owned identifiers through the quoting + helpers; savepoint/table keys are compile-time constants); + `env::var` confined to test harness plumbing (re-grepped); + no `fs` writes outside tempdirs; no process spawning; no + secret/DSN material in any of the 17 library-path `eprintln!`s. +- **Core crate**: zero panics/unsafe/`unwrap` outside tests, its + three-dependency posture intact, `#[non_exhaustive]` placement and + the hidden-constructor seam exactly per ADR-017, object-safety and + validation coverage as pinned. (Post-fix, pre-abort-armor caveat in + item 5 above.) +- **Mem engine**: wasm32 subset verified (`cargo check` clean, tokio + sync+time only, no `spawn_blocking`/`fs`/`thread`); all deadline + arithmetic saturating; queue state machine, trim semantics, three-way + arithmetic, scheduler boundaries/catch-up/leadership re-read clean. +- **pg engine**: injection-clean, `in_tx` pool accounting exact on + every arm, forwarder reconnect invariants sound outside the recorded + TOCTOU/hang arms, `FOR UPDATE SKIP LOCKED` claim semantics and + scheduler serialization consistent with sqlite/mem. +- **Cross-cutting**: no `todo!`/`unimplemented!`/`dbg!`/`.ok()` anywhere; + metadata coherent (versions, path-dep bounds, publish fields, + `exclude = ["fuzz"]`); README/CHANGELOG claims spot-read true + against code; fuzz/shared targets thin and invariant-correct. + +## Gates re-run after the fixes (this session) + +`cargo build`, `cargo test --workspace` (green server-less), the full +live pg column (127 + 25 + 9 — including the new `in_tx_reads_see_own_writes` +negative-seed leg against the real server), `cargo clippy +--all-targets -- -D warnings`, `cargo fmt --check`, the fuzz corpus +replay (6/6 — including the corrected encode/invariant), the wasm32 +check, and the doc gate — **all green**. \ No newline at end of file diff --git a/fuzz/shared/src/payload_codec.rs b/fuzz/shared/src/payload_codec.rs index 66c58d7..74895df 100644 --- a/fuzz/shared/src/payload_codec.rs +++ b/fuzz/shared/src/payload_codec.rs @@ -166,8 +166,11 @@ fn canonicalization_ulp(x: f64) -> f64 { /// Deep inputs decode loud and typed: past serde_json's recursion /// limit, deserialization fails with `Error::Codec` — never a stack -/// overflow, never an untyped panic. The encode side handles deep -/// (but sub-limit) data as the normal round-trip. +/// overflow, never an untyped panic. The encode side refuses the same +/// depth region (typed `Codec`, never a serializer stack overflow): a +/// `Value` nested past the decode limit could never be read back from +/// the stored bytes, so storing it would be silent loss at read time. +/// Data through the encode/decode boundary is the normal round-trip. fn assert_deep_input_loudness() { for depth in [RECURSION_LIMIT * 4, 4096] { let deep_bytes = nested_array_bytes(depth); @@ -185,20 +188,33 @@ fn assert_deep_input_loudness() { } Ok(_) => panic!("deep bytes must not decode to the typed target at depth {depth}"), } - let deep_value = nest_value(Value::Null, RECURSION_LIMIT + 1); - let deep_encoded = - encode_payload(&deep_value).expect("a sub-limit-depth `Value` must encode"); - match payload_as::(&deep_encoded) { + let deep_value = nest_value(Value::Null, RECURSION_LIMIT * 4); + match encode_payload(&deep_value) { Err(Error::Codec(_)) => {} Err(other) => { - panic!("the stored deep form must decode as typed `Codec`, got {other:?}") - } - Ok(_) => { - panic!("stored bytes nested past the decode limit must decode loud, not silent") + panic!("past-limit nesting must be a typed `Codec` encode failure, got {other:?}") } + Ok(_) => panic!( + "nesting past the decode limit must never encode (the stored bytes could \ + never be read back)" + ), } } + // The encode/decode boundary (RECURSION_LIMIT - 1 containers read + // back; RECURSION_LIMIT containers fail to decode): symmetry — + // everything decodable encodes, nothing undecodable stores. + let boundary_value = nest_value(Value::Null, RECURSION_LIMIT - 1); + let boundary_encoded = + encode_payload(&boundary_value).expect("a decodable `Value` must encode"); + let back: Value = payload_as(&boundary_encoded).unwrap_or_else(|e| { + panic!("stored bytes at the encode/decode boundary must round-trip: {e:?}") + }); + assert_eq!( + back, boundary_value, + "the boundary `Value` round-trip must hold" + ); + let shallow_bytes = nested_array_bytes(8); let shallow_value = nest_value(Value::from(0u8), 8); let shallow_encoded = encode_payload(&shallow_value).expect("a shallow `Value` must encode");