From acdef9405820da1e018d245a11991dfedef0053a Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sun, 11 Oct 2026 10:50:24 +0000 Subject: [PATCH] =?UTF-8?q?sqlite:=20the=20listen=20bridge=20takes=20the?= =?UTF-8?q?=20stream=20bridge's=20std-thread=20shape=20(review=20004=20ite?= =?UTF-8?q?m=208)=20=E2=80=94=20one=20dedicated=20named=20thread=20per=20s?= =?UTF-8?q?ubscription=20doing=20blocking=5Fsend,=20not=20a=20spawn=5Fbloc?= =?UTF-8?q?king=20task=20(past=20~500=20listeners=20the=20blocking-pool=20?= =?UTF-8?q?ceiling=20queued=20the=20whole=20engine's=20seams);=20spawn=20f?= =?UTF-8?q?ailure=20unsubscribes=20and=20fails=20closed=20with=20Error::Da?= =?UTF-8?q?tabase,=20matching=20the=20stream=20bridge's=20posture;=20notif?= =?UTF-8?q?y.rs=20+=20engine-sqlite.md=20docs=20aligned;=20task=20->=20com?= =?UTF-8?q?pleted?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- alkstore-sqlite/src/notify.rs | 33 ++++++++++---- docs/architecture/engine-sqlite.md | 2 +- tasks/sqlite-fix-listener-bridge-alignment.md | 44 ++++++++++++++++++- 3 files changed, 67 insertions(+), 12 deletions(-) diff --git a/alkstore-sqlite/src/notify.rs b/alkstore-sqlite/src/notify.rs index 21f79ea..1fcff24 100644 --- a/alkstore-sqlite/src/notify.rs +++ b/alkstore-sqlite/src/notify.rs @@ -9,9 +9,12 @@ //! never produces [`Error::PayloadTooLarge`](alkstore::Error) (ADR-016 //! §5); the payload crosses the notify table as the serde_json //! serialization's UTF-8 text, the byte form ADR-020 §4 pins. -//! - `listen` — the watcher-fanout bridge: one `spawn_blocking` thread -//! per subscription (`engine-sqlite.md`'s mapping row) looping -//! `recv()` on the substrate's sync feed (1-slot — bursts coalesce) +//! - `listen` — the watcher-fanout bridge: one dedicated `std::thread` +//! per subscription (the stream bridge's aligned shape — review 004 +//! item 8; not a tokio `spawn_blocking` per subscription, which +//! would queue the engine's seams behind parked bridges at tokio's +//! blocking-pool ceiling) looping `recv()` on the substrate's sync +//! feed (1-slot — bursts coalesce) //! and `blocking_send`-ing [`Wake`] values into a tokio channel the //! [`WakeReceiver`] owns. Wakes are coalesced hints (overtrigger on //! purpose, ADR-006): the watcher fires on *any* commit, so a wake @@ -88,6 +91,10 @@ pub(crate) fn notify( /// against a store whose watcher has *died* (the db file was replaced) /// fails closed the same way — a new subscription never enrolls onto a /// dead feed to park forever (review 004 item 2's new-subscriber arm). +/// The bridge runs on a dedicated `std::thread` (the stream bridge's +/// shape — review 004 item 8); a thread-spawn failure unsubscribes and +/// fails closed with [`Error::Database`](alkstore::Error) (the stream +/// bridge's posture). pub(crate) fn listen( watcher: Arc, closed: Arc, @@ -102,13 +109,21 @@ pub(crate) fn listen( let (id, rx) = watcher.subscribe().map_err(database_error)?; let (wake_tx, wake_rx) = tokio::sync::mpsc::channel(WAKE_CHANNEL_CAPACITY); let channel = channel.to_string(); - let _bridge = tokio::task::spawn_blocking(move || { - while let Ok(()) = rx.recv() { - if wake_tx.blocking_send(Wake::new(channel.clone())).is_err() { - break; + let spawned = std::thread::Builder::new() + .name(format!("alkstore-listen-bridge-{channel}")) + .spawn(move || { + while let Ok(()) = rx.recv() { + if wake_tx.blocking_send(Wake::new(channel.clone())).is_err() { + break; + } } - } - }); + }); + if let Err(e) = spawned { + watcher.unsubscribe(id); + return Err(Error::database(std::io::Error::other(format!( + "failed to spawn listen bridge thread: {e}" + )))); + } Ok(Box::new(SqliteWakeReceiver { wake_rx, watcher, diff --git a/docs/architecture/engine-sqlite.md b/docs/architecture/engine-sqlite.md index 2a50d89..031ed5f 100644 --- a/docs/architecture/engine-sqlite.md +++ b/docs/architecture/engine-sqlite.md @@ -79,7 +79,7 @@ per-contract obligations live in the core spec and are not restated here. | Contract piece | Engine realization | |---|---| -| notify / listen | honker's notify functions inside the caller's tx; `listen()` bridges the watcher's fanout into a tokio receiver (one `spawn_blocking` thread per subscription doing `blocking_send`) | +| notify / listen | honker's notify functions inside the caller's tx; `listen()` bridges the watcher's fanout into a tokio receiver (one dedicated `std::thread` per subscription doing `blocking_send` — the stream bridge's aligned shape, review 004 item 8; the same thread is not a `spawn_blocking` task, so parked bridges never consume the blocking pool the engine's seams share) | | streams | the forked substrate's stream machinery (inherited near-verbatim — the `__alkstore_stream` table's field set is already the contract's shape (its `topic` column carries the contract's `stream` field — the one name delta, ADR-012 §3's fidelity posture), the offset-ASC read path, keyed publish's nullable key column, monotone offset saves ([ADR-015](decisions/015-streams-depth.md)); explicit offset saves through the tx seam; `trim_to` is a `DELETE FROM … WHERE offset <= ?` inside the writer-slot lease) | | queues | owned queue machinery (forked substrate re-derived on [ADR-010](decisions/010-queue-semantics-depth.md) — stamps, no-stranded-rows sweep, dead-visible `get_job`); ADR-003's ride posture superseded on ownership by [ADR-011](decisions/011-sqlite-substrate-fork.md) | | named locks | the substrate's lock machinery (`lock_renew` carries the renew semantics the guarantee row pins; re-acquire does not refresh TTL — inherited deliberately, [ADR-012](decisions/012-forked-substrate-design.md) §4) | diff --git a/tasks/sqlite-fix-listener-bridge-alignment.md b/tasks/sqlite-fix-listener-bridge-alignment.md index c30b10c..9e40f39 100644 --- a/tasks/sqlite-fix-listener-bridge-alignment.md +++ b/tasks/sqlite-fix-listener-bridge-alignment.md @@ -1,7 +1,7 @@ --- id: sqlite-fix-listener-bridge-alignment name: Align the sqlite listen bridge to the stream bridge's std-thread shape (review 004 item 8) -status: pending +status: completed depends_on: [sqlite-fix-watcher-death-subscribe] scope: narrow risk: low @@ -53,6 +53,46 @@ stream bridge does. > To be filled by implementation agent +Decisions of record the description didn't pin: + +- The two bridges' shapes are now exactly aligned: a named + `std::thread::Builder` thread (`alkstore-listen-bridge-{channel}`, + mirroring the stream bridge's `alkstore-stream-bridge-{stream}`), the + same loop body, and the stream bridge's spawn-failure posture — a + failed spawn unsubscribes the just-enrolled substrate feed and fails + closed with `Error::Database` (the stream bridge's W-2 no-panic- + across-the-seam shape). The spawn-failure arm is new here + (`spawn_blocking` was infallible at spawn); it matches the stream + bridge rather than being named as a residual difference. +- Docs updated, not code-adjacent only: `notify.rs`'s module docs + (the bridge bullet + the `listen` entry-point doc now pin the + std-thread shape and the spawn-failure arm), and `engine-sqlite.md`'s + notify/listen mapping row (rewritten to the dedicated-thread shape + with the alignment reason). No PROVENANCE row: the register covers + `src/substrate/` only, and this change is engine-level (`notify.rs`). +- No new test: the shape is not contract-observable at the receiver + surface (close semantics, wake delivery, no-replay all unchanged); + the existing pins cover the observable posture, including the + dead-watcher sign from the dependency task + (`listen_after_watcher_death_fails_closed` — green unchanged). + ## Summary -> To be filled on completion \ No newline at end of file +> To be filled on completion + +`listen()`'s watcher-fanout bridge now runs on a dedicated named +`std::thread` instead of a tokio `spawn_blocking` task per +subscription — the stream bridge's shape, eliminating the blocking-pool +ceiling queueing (review 004 item 8); parked listen bridges no longer +consume the shared blocking pool the engine's seams run on. Spawn +failure fails closed (`Database` + unsubscribe), matching the stream +bridge. Doc alignment: `notify.rs` module + entry-point docs, +`engine-sqlite.md`'s mapping row. + +Verified: `cargo build`, `cargo test` (workspace, all green), +`cargo test -p alkstore-sqlite` (201 + 25 tests, green — close +semantics / wake delivery / no-replay pins and the dead-watcher-sign +tests unchanged and passing), fuzz gate +(`cargo test --manifest-path fuzz/shared/Cargo.toml`), clippy +`-D warnings` clean, `cargo fmt --check` clean, rustdoc `-D warnings` +doc gate clean. \ No newline at end of file