---
id: pg-engine-streams
name: Postgres engine — streams (`StreamHandle`, reads, subscribe, trim)
status: pending
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

- [ ] `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`
- [ ] Reads yield `offset ASC` (global FIFO); extent guard
      (`limit <= 0` → empty `Vec`) at trait-impl entry; offsets
      immutable (trim leaves gaps, never renumbers — tested)
- [ ] `save_offset` monotone (regression save = silent no-op);
      `get_offset` reads the checkpoint; save `Err` = `Database` only
- [ ] `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
- [ ] `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
- [ ] Restart durability: publish → close store → reopen →
      `subscribe` resumes from the saved offset
- [ ] Validation + closed-store fail-closed on all entry points
- [ ] `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

> To be filled by implementation agent

## Summary

> To be filled on completion