--- id: pg-engine-streams name: Postgres engine — streams (`StreamHandle`, reads, subscribe, trim) status: completed depends_on: [pg-engine-seam-tx, pg-engine-notify-listen] scope: moderate risk: medium impact: component level: implementation tags: [wave-4, postgres-engine] --- ## Description Implement the stream mechanism's pg arm — the full `StreamHandle` trait and the durable `subscribe` receiver, over the schema task's events/ offsets tables and the forwarder's wake substrate (ADR-015's depth, engine-postgres.md's streams mapping row): - **`publish` / `publish_with_key`** (auto-commit): pool checkout, `INSERT` into the events table (bigserial id = the assigned offset; nullable key column — carried metadata, empty-`Some`-key → `InvalidName` per the tx path's rule), `encode_payload` at the seam (`Codec` propagation), return to pool. Then **wake**: `pg_notify` on the stream's wake channel (the mechanism-name-is-the-channel realization pinned in the plan's Wave 4 section) — a separate statement on the pooled connection (auto-commit: its own implicit tx; the wake is best-effort, not commit-atomic with the insert — the no-replay hole covers a gap between them; the durable row is the truth). - **`read_since` / `read_from_consumer`**: pool reads, `ORDER BY id ASC` (global FIFO — bigserial offsets, immutable, never renumbered; ADR-015 §4's ordering row), the **extent guard** (`limit <= 0` → empty `Vec` at trait-impl entry, ADR-023 §2 — never `LIMIT -1`-class dialect semantics; pg's `LIMIT` with a negative is an error, which the guard makes unreachable). `read_from_consumer` anchors at the consumer's stored offset. - **`save_offset` / `get_offset`**: the monotone upsert (a save below the stored checkpoint is a silent no-op — `GREATEST`-guarded upsert or `WHERE offset > stored`; one owner of the monotonicity rule) and the checkpoint read. Both save forms' `Err` arm carries `Database` only (ADR-021 §5). - **`trim_to(horizon)`**: `DELETE … WHERE id <= horizon` on a pool connection (wakes nothing — ADR-015 §5's posture; the boundary argument is total: a negative horizon deletes nothing, idempotently), returning the deleted count. Surviving events keep their offsets (gaps legal). - **`subscribe(consumer)`** — the durable receiver, the structural twin of the SQLite engine's bridge with the pg deltas: - Attach-before-read ordering (subscribe to the stream's wake channel *before* the initial cursor read — a commit cannot fall between attach and wait). - Initial cursor read of the consumer's *stored* offset; page-drain to the tail (`offset ASC`); park on the wake feed (the forwarder fanout for this stream's channel); re-drain on wake (wake-driven with table re-read — events never require polling to become visible, core-contract.md's streams section). - **Close semantics (the pg arm, ADR-021 §5)**: the receiver stays open across forwarder reconnects — reconnect-wakes arrive on the wake feed and simply trigger re-drains (which is the recovery: the no-replay hole's gap is healed by the re-drain reading the durable rows); closes only at engine shutdown (terminal `None`). - **Position semantics**: `position` = the last-*yielded* event's offset (0 before any yield ⇒ `save_offset()` before any yield is a no-op, ADR-021 §5); the bridge's internal cursor advances on drained-from-storage events — a stalled consumer's `save_offset()` never checkpoints unobserved events. Backpressure: bounded channel + backpressure-aware send. - `recv()`'s `Err` carries `Database` only (ADR-021 §5 — remap non-`Database` read errors with the original as source, the SQLite `database_only` helper's shape); `try_recv` arms mirror the wake receiver (idle `Ok(None)`, `Err(Closed)` after shutdown-close). - The receiver's `read_since` (anchored, extent-guarded) and sync `save_offset` (the last-yielded offset through the same monotone op) per ADR-019 §6. - **Validation**: stream names via `validate_shared_name` at `store.stream()` and (defensively) the handle's reads; consumers via `validate_local_name` (non-empty only — reserved prefixes legal for consumers). - **Closed-store posture**: every handle op checks the store's closed flag first and fails closed with `Database` (the wave-3 shape). ## Acceptance Criteria - [x] `publish`/`publish_with_key` return the assigned bigserial offset; key round-trips (`None` vs `Some`); `encode_payload`'s `Codec` propagates; empty-`Some`-key rejected `InvalidName` - [x] Reads yield `offset ASC` (global FIFO); extent guard (`limit <= 0` → empty `Vec`) at trait-impl entry; offsets immutable (trim leaves gaps, never renumbers — tested) - [x] `save_offset` monotone (regression save = silent no-op); `get_offset` reads the checkpoint; save `Err` = `Database` only - [x] `trim_to` exact-boundary (`<=`), returns deleted count, wakes nothing, negative horizon deletes nothing; reads resume at the horizon's first remaining row; saved offsets below the horizon stay valid - [x] `subscribe`: replay from stored offset, wake-driven delivery, position = last-yielded offset, backpressure holds, receiver stays open across forwarder reconnects (reconnect-wake triggers re-drain — the gap-healing property tested), terminal close at engine shutdown only - [x] Restart durability: publish → close store → reopen → `subscribe` resumes from the saved offset - [x] Validation + closed-store fail-closed on all entry points - [x] `cargo test -p alkstore-postgres` (harness server), clippy `-D warnings`, fmt clean; gates green server-less ## References - docs/architecture/core-contract.md (streams section) - docs/architecture/decisions/015-streams-depth.md - docs/architecture/decisions/019-mechanism-handle-surfaces.md §6 - docs/architecture/decisions/021-tx-reads-and-value-shape-fixes.md §5 - docs/architecture/decisions/023-fourth-review-round.md §2 (extent guard) - docs/architecture/engine-postgres.md (Mapping the contract: streams) - alkstore-sqlite/src/stream.rs (the structural twin) - alkstore/src/stream_handle.rs, event_receiver.rs (the traits) ## Notes > Decisions of record the implementation made that the description > didn't pin: - **The wake statement is `SELECT pg_notify($1, '')` on the same pooled connection, posted-insert, best-effort**: auto-commit — the notify fires at its own implicit tx's commit, a separate statement per the description (the gap between insert-commit and wake-commit is the no-replay hole; the durable row is the truth). A wake failure is logged to stderr and swallowed — the durable row already landed, and a wake-fee must never fail (or stall) a publish. - **The subscribe receiver is a natively-async bridge task** (the structural twin of the SQLite bridge, without the std thread): a tokio task holds a broadcast subscription (taken before the attach read — attach-before-read ordering), drains pages to the tail, parks on `broadcast::recv`, re-drains on wake (including the lagged arm — lagged is woken-state too, surfaced-by-drain). Channel capacity 16 (the SQLite twin's smoothing-buffer shape); `send().await` is backpressure-aware — a stalled consumer parks the bridge, delivery resumes exactly where it stopped. - **`EventReceiver::save_offset`'s sync signature needed a real sync→async bridge** (this engine has no blocking writer slot to lease — the SQLite twin's sync posture does not exist here). Probed exhaustively: `Handle::block_on` panics on a runtime worker (both flavors), `Runtime::new().block_on` nested panics and would build a runtime per call, `block_in_place` + `block_on` of a *different* runtime is safe only on multi-thread workers, and a plain (non-runtime) thread can block directly. Landed shape: a lazily built, process-wide, stateless one-worker idle "wake runtime" (`stream.rs`'s `WAKE_RUNTIME`); `drive_sync` runs the save future `block_on` it — directly on the caller's thread when outside any runtime or inside a foreign `spawn_blocking`, via `block_in_place` on a multi-thread worker, via a scoped std thread on a current-thread worker (the only escape there). The runtime carries no engine state; each save checks out its own pool connection *on the runtime that drives it*. Idle cost: one parked runtime (one blocked driver thread). - **The bridge spawn failure arm is `is_finished()`-detected** (+ abort/`Database`): `tokio::spawn` cannot fail directly — a shut-down runtime drops the task before it runs, so the spawn site checks finished-ness and fails the subscribe closed (no boxed receiver surfacing as a lie, the W-2 posture's pg shape). - **Decode ownership moved**: `stream_events_from_rows` moved from `tx.rs` into `stream.rs` as the one `pub(crate)` decode owner (the tx reads import it — the SQLite twin's decode-sharing shape, the `stream` field mapping and the `Codec` posture guard live once). `NOTIFY_PAYLOAD_LIMIT` is not needed here — the wake payload is the empty string (stream wakes carry no payload; the durable row is the truth; the 8000-byte budget never binds). - **`trim_to` is a plain parameterized DELETE on a pool connection** (wakes nothing — no notify statement at all; ADR-015 §5). The boundary argument needs *no* guard on SQL: `id <= -7` matches no rows server-side, idempotently (the totality is native here — contrast the SQLite arm's writer-lease). - **The wake-feed arm shape in `run_subscribe_loop`**: store-close is checked at every read arm (drain, attach) and exits the loop (`PageArm::Closed`) — the consumer sees the terminal close via the channel disconnect; a mid-drain error while open surfaces one `Err(Database)` and waits for the next wake (the SQLite twin's transient arm, verbatim). `database_only` remaps non-`Database` errors (the decode-side `Codec` guard) into the opaque fallback with the source preserved. - **The `closed_check` helper from the store is not reused on the handle** (the handle checks its own `closed` arc inline at each op entry — the ops return from sync context into boxed futures, so the check happens before the `Box::pin`, matching the SQLite twin's placement). - **Test-infra note**: the "wakes nothing" and gap-heal pins use the kill_listener_backend shape from the notify tests (the reconnect-wake then re-drain cycle is what heals the gap); the drop-unregister pin probes the raw forwarder fanout (no cross-session LISTEN catalog view exists — the notify tests' finding reused). The trimmed-horizon reads-resume assertion pins the first remaining row explicitly (`offsets[2]`). ## Summary > What landed, verified how: - **`alkstore-postgres/src/stream.rs`** (new): the stream mechanism's pg arm — `PgStreamHandle` (the full `StreamHandle` trait: `publish`/`publish_with_key` auto-commit INSERT .. RETURNING id on a pool connection with `encode_payload` at the seam and the best-effort `pg_notify` wake on the stream's channel after the insert (separate statement, failure logged-and-swallowed), `read_since`/`read_from_consumer` with the ADR-023 §2 extent guard and `id ASC` ordering, monotone `save_offset`/`get_offset` (the shared upsert owner), `trim_to`'s pool-connection DELETE) and the durable `subscribe` receiver (`PgEventReceiver` + the async bridge task: attach-before-read, replay from the stored offset, wake- driven re-drains — reconnect-wakes included, terminal close at engine shutdown only, position = last-yielded offset, backpressure- aware delivery, sync `save_offset` bridged through the stateless idle wake runtime, `recv()` Err = `Database` only). - **`tx.rs`**: now imports `stream_events_from_rows` from `stream.rs` (the one decode owner; the forked copy deleted). - **`store.rs`**: `Store::stream` wired (validation + closed-store fail-closed + handle construction); the open task's stub-surface test updated (`stream` asserted wired, `queue` the remaining-stub representative). - **Tests** (`store/stream_tests.rs`, 21 new): the round-trip (offsets/stream-field/keys/decode), cross-store wake delivery, the extent guard (0/-1/MIN on both read forms), trim (negative/zero/ exact-boundary/idempotent/no-renumbering/reads-resume/saved-offset- survives/wakes-nothing), monotone saves across forms (incl. the receiver pre-yield no-op), replay + wake-driven delivery, gap-heal across a real backend-kill reconnect (the gap event arrives via the re-drain; the receiver never closes), terminal shutdown close (`None` + `Err(Closed)` + never reopens), tail-idle, independent subscribers, receiver-drop UNLISTEN (raw-fanout probe), key round-trip on every read form, byte-exact stored payload (the raw BYTEA row vs the serde_json serialization), tx-compose (commit/ghost/monotone-tx-save), receiver `read_since` anchoring + extent guard, stalled-consumer saves-only-observed, backpressure (48-event burst delivered exactly in ASC order), entry-point validation (shared kinds + empty-Some keys + reserved-prefix-local consumers + residue-free rejects), and closed-store fail-closed on every op + constructors. - **Verification**: `cargo test -p alkstore-postgres` against the harness server (pglo-poc :15432): 72 lib + 9 schema tests green, repeated ×4 (the stream subset 4× stable); workspace `cargo test` green server-less (pg tests skip per convention, 10 result lines all ok); `cargo clippy --all-targets -- -D warnings` clean; `cargo fmt --check` clean.