sqlite: the listen bridge takes the stream bridge's std-thread shape (review 004 item 8) — one dedicated named thread per subscription doing blocking_send, not a spawn_blocking task (past ~500 listeners the blocking-pool ceiling queued the whole engine's seams); spawn failure unsubscribes and fails closed with Error::Database, matching the stream bridge's posture; notify.rs + engine-sqlite.md docs aligned; task -> completed
This commit is contained in:
1 parent
0fadf0f224
commit
acdef94058
3 files changed
+66
-11
No files matched your search
@@ -9,9 +9,12 @@
|
|||||||
//! never produces [`Error::PayloadTooLarge`](alkstore::Error) (ADR-016
|
//! never produces [`Error::PayloadTooLarge`](alkstore::Error) (ADR-016
|
||||||
//! §5); the payload crosses the notify table as the serde_json
|
//! §5); the payload crosses the notify table as the serde_json
|
||||||
//! serialization's UTF-8 text, the byte form ADR-020 §4 pins.
|
//! serialization's UTF-8 text, the byte form ADR-020 §4 pins.
|
||||||
//! - `listen` — the watcher-fanout bridge: one `spawn_blocking` thread
|
//! - `listen` — the watcher-fanout bridge: one dedicated `std::thread`
|
||||||
//! per subscription (`engine-sqlite.md`'s mapping row) looping
|
//! per subscription (the stream bridge's aligned shape — review 004
|
||||||
//! `recv()` on the substrate's sync feed (1-slot — bursts coalesce)
|
//! 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
|
//! and `blocking_send`-ing [`Wake`] values into a tokio channel the
|
||||||
//! [`WakeReceiver`] owns. Wakes are coalesced hints (overtrigger on
|
//! [`WakeReceiver`] owns. Wakes are coalesced hints (overtrigger on
|
||||||
//! purpose, ADR-006): the watcher fires on *any* commit, so a wake
|
//! 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)
|
/// against a store whose watcher has *died* (the db file was replaced)
|
||||||
/// fails closed the same way — a new subscription never enrolls onto a
|
/// 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).
|
/// 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(
|
pub(crate) fn listen(
|
||||||
watcher: Arc<SharedUpdateWatcher>,
|
watcher: Arc<SharedUpdateWatcher>,
|
||||||
closed: Arc<AtomicBool>,
|
closed: Arc<AtomicBool>,
|
||||||
@@ -102,13 +109,21 @@ pub(crate) fn listen(
|
|||||||
let (id, rx) = watcher.subscribe().map_err(database_error)?;
|
let (id, rx) = watcher.subscribe().map_err(database_error)?;
|
||||||
let (wake_tx, wake_rx) = tokio::sync::mpsc::channel(WAKE_CHANNEL_CAPACITY);
|
let (wake_tx, wake_rx) = tokio::sync::mpsc::channel(WAKE_CHANNEL_CAPACITY);
|
||||||
let channel = channel.to_string();
|
let channel = channel.to_string();
|
||||||
let _bridge = tokio::task::spawn_blocking(move || {
|
let spawned = std::thread::Builder::new()
|
||||||
while let Ok(()) = rx.recv() {
|
.name(format!("alkstore-listen-bridge-{channel}"))
|
||||||
if wake_tx.blocking_send(Wake::new(channel.clone())).is_err() {
|
.spawn(move || {
|
||||||
break;
|
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 {
|
Ok(Box::new(SqliteWakeReceiver {
|
||||||
wake_rx,
|
wake_rx,
|
||||||
watcher,
|
watcher,
|
||||||
|
|||||||
@@ -79,7 +79,7 @@ per-contract obligations live in the core spec and are not restated here.
|
|||||||
|
|
||||||
| Contract piece | Engine realization |
|
| 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) |
|
| 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) |
|
| 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) |
|
| 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) |
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
---
|
---
|
||||||
id: sqlite-fix-listener-bridge-alignment
|
id: sqlite-fix-listener-bridge-alignment
|
||||||
name: Align the sqlite listen bridge to the stream bridge's std-thread shape (review 004 item 8)
|
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]
|
depends_on: [sqlite-fix-watcher-death-subscribe]
|
||||||
scope: narrow
|
scope: narrow
|
||||||
risk: low
|
risk: low
|
||||||
@@ -53,6 +53,46 @@ stream bridge does.
|
|||||||
|
|
||||||
> To be filled by implementation agent
|
> 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
|
## Summary
|
||||||
|
|
||||||
> To be filled on completion
|
> 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.
|
||||||
Reference in new issue
Block a user