From 4caea21098d0d85e1338d516599aca50314b79d6 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Thu, 8 Oct 2026 22:49:00 +0000 Subject: [PATCH] =?UTF-8?q?Wave=204=20decomposed:=20Postgres=20engine=20?= =?UTF-8?q?=E2=80=94=2011=20tasks=20(schema,=20open/opts,=20seam,=20forwar?= =?UTF-8?q?der,=20mechanisms,=20scheduler/outbox,=20integration,=20review?= =?UTF-8?q?=20gate);=20plan=20synced?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/plans/implementation.md | 90 +++++++++++++++- tasks/pg-engine-integration.md | 101 ++++++++++++++++++ tasks/pg-engine-locks.md | 93 +++++++++++++++++ tasks/pg-engine-notify-listen.md | 139 +++++++++++++++++++++++++ tasks/pg-engine-open-opts.md | 113 ++++++++++++++++++++ tasks/pg-engine-queues.md | 141 +++++++++++++++++++++++++ tasks/pg-engine-scheduler-outbox.md | 154 ++++++++++++++++++++++++++++ tasks/pg-engine-schema.md | 110 ++++++++++++++++++++ tasks/pg-engine-seam-tx.md | 129 +++++++++++++++++++++++ tasks/pg-engine-streams.md | 126 +++++++++++++++++++++++ tasks/review-wave-4.md | 94 +++++++++++++++++ 11 files changed, 1287 insertions(+), 3 deletions(-) create mode 100644 tasks/pg-engine-integration.md create mode 100644 tasks/pg-engine-locks.md create mode 100644 tasks/pg-engine-notify-listen.md create mode 100644 tasks/pg-engine-open-opts.md create mode 100644 tasks/pg-engine-queues.md create mode 100644 tasks/pg-engine-scheduler-outbox.md create mode 100644 tasks/pg-engine-schema.md create mode 100644 tasks/pg-engine-seam-tx.md create mode 100644 tasks/pg-engine-streams.md create mode 100644 tasks/review-wave-4.md diff --git a/docs/plans/implementation.md b/docs/plans/implementation.md index 822763b..3fe0a51 100644 --- a/docs/plans/implementation.md +++ b/docs/plans/implementation.md @@ -1,6 +1,6 @@ --- status: draft -last_updated: 2026-10-08 (wave 3 implemented + reviewed; review-gate fixes: substrate boundary move, writer-slot error-arm release; wave 4/5 decomposition may proceed) +last_updated: 2026-10-08 (wave 4 decomposed — 11 tasks; waves 1–3 implemented + reviewed) --- # alkstore — Implementation plan @@ -35,7 +35,7 @@ parallel). | [1](#wave-1--foundations) | Workspace scaffold; core crate (errors, value types, full trait surface); contract-suite scaffold | — | implemented + reviewed (2026-10-08) | | [2](#wave-2--sqlite-substrate-fork) | honker-core fork into `alkstore-sqlite/src/substrate/`: port, deltas, provenance, test floor | wave 1 (workspace scaffold only) | implemented + reviewed (2026-10-08) | | [3](#wave-3--sqlite-engine) | SQLite engine: connection architecture, re-derived queue ops on contract v1, scheduler/outbox, tx seam, SQLite backlog column | waves 1 + 2 | implemented + reviewed (2026-10-08) | -| 4 | Postgres engine: schema bootstrap, pool/open, listener/forwarder, all mechanisms, tx seam, pg backlog column | wave 1 | not yet decomposed | +| 4 | [Postgres engine](#wave-4--postgres-engine): schema bootstrap, pool/open, listener/forwarder, all mechanisms, tx seam, pg backlog column | wave 1 | decomposed (11 tasks) | | 5 | Contract suite: the cross-engine equivalence properties (core-contract.md §Verification backlog), version-stamped per ADR-017 | waves 3 + 4 | not yet decomposed | | 6 | Release readiness: crate docs, deployment matrix final pass, README (written last, honestly), publish prep; mem-engine and fuzzing decisions | wave 5 | not yet decomposed | @@ -101,6 +101,86 @@ shaped this wave's task set: panic probe → wave 5's contract-suite rows; N-1's growth-posture review → wave 6. +## Wave 4 — Postgres engine + +The second engine: the core contract implemented on tokio-postgres + +deadpool-postgres per engine-postgres.md — schema bootstrap (one +engine-owned schema, default `alkstore`, ADR-010 §8), the pool/open +constructor, the hand-rolled LISTEN forwarder (dedicated non-pooled +listener connection, bounded broadcast fanout, exponential-backoff +reconnect with re-LISTEN, the synthetic reconnect-wake on +`__alkstore_listener_reconnected__`), the full `Store`/`TxHandle`/ +mechanism-handle trait impls (natively async — no `spawn_blocking` +seam; the tx handle holds the pooled object directly, ADR-007's pg +arm), the re-derived queue machinery (`FOR UPDATE SKIP LOCKED` claims, +LISTEN-driven wake with re-poll safety net), the scheduler leader loop +and outbox helper, and the engine's backlog column in the contract +suite (factory-parameterized; wave 5 runs the same rows against both +engines). + +The wave is deliberately shaped as wave 3's structural twin — same +task rhythm (constructor → seam → mechanisms in parallel → +scheduler/outbox → integration → review gate) — because the two +engines implement the same contract and the wave-3 task set's +boundaries proved clean. The deltas are the engine's own: no +substrate (the pg machinery is re-derived greenfield per ADR-004/005, +with the pg-boss schema family as design reference), a natively-async +seam (Send+Sync client, POC #2 compile-probe), the forwarder as the +wake substrate (with its two POC-pinned deadlock pitfalls owned as +test-pinned failure modes), and the `PayloadTooLarge` client-side +check (this engine *produces* the variant — the one runtime carriage +of an engine asymmetry, ADR-016 §5). + +Wave-3 outcomes absorbed into the task set: + +- **The `#[doc(hidden)]` constructors** (`Job::from_row`, + `StreamEvent::from_row`, `Schedule::new`, `Wake::new`) already + exist — wave 3's pre-work task built them engine-agnostically; wave + 4 consumes them as-is (no wave-4 equivalent needed). +- **The engine-side arithmetic is re-owned, not shared**: the + equal-jitter backoff curve, the opts-resolution rules + (delay-over-`run_at`, relative-`expires`, the derived stamp sets + 300/3/5/none and 60/5/5), and the outbox backing-queue derivation + are each engine's own implementation (ADR-012 §2's one-owner rule + applied per engine); the pg tasks re-implement them from the ADR + text, and the contract suite (wave 5) pins the two engines' + arithmetic to identical outputs. The SQLite engine's `resolution.rs` + is reference, not a dependency. +- **The `@every` spec grammar has no shared parser on this side** — + the SQLite engine reuses the substrate's `parse_every_interval` + (one-line delta D-31's neighbor); the pg engine owns its parser + (the grammar is small and pinned: `@every `, `s|m|h|d`). + Equivalence of *behavior* (what fires, when) is suite-pinned; the + parsers themselves are per-engine. +- **The wave-3 review's deferred notes** (reader-pool close path, + commit-error-arm coverage) stay SQLite-scoped — the pg engine's + pool is deadpool-managed (no hand-rolled close path to review) and + its commit-error arm gets coverage in the pg tasks' own tests. +- **The suite's factory contract** (isolation + idempotent teardown) + is what the pg factory implements — a fresh schema per `open` per + the SQLite factory's precedent (fresh temp file there); the + POC's shared-server parallel-interference caveat (findings, + invocation note) is answered by exactly that isolation, not by + sequential-only harnesses. + +Test posture: the engine's tests need a live Postgres server. The +harness convention is the POC's (dockerized `postgres:16-alpine` on +:15432, the `pglo-poc` container — still running in the dev +environment); connection settings ride environment/config, never +hardcoded. Tests that cannot reach a server skip cleanly (the +`SKIP`-posture), so the workspace gates stay green server-less — but +the wave's acceptance requires the suite green *against* the server. + +One design point the architecture leaves to implementation (pinned in +the wave-4 tasks, flagged here for visibility): **the queue/stream +wake-channel naming**. The contract pins LISTEN as the pg wake +trigger and pins the channels/queues shared namespace, but no doc +names the channel an enqueue/publish notifies on. The tasks pin the +natural realization — the mechanism name *is* the wake channel +(`listen(queue_name)` receives that queue's enqueue wakes; same for +streams), legal by construction under the shared namespace — with +the reserved reconnect-wake channel staying the only reserved string. + ## Decided points - **Contract-suite layout — option (a)**: a small internal @@ -159,4 +239,8 @@ decomposes. Specific gates: `open_writer_connection` boundary move out of `substrate/mod.rs`; the writer-slot error-arm stranding in `with_writer`/`begin`/`commit`); two minor notes deferred to - wave 5/6 (reader-pool close path, commit-error-arm coverage). \ No newline at end of file + wave 5/6 (reader-pool close path, commit-error-arm coverage). +- **Wave 4 decomposition** (2026-10-08) — shaped as wave 3's + structural twin; wave-3 outcomes absorbed (constructors exist; + arithmetic re-owned per engine; per-engine `@every` parser; + SQLite-scoped deferred notes stay put). See the Wave 4 section. \ No newline at end of file diff --git a/tasks/pg-engine-integration.md b/tasks/pg-engine-integration.md new file mode 100644 index 0000000..35b12d7 --- /dev/null +++ b/tasks/pg-engine-integration.md @@ -0,0 +1,101 @@ +--- +id: pg-engine-integration +name: Postgres engine — integration (full-surface wiring, suite adoption, crate docs) +status: pending +depends_on: [pg-engine-notify-listen, pg-engine-streams, pg-engine-queues, pg-engine-locks, pg-engine-scheduler-outbox] +scope: moderate +risk: medium +impact: phase +level: implementation +tags: [wave-4, postgres-engine, integration] +--- + +## Description + +Close the wave's integration gaps once every mechanism is wired — the +structural twin of wave 3's integration task (no lint-suppression +removal on this side: the pg engine is greenfield, no `#![allow]` +posture ever existed — the sweep below is the same discipline without +that step): + +- **Full-surface sweep**: every `Store`/`TxHandle`/mechanism trait + method implemented (no `todo!`/`unimplemented!`/error-stub + remaining); the engine crate's lib re-exports its public surface + (`open`, `PgOpts`, the concrete store type) per the module-per-file + convention. Genuinely dead code must be **cut, not suppressed** — + clippy `-D warnings` green without allows. +- **Contract-suite adoption** (the plan's "waves 3 and 4 adopt the + harness"): implement `StoreFactory` for the pg engine — **a fresh + schema per `open`** (the SQLite factory's fresh-temp-file + precedent; the POC's shared-server parallel-interference caveat is + answered by exactly this isolation — each property's rows live in + its own schema, so parallel property runs never observe one + another), idempotent teardown (`DROP SCHEMA … CASCADE`; safe + against an abandoned store per the factory contract's teardown + posture). Pin the factory's isolation/idempotence contract (the + SQLite factory's pin shape). + Wire the engine's backlog column — the rows this engine owns now, + running against the pg factory: + - The ADR-023-stamped rows (extent-clamp, duration-refusal, + `encode_payload` round-trip) — already written suite-side, + factory-parameterized; they run against pg by this task's + factory existing (verify, don't rewrite). + - **The pg arm of the engine-scoped rows**: `PayloadTooLarge` + produced (the pg rejection arm — oversized `notify`/`notify_tx` + rejected client-side with the limit in the variant; the row text + the SQLite integration task left for "wave 4's adoption"), + drop = rollback no-ghosts, in-tx read-your-own-writes, + enqueue-opts resolution, receiver close arms (the pg arm: stays + open across reconnects, closes at shutdown — the row's + engine-asymmetric legs may need the row split or parameterized + per the suite's convention; implementer's choice, documented). + - The SQLite-scoped rows (plain-path open §3, poll cadence §4) + simply don't run against pg — the SQLite integration task's + Notes already pinned that disposition. +- **Crate docs**: the engine crate's lib-level doc comments state the + multi-host posture and the listener budget line (ADR-016, + deployment.md) — the identity statements this crate owns (the open + task drafted them; this task verifies they reflect the finished + engine). +- **Gates**: full workspace build/test/clippy/fmt (the pg tests skip + cleanly server-less but must be green against the harness server); + coverage spot-check on the engine crate (no large uncovered regions + outside error arms — the wave-3 gate's bar). + +## Acceptance Criteria + +- [ ] No `todo!`/`unimplemented!`/error-stub in the engine crate; + full trait surface implemented; clippy `-D warnings` green + without allows +- [ ] `StoreFactory` implemented for pg (fresh schema per `open`, + idempotent CASCADE teardown); the factory's + isolation/idempotence contract pinned +- [ ] The backlog column runs green against the pg factory: the three + ADR-023 rows verified (not rewritten), the pg arm of the five + engine-scoped rows present + stamped +- [ ] The `PayloadTooLarge` pg arm row present (the SQLite + integration task's deferred adoption discharged) +- [ ] Crate lib docs carry the multi-host + listener-budget posture + statements, reflecting the finished engine +- [ ] Workspace gates green: `cargo build`, `cargo test` (incl. + against the harness server), clippy `-D warnings`, fmt clean +- [ ] Coverage spot-check: no large uncovered regions outside error + arms + +## References + +- docs/plans/implementation.md (Wave 4 section) +- docs/architecture/core-contract.md (§Verification backlog) +- docs/architecture/decisions/022-contract-suite-layout.md +- docs/architecture/decisions/023-fourth-review-round.md (backlog additions) +- docs/architecture/decisions/016-deployment-honesty.md §5 +- alkstore-sqlite/tests/contract_suite.rs (the factory precedent) +- alkstore-contract-suite/src/properties.rs (the rows to run/adopt) + +## Notes + +> To be filled by implementation agent + +## Summary + +> To be filled on completion \ No newline at end of file diff --git a/tasks/pg-engine-locks.md b/tasks/pg-engine-locks.md new file mode 100644 index 0000000..1f8bcad --- /dev/null +++ b/tasks/pg-engine-locks.md @@ -0,0 +1,93 @@ +--- +id: pg-engine-locks +name: Postgres engine — named locks (`try_lock`, `Lock` handle, duration guards) +status: pending +depends_on: [pg-engine-seam-tx] +scope: narrow +risk: low +impact: component +level: implementation +tags: [wave-4, postgres-engine] +--- + +## Description + +Implement the named-lock mechanism's pg arm — the smallest mechanism, +and the one with the strongest POC ground (the pg POC's lock probe +pinned acquire/release/renew with TTL and expiry re-acquisition +directly). Over the schema task's locks table: + +- **`try_lock(name, owner, ttl)`** — shared-namespace + non-empty + owner validation (`InvalidName`); the **duration guard** (`ttl <= + 0` → opaque `Err(Database)`, detail in the source chain — ADR-023 + §2's duration kind; an exclusion mechanism must not silently hand + back a lease the caller believes it holds); the closed-store check; + then the acquire op on a pool connection: opportunistic-expiry- + delete + insert-or-reacquire + read-back (the substrate's + `lock_acquire` semantics, re-derived — the POC's probe shape). + `Ok(None)` = held elsewhere (no-work is a value); `Ok(Some(boxed + Lock))` on grant. +- **`Lock` handle** (`renew`, consuming `release`): + - `renew(ttl)` — the same duration guard; the new **full TTL + window from now** (not additive); `false` = lost it (expired and + re-acquired by another owner — the owner-scoped update's 0-rows). + - `release(self: Box)` — consuming; the owner-scoped delete; + `true` = was held (row gone by this owner's hand), `false` = + row-less or foreign (the substrate's row-scoped truth-teller + posture — no handle-local "held" flag caching). +- **Guarantee row posture** (ADR-008 §7): mutual exclusion bounded by + TTL + renewal; expiry lapses *silently* (no revocation event — + expired rows are deleted opportunistically by the next acquire's + expiry-delete, no general expiry sweeper — the inherited posture, + documented); re-acquirable after expiry (the POC-pinned property — + pin it engine-side here; the contract-suite row remains for the + suite itself). +- **Same-owner re-acquire**: pin the inherited disposition (the + substrate's shape: the same owner's re-acquire keeps the original + row, TTL *not* refreshed, reports granted — the SQLite task's + pinned posture; engine-sqlite.md's mapping row calls it inherited + deliberately, and the pg arm should match it for cross-engine + equivalence — flag in Notes if the pg realization differs). +- **Post-lapse release arms**: pin both (the stale holder's release + deletes its own stale row → `true`; after a foreign re-acquire → + `false`) — the SQLite task's pinned shape, matched here. +- **Closed-store posture**: `try_lock`/`renew`/`release` fail closed + with opaque `Database` (the engine-wide shape). +- **No `lock_tx`** — deliberate absence (core-contract.md's named- + locks section); nothing to build, just don't invent one. + +## Acceptance Criteria + +- [ ] `try_lock`: validation (name + owner), duration guard + (`ttl <= 0` → opaque `Err`, detail in the source chain), + closed-store check, then acquire — `Ok(None)` on held-elsewhere +- [ ] Exclusion + loser shape pinned (two owners, one wins); expiry + re-acquisition pinned (the POC's property, engine-side) +- [ ] `renew`: duration guard; full-window reset; `false` on + lost/expired-and-reacquired +- [ ] `release`: consuming, owner-scoped, both boolean arms pinned + (stale-row delete = `true`; foreign/row-less = `false`) +- [ ] Same-owner re-acquire disposition pinned (and flagged if it + differs from the SQLite arm's) +- [ ] Silent lapse posture documented (no revocation event; no + general expiry sweeper) +- [ ] `cargo test -p alkstore-postgres` (harness server), clippy + `-D warnings`, fmt clean; gates green server-less + +## References + +- docs/architecture/core-contract.md (named locks section) +- docs/architecture/decisions/008-contract-v1-pinning.md §7 (guarantee row) +- docs/architecture/decisions/023-fourth-review-round.md §2 (duration kind) +- docs/architecture/engine-postgres.md (Mapping the contract: named locks) +- docs/research/poc-pg-posture-findings.md (the lock probe row) +- alkstore-sqlite/src/lock.rs (the structural twin — match its pinned dispositions) +- alkstore/src/lock.rs (the trait) + +## Notes + +> To be filled by implementation agent + +## Summary + +> To be filled on completion \ No newline at end of file diff --git a/tasks/pg-engine-notify-listen.md b/tasks/pg-engine-notify-listen.md new file mode 100644 index 0000000..bb33368 --- /dev/null +++ b/tasks/pg-engine-notify-listen.md @@ -0,0 +1,139 @@ +--- +id: pg-engine-notify-listen +name: Postgres engine — LISTEN forwarder (full behavior) + notify/listen surface +status: pending +depends_on: [pg-engine-seam-tx] +scope: moderate +risk: high +impact: component +level: implementation +tags: [wave-4, postgres-engine, forwarder] +--- + +## Description + +Complete the LISTEN forwarder's behavior (the open task stood up the +skeleton) and implement `Store::notify` and `Store::listen` — the pg +arm of the wake contract (ADR-006). The forwarder is hand-rolled +machinery (ADR-004's decision; `postgres-notify` derive-not-adopted) +— its failure modes are *learned territory*, POC-pinned; this task +owns them as passing tests. + +**Forwarder completion** (on the skeleton's dedicated non-pooled +connection + poll loop + bounded broadcast): + +- **Reconnect**: exponential backoff 50 ms → 2 s cap (the POC's + policy); on each successful reconnect, re-LISTEN from the live + channel list (the store's channel registry — `listen` adds to it, + receiver drop removes it) and broadcast the **synthetic + reconnect-wake** on the reserved channel + `__alkstore_listener_reconnected__` (ADR-008 §4's one v1-reserved + string — the POC's ad-hoc `__listener_reconnected__` renamed into + the namespace) to *every* subscriber's receiver (ADR-006's + reconnect-recovery semantics — the no-replay hole's honest + communication). +- **Fanout**: `poll_message`'s notifications fan out per channel — + each `listen(channel)` subscriber receives wakes for its channel + only (plus the reserved reconnect-wake, which every subscriber + receives). Bounded broadcast per subscriber or one shared bounded + channel with per-subscriber routing — implementer's choice, but + **lag is surfaced, not silent** (the POC's `Lagged(n)` posture: + log/record and keep going; a lagged subscriber's missed wakes are + the contract's best-effort-hint shape, recovery is re-read). +- **Wake content**: `Wake { channel }` only — never payloads, never + ids (ADR-008 §3). Per-notify delivery (no coalescing — LISTEN + delivers each committed NOTIFY; the SQLite engine's coalescing is + the documented engine asymmetry). +- **Receiver close semantics** (ADR-021 §5's pg arm — the documented + engine asymmetry): the receiver **stays open across forwarder + reconnects** (reconnect-wakes arrive on it; no close) and closes + only at engine shutdown (`close()`/Drop → all receivers terminal + `None`). This is the inverse of SQLite's death-closes-receivers — + pin it by test (kill the listener's backend mid-subscription; + receivers stay open; the reconnect-wake arrives; post-reconnect + deliveries work). +- **Identifier quoting**: channel names are quoted identifiers + (`LISTEN "{name}"` with `"` doubling — the POC's `quote_ident` + shape); validation already rejects empty/reserved, but quoting is + what makes arbitrary legal names (colons, etc.) work. + +**`Store::notify(channel, payload)`**: auto-commit path — pool +checkout, `SELECT pg_notify($1, $2)` in its own implicit transaction, +return to pool. The 8000-byte client-side check (typed +`PayloadTooLarge` before any round trip) — the seam task built it for +`notify_tx`; this task wires the auto-commit twin through the same +check (one owner of the limit constant). + +**`Store::listen(channel)`**: validate (shared-namespace kinds — +`InvalidName`/`ReservedName`), register the channel in the store's +channel registry (issuing `LISTEN` on the forwarder's connection — +the dynamic channel set), return the boxed `PgWakeReceiver` bridging +the fanout: +- `recv()` → `Some(Wake { channel })` per notification; `None` only + at engine shutdown (terminal — never reopens; ADR-021 §5's pg arm). +- `try_recv` / `recv_timeout` per the core trait's pinned arms + (`Closed` distinguishable from idle; timeout expiry = `Ok(None)`). +- Receiver drop unregisters the channel (issuing `UNLISTEN` when the + last subscriber for that channel drops — the channel registry's + refcount; zero subscribers ⇒ no LISTEN needed). +- `listen` on a closed store fails closed (`Database`). + +**Wake-arrives honesty**: tests pin wake-arrives (cross-connection +commit → wake), the reconnect-wake path, and close-at-shutdown — not +wake-exclusivity (a wake may arrive for unrelated activity on the +shared channel namespace; the contract's best-effort-hint posture). + +**The POC's two deadlock pitfalls, pinned as tests** (they are the +hand-rolled posture's owned failure modes, ADR-004): +1. The poll loop runs before the first client query on the listener + connection (structurally excluded by the skeleton; pin the + ordering with a test that issues a query immediately after + listener spawn). +2. The listener's `Client` stays alive for the listener's lifetime + (pin: a long-lived listener survives unrelated client churn). + +## Acceptance Criteria + +- [ ] Forwarder reconnects with 50 ms → 2 s exponential backoff; + re-LISTENs from the live channel set; broadcasts the synthetic + reconnect-wake on `__alkstore_listener_reconnected__` to every + subscriber after each reconnect — tested via backend kill +- [ ] No-replay honesty pinned: a commit during a connection gap is + never re-delivered (`!saw_replay` posture); recovery is the + reconnect-wake + consumer re-read +- [ ] `notify` delivers commit-atomic auto-commit notifies; the + 8000-byte check is typed, client-side, before any round trip + (both `notify` and `notify_tx` — one limit owner) +- [ ] `listen` returns a working `WakeReceiver`: per-notify wakes, + `Wake { channel }` only, channel-scoped fanout, multi-channel + on one listener connection, N subscribers per channel +- [ ] Receiver close semantics: stays open across reconnects + (reconnect-wake arrives), closes only at engine shutdown + (terminal `None`); `try_recv`/`recv_timeout` arms match the + pinned shapes +- [ ] Receiver drop unregisters (UNLISTEN at last-subscriber drop) +- [ ] Channel validation on both entry points; closed-store + fail-closed on both +- [ ] The two deadlock pitfalls pinned by tests +- [ ] `cargo test -p alkstore-postgres` (harness server), clippy + `-D warnings`, fmt clean; gates green server-less + +## References + +- docs/architecture/engine-postgres.md (Connection architecture: forwarder) +- docs/architecture/decisions/004-postgres-driver.md (forwarder decision, pitfalls) +- docs/architecture/decisions/006-wake-and-delivery-contract.md (wake contract, reconnect-wake) +- docs/architecture/decisions/008-contract-v1-pinning.md §3, §4 (wake type, reserved string) +- docs/architecture/decisions/016-deployment-honesty.md §5 (PayloadTooLarge) +- docs/architecture/decisions/021-tx-reads-and-value-shape-fixes.md §5 (receiver arms, pg close semantics) +- docs/research/poc-pg-posture-findings.md (Sub-module L — the measured ground) +- alkstore-sqlite/src/notify.rs (the structural twin — inverse close semantics) +- /workspace/alkstore-pg-posture-poc/src/listen.rs (the POC's ~90-line shape) + +## Notes + +> To be filled by implementation agent + +## Summary + +> To be filled on completion \ No newline at end of file diff --git a/tasks/pg-engine-open-opts.md b/tasks/pg-engine-open-opts.md new file mode 100644 index 0000000..e211241 --- /dev/null +++ b/tasks/pg-engine-open-opts.md @@ -0,0 +1,113 @@ +--- +id: pg-engine-open-opts +name: Postgres engine — `open` constructor, `PgOpts`, pool + listener wiring +status: pending +depends_on: [pg-engine-schema] +scope: moderate +risk: medium +impact: component +level: implementation +tags: [wave-4, postgres-engine] +--- + +## Description + +Stand up the pg engine's public surface: the `open` constructor, +`PgOpts`, and the connection architecture of engine-postgres.md's +"Connection architecture" section — the structural twin of wave 3's +`sqlite-engine-open-opts`, with the pg deltas. + +- `PgOpts` (engine-crate type, ADR-008 §6's config split; NOT + `#[non_exhaustive]` — opts structs consumers construct, ADR-017 §3's + exemption; `Default` + `Clone`): at minimum + - connection string / config (the tokio-postgres `Config`-parseable + form — the consumer's server address, credentials; the engine + never invents defaults for these, `open` fails `Database` if + unparseable/unreachable), + - `schema: String` (default `"alkstore"` — ADR-010 §8), + - `max_size` (the deadpool pool size; default documented, the + deployment matrix's budget line — the listener adds +1 outside + the pool), + - `synchronous_commit: bool` (default `true` — ship config, + ADR-004/deployment.md's measured trade; wired as a per-session + SET at connect, POC-verified mechanics). +- `open(config, opts) -> Result>` (concrete + `PgStore` re-exported, mirroring the SQLite engine's + boxed-constructor + pub-concrete shape): + - Build the deadpool pool (`RecyclingMethod::Fast` — no `DISCARD + ALL` recycling, the POC's zero-error posture). + - Run the schema task's bootstrap on a pool connection at open + (idempotent; open fails `Database` if bootstrap fails). + - Spawn the **forwarder** (the listener task) — the wake substrate + every later mechanism task rides. The forwarder is this task's + hard part; its full behavior (reconnect, re-LISTEN, reconnect- + wake) is the notify-listen task's, but the *skeleton* lives here: + a dedicated non-pooled connection (deadpool#360 — pooled + connections cannot deliver), the `poll_message` loop fanning out + into a bounded broadcast channel (lag surfaced, not silent), the + two POC-pinned deadlock pitfalls respected (the poll loop running + before the first client query on the listener connection; the + `Client` kept alive for the listener's lifetime), and the + dynamic channel set (LISTENs re-issued from the live channel + list — the notify-listen task adds channels to it). + - The store struct holds pool + forwarder handle + schema name + behind the `Store` trait; `close()`/`Drop` tear down cleanly + (drop the listener connection, close the pool). +- **Trait stubs**: all `Store` methods return + `Err(Database("… wiring lands with the … task"))` — the wave-3 + posture; the mechanism tasks replace them. `with_tx` surfaces the + `begin_tx` stub naturally. +- **Error mapping posture established here** (the seam task's + foundation): tokio-postgres/deadpool errors → `Error::Database` + with the source chain preserved (ADR-008 §5's opaque fallback) — + the two mappings (`pg_error`, `pool_error` or one helper) the + mechanism tasks reuse. +- **No `spawn_blocking` seam on this engine** — natively async + (Send+Sync client, POC #2 compile-probe); the engine crate's tokio + dep already carries what's needed. State the posture in the crate + docs (the SQLite engine's `# Posture` heading shape: multi-host + native per ADR-016, the listener budget line per deployment.md). + +Tests against the harness server: open/close round-trip, bootstrap +ran (tables exist), pool bounds, forwarder skeleton up (a LISTEN +issued through it delivers — the minimal fanout proof, full forwarder +behavior in the notify-listen task), unreachable-server `open` fails +`Database`, opts flow (schema name lands in the created schema's +name; `synchronous_commit` observable via `SHOW`). + +## Acceptance Criteria + +- [ ] `PgOpts` with connection config, `schema` (default + `alkstore`), `max_size`, `synchronous_commit` (default on); + documented; not `#[non_exhaustive]` +- [ ] `open` boots pool + bootstrap + forwarder skeleton; failure at + any step is a typed `Database` error (source chain preserved) +- [ ] Forwarder skeleton: dedicated non-pooled connection, poll loop + before first query, bounded broadcast fanout, dynamic channel + set; the two POC deadlock pitfalls structurally excluded +- [ ] `close()`/`Drop` teardown clean (listener connection dropped, + pool closed); post-close ops fail closed +- [ ] Error-mapping helpers exist and are the documented reuse point +- [ ] Crate docs carry the `# Posture` statements (multi-host, + listener budget, no spawn_blocking) +- [ ] `cargo test -p alkstore-postgres` (harness server), clippy + `-D warnings`, fmt clean; gates green server-less (tests skip) + +## References + +- docs/architecture/engine-postgres.md (Connection architecture) +- docs/architecture/decisions/004-postgres-driver.md (the forwarder + shape, the two pinned pitfalls, RecyclingMethod::Fast) +- docs/architecture/decisions/008-contract-v1-pinning.md §6 +- docs/architecture/deployment.md (budgets, synchronous_commit) +- docs/research/poc-pg-posture-findings.md (Sub-modules L/T; the + invocation note's isolation caveat) +- alkstore-sqlite/src/store.rs, opts.rs (the structural twin's shape) + +## Notes + +> To be filled by implementation agent + +## Summary + +> To be filled on completion \ No newline at end of file diff --git a/tasks/pg-engine-queues.md b/tasks/pg-engine-queues.md new file mode 100644 index 0000000..db5c739 --- /dev/null +++ b/tasks/pg-engine-queues.md @@ -0,0 +1,141 @@ +--- +id: pg-engine-queues +name: Postgres engine — queues (`Queue`, `FOR UPDATE SKIP LOCKED` claims, job handles, backoff curve) +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 queue mechanism's pg arm — the re-derived queue +machinery (ADR-004/005's greenfield posture, pg-boss schema family as +design reference; ADR-010's depth) over the schema task's job/dead +tables and the forwarder's wake substrate: + +- **`Store::queue(name, opts)`** — validated constructor (shared- + namespace kinds; construction stays free, queues are names not + registered objects). The handle carries its `QueueOpts` as the + stamp source. +- **`enqueue`** (auto-commit): resolves `EnqueueOpts` over the + handle's stamps via the seam task's `resolve_enqueue_opts` + + `stamps_with_override` (the pg engine's own `resolution.rs` — + one clock read per enqueue, delay-over-`run_at`, relative- + `expires`, `max_attempts` the one per-job override), `encode_payload` + at the seam, `INSERT` on a pool connection. Then **wake**: + `pg_notify` on the queue's wake channel (the mechanism-name-is- + the-channel realization) — the LISTEN-driven claim path's push + half. +- **`claim_one` / `claim_batch`** — exactly-once handout under + concurrency: the `FOR UPDATE SKIP LOCKED` claim (POC-pinned under + 4×4 concurrency) on a pool connection, in one statement or a + CTE-scoped transaction — implementer's choice, but the + exactly-once property is what the tests pin (two workers, one row; + N×M concurrent drain, zero double-claims). Semantics per ADR-010: + ordering `priority DESC, run_at ASC, enqueue order`; `attempts += + 1` per claim (a reclaim consumes an attempt); the claim sets the + row's deadline from the job's own stamped `visibility_timeout_s`; + the pre-claim dead-letter sweep for exhausted reclaimables (the + substrate's D-12 shape — a worker dying right after its last + allowed claim dead-letters, not strands); `worker_id` stamps the + claimant column (non-empty validated, consumer-local — ADR-019 §2). + The **extent guard** (`n <= 0` → empty `Vec` at trait-impl entry, + ADR-023 §2). Claims return boxed `JobHandle`s (ops + row value). +- **`JobHandle` impl** (ADR-019 §3, ADR-010 §2's uniform validity + predicate — `processing` + unexpired claim deadline; refusal = + `Ok(false)`, never error): + - one-shot `ack`/`retry`/`fail` as `'static` boxed futures over the + consumed handle (mirroring `TxHandle::commit`'s shape); + repeatable `heartbeat(&self, extend)` — absolute reset of the + claim deadline (not additive), late heartbeat refused. + - `ack` deletes the row; `retry(err, None)` computes the queue's + **equal-jitter exponential curve** — the pg engine's own + implementation of ADR-010 §3's pinned form + (`base * 2^(attempt-1)`, jitter over `[half, cap]` integerized + inclusive, capped 3600 s; attempt index = the row's post-claim + count). **The jitter's randomness source is this task's flagged + choice**: the SQLite engine used a std-only `RandomState`-hash + draw (no `rand` dep, ADR-005's posture) — reuse that approach + unless a pg-native source is materially better, and **flag the + choice in Notes**. `retry(err, Some(d))` overrides; `fail(None)` + pins `"failed"`; `retry`'s exhaust string pins + `"max attempts exceeded"` (the engine-default strings land + identically on both engines — the backlog row's wording). + - Dead-letter moves (retry-at-budget, fail) write the dead table + with `last_error`/`died_at`; dead rows are `get_job`-visible. +- **`ack_batch(ids)`** — the batch ack applies the uniform predicate + per id (lapsed or non-claimed ids silently not counted; count + returned; no worker filter — the wave-3 D-31 delta's shape is the + contract's, not a substrate artifact). +- **`cancel(job_id)`** — unconditional delete (pending or processing, + not an interrupt); `Ok(false)` when row-less. +- **`get_job(job_id)`** — dead-visible pure read (pool connection; + `Option` with `last_error`/`died_at` on dead rows; the + resolved values — not raw opts — are what the row shows, ADR-020). +- **`sweep_expired()`** — both-states no-stranded-rows move (any + past-expiry row → dead with `last_error='expired'`) + retention + enforcement (`dead_letter_retention_s`, `None` = forever) — + ADR-010 §5; the queue-scoped handle form. +- **LISTEN-driven claim with re-poll safety net** (the engine's + default consumption posture, ADR-004): the *contract* claim ops are + pull-shaped (callers claim in their own loops); the wake substrate + makes those loops reactive — a consumer that `listen(queue_name)`s + and claims on wake gets the measured 3–6 ms path; the re-poll + safety net (interval-tunable, `PgOpts`-carried if wired — + implementer's choice, documented either way) covers LISTEN loss + (the no-replay hole). The engine's own tests exercise the + wake-driven claim loop (commit → notify → wake → claim, the POC's + property) to pin the posture; the contract surface itself stays + pull-shaped. +- **Closed-store posture** on all queue ops (the wave-3 shape). + +## Acceptance Criteria + +- [ ] `enqueue` resolves opts over the handle's stamps (the shared + resolution arithmetic; `max_attempts` the one override); + payload bytes exact; wake notify fires on the queue's channel +- [ ] `claim_one`/`claim_batch`: exactly-once under concurrency + (two-workers-one-row; 6×40 concurrent drain, zero + double-claims); ordering row pinned; reclaim consumes an + attempt; deadline from the row's stamp; extent guard + (`n <= 0` → empty `Vec`); pre-claim exhausted-reclaimable sweep +- [ ] `JobHandle`: validity predicate uniform (lapse refuses all ops, + `Ok(false)`); heartbeat absolute-reset + late-refused; ack + deletes; retry curve matches ADR-010 §3's range/cap + (statistical + cap tests, the SQLite task's shape); dead-letter + defaults (`"max attempts exceeded"`, `"failed"`, caller strings + verbatim); dead rows `get_job`-visible +- [ ] `ack_batch` per-id predicate (mixed lapsed/valid ids); `cancel` + unconditional both states; `sweep_expired` both-states + + retention (incl. forever-none) +- [ ] Wake-driven claim loop pinned end-to-end (enqueue → wake → + claim, cross-connection) +- [ ] Validation (empty/reserved queue names, empty `worker_id`) + + closed-store fail-closed on all queue ops +- [ ] `cargo test -p alkstore-postgres` (harness server), clippy + `-D warnings`, fmt clean; gates green server-less + +## References + +- docs/architecture/core-contract.md (queues section) +- docs/architecture/decisions/010-queue-semantics-depth.md (§1–§5, the depth) +- docs/architecture/decisions/019-mechanism-handle-surfaces.md §1–§3 +- docs/architecture/decisions/020-enqueue-opt-semantics-and-bridges.md +- docs/architecture/decisions/023-fourth-review-round.md §2 (extent guard) +- docs/architecture/decisions/004-postgres-driver.md (re-derive posture, LISTEN-driven default) +- docs/architecture/queues.md (lifecycle, consumption posture) +- docs/research/reference-pgboss-rs-semantics.md (design reference) +- alkstore-sqlite/src/queue.rs (the structural twin — same semantics, own SQL) +- alkstore/src/queue.rs, job_handle.rs (the traits) + +## Notes + +> To be filled by implementation agent + +## Summary + +> To be filled on completion \ No newline at end of file diff --git a/tasks/pg-engine-scheduler-outbox.md b/tasks/pg-engine-scheduler-outbox.md new file mode 100644 index 0000000..e013bf2 --- /dev/null +++ b/tasks/pg-engine-scheduler-outbox.md @@ -0,0 +1,154 @@ +--- +id: pg-engine-scheduler-outbox +name: Postgres engine — scheduler (`schedule`/`unschedule`/`run_schedules`) and outbox helper +status: pending +depends_on: [pg-engine-queues, pg-engine-locks] +scope: broad +risk: high +impact: component +level: implementation +tags: [wave-4, postgres-engine] +--- + +## Description + +Implement the scheduler collapse's pg arm (ADR-009) and the outbox +helper (ADR-014/019 §5) — the wave's last mechanism pair, riding the +queues task's claim machinery and the locks task's lock machinery: + +**Scheduler** (over the schema task's schedule table): + +- **`schedule(name, spec, queue, payload, opts)`** — upsert by name; + entry-point validation on all three name-bearing arguments + (schedule name like a queue name; the **queue argument** identically + per ADR-021 §3 — a schedule cannot target a reserved name, so the + outbox backing queue stays reachable only through the outbox + surface); payload encoded at the seam (`Codec` propagation); the + spec grammar is `@every ` only (`s|m|h|d`) — **the pg + engine owns its parser** (no shared substrate parser on this side; + the grammar is small and pinned; cron strings → `InvalidSpec`). + Rejection is pre-storage (`InvalidSpec` before any round trip). +- **`unschedule(name)`** — the unregister; `Ok(false)` when absent. +- **`run_schedules(stop: StopToken)`** — the leader loop, the + structural twin of the SQLite engine's runner with the pg deltas: + - **Leadership via the engine's own lock machinery** on the + reserved name `__alkstore_scheduler` (ADR-009 §1) — `try_lock` + with a per-runner-instance-unique owner token (the wave-3 + finding's lesson: a shared constant would make two processes each + see "self" holding; the token is pid + subsec-nanos or + equivalent). TTL engine-internal (10 s), renewed at the top of + every loop iteration **and** on a cadence across the sleep (the + wave-3 posture: without in-sleep renewals, a boundary further out + than the TTL lapses its own lease by drift). A failed mid-sleep + renewal is not acted on at the sleep site — the loop's + top-of-iteration renew owns the loss decision. + - **Acquire-time refusal is `Err(LeadershipLost)`** — the same + matchable loss a mid-run loser returns (the wave-3 posture: no + runner-loop exists without the lock; the respawn recipe is + identical either way). + - **The tick**: fire enqueues + boundary advance commit together — + on pg this is one pool-checked-out transaction (`BEGIN` … + `COMMIT` on the held object, not `BEGIN IMMEDIATE` — pg's + default isolation gives the row-locked fire tx ADR-009 §4 + names; the schedule row's `SELECT … FOR UPDATE` inside the tick + tx is the row lock that serializes rogue second tickers). The + tick body: boundary math over the registered specs (the + **one-firer-per-boundary** guarantee — at-least-once per elapsed + boundary, the **64-boundary catch-up cap with skip-forward** + past the cap, honker parity), each fire an ordinary enqueue + stamped with the `ScheduleOpts` resolved over the target queue's + *derived engine defaults* (no handle is open — ADR-020 §3), then + the soonest-boundary read for the sleep deadline. + - **Wake-driven tick advance** (the pg delta over the SQLite + runner's pure-sleep shape): the runner parks on the sleep + deadline, but an enqueue-wake on the scheduler's own wake channel + (or a plain bounded-wait interrupt — implementer's choice, + documented) may shorten the sleep so a newly-registered schedule + is noticed promptly; correctness never rides the wake (the + deadline always fires; the wake only shortens idle latency — + the SQLite runner's 60 s idle-tick posture is the acceptable + floor if the wake path proves fussy; flag the choice in Notes). + - **Clean stop** (`StopToken::cancel`) → `Ok(())`, the leadership + lock released cleanly; **loss** → `Err(LeadershipLost)` before + any tick. +- **Boundary math is this engine's own implementation** (ADR-012 §2's + one-owner rule — the substrate's `scheduler_tick`/`scheduler_soonest` + are SQLite-side): the `@every` interval arithmetic, the boundary + advance, the 64-cap catch-up, and the skip-forward. The contract + suite (wave 5) pins the two engines' boundary behavior to identical + outcomes. + +**Outbox** (a helper over queues — no separate mechanism): + +- **`Store::outbox(name)`** — validated constructor (empty → + `InvalidName`, reserved-prefixed → `ReservedName`); carries no opts + of its own (the backing queue's derived 60/5/5 set is + engine-documented, ADR-010 §3a). +- **`enqueue(payload, opts)`** — the auto-commit path into the + derived backing queue: derive `__alkstore_outbox:{name}` (the + reserved prefix makes it unreachable by `enqueue_tx`/`queue()` by + design), stamp the 60/5/5 set with the `EnqueueOpts::max_attempts` + override, full ADR-020 resolution, ride the queues task's enqueue + machinery. +- **`run_once(worker_id, delivery) -> bool`** — the pull op: claim + one job from the backing queue via the ordinary claim machinery + (the queues task's), run the delivery callable (core's `Delivery` + trait object — ADR-019 §5's pinned shape), `ack` on `Ok`, + `retry(err, None)` (the queue's curve; the delivery error's + `Display` string as the retry's exhaust string — the wave-3 + posture) on `Err`, `false` on empty. **No engine-issued heartbeat + inside delivery** (honker parity, inherited deliberately — the + boxed `JobHandle`'s own `heartbeat` remains available to the + closure); the boxed `JobHandle` handed to the caller. +- **`outbox_enqueue_tx`** landed with the seam task; this task's + tests may extend its coverage (the commit-atomicity row is + suite-owned in wave 5). + +## Acceptance Criteria + +- [ ] `schedule`: validation triad (name/spec/queue), `@every`-only + grammar (cron → `InvalidSpec` pre-storage), upsert read-back; + the pg-owned parser pinned by tests (accepts/rejects the + grammar's exact surface) +- [ ] `run_schedules`: leadership via `__alkstore_scheduler` with a + per-instance owner token; acquire-time refusal = + `Err(LeadershipLost)`; tick fires + boundary advance in one + transaction (row-locked — a rogue second ticker cannot + double-fire); loss returns `Err(LeadershipLost)` before any + tick; clean stop releases the lock; in-sleep lease renewals + hold across boundaries further out than the TTL +- [ ] One-firer-per-boundary: two runners on the same db, exactly one + fire per boundary (the band test); 64-cap catch-up + + skip-forward with no in-window refire +- [ ] Fired jobs stamp the derived defaults resolved over + `ScheduleOpts` (`get_job`-visible) +- [ ] `outbox`: constructor validation; backing queue unreachable by + `enqueue_tx`/`queue()`; 60/5/5 stamps + `max_attempts` + override; `run_once` ok/err/false dispositions with the curve + gap asserted; in-delivery heartbeat renewal available; + exhaustion to dead with the delivery error string +- [ ] `cargo test -p alkstore-postgres` (harness server), clippy + `-D warnings`, fmt clean; gates green server-less + +## References + +- docs/architecture/decisions/009-scheduler-collapse.md (§1–§6) +- docs/architecture/decisions/014-outbox-tx-enqueue.md +- docs/architecture/decisions/019-mechanism-handle-surfaces.md §4, §5 +- docs/architecture/decisions/020-enqueue-opt-semantics-and-bridges.md §3 +- docs/architecture/decisions/021-tx-reads-and-value-shape-fixes.md §3 +- docs/architecture/engine-postgres.md (Mapping the contract: scheduler/outbox) +- docs/architecture/queues.md (scheduler section) +- alkstore-sqlite/src/scheduler.rs, outbox.rs (the structural twin — + same semantics, own boundary math; its Notes carry the flake-fix + lessons: drive catch-up tests from far-out horizons so no live + boundary races the assertion window) +- alkstore/src/stop_token.rs, outbox.rs (the traits) + +## Notes + +> To be filled by implementation agent + +## Summary + +> To be filled on completion \ No newline at end of file diff --git a/tasks/pg-engine-schema.md b/tasks/pg-engine-schema.md new file mode 100644 index 0000000..c77925d --- /dev/null +++ b/tasks/pg-engine-schema.md @@ -0,0 +1,110 @@ +--- +id: pg-engine-schema +name: Postgres engine — schema bootstrap (engine-owned schema, table family) +status: pending +depends_on: [] +scope: narrow +risk: low +impact: component +level: implementation +tags: [wave-4, postgres-engine] +--- + +## Description + +Stand up the pg engine's data model: the engine-owned PostgreSQL schema +and its table family — the dependency root every later wave-4 task +rides. Per ADR-010 §8 (one engine-owned schema, default `alkstore`, +queues as rows — no per-queue tables/schemas) and engine-postgres.md's +mapping table. + +A `schema.rs` module in `alkstore-postgres`: + +- **Schema name is engine configuration** (default `alkstore`) — it + rides `PgOpts` (the open task wires it; this task's bootstrap takes + the schema name as an argument). +- **Bootstrap is idempotent DDL**, runnable at open on any connection: + `CREATE SCHEMA IF NOT EXISTS` + `CREATE TABLE IF NOT EXISTS` + + indexes. All engine-owned objects live in the one schema; consumer + tables co-tenant the instance untouched (deployment.md's co-tenancy + posture). +- **Table family** (column sets follow the contract's value shapes — + the `#[doc(hidden)]` constructors' full field lists in + `alkstore/src/{job,stream,schedule}.rs` are the field lists of + record; ADR-019 §1/§3, ADR-015 §3, ADR-009 §3, ADR-021 §2): + - **job table** (live rows — pending and processing): queue name, + payload, enqueue/ready-time resolution columns (`run_at`, + `expires_at`), `priority`, the stamped `QueueOpts` columns + (`visibility_timeout_s`, `max_attempts`, `backoff_base_s`, + `dead_letter_retention_s` — ADR-010 §3a), `attempts`, + `worker_id`, `claimed_at`, `claim_expires_at`, `last_error`. + Claim-ordering index per ADR-010's ordering row (`priority DESC, + run_at ASC, enqueue order`). + - **dead table**: the job row's diagnosis fields + `died_at`, + `last_error`; retention index for the sweep's retention + enforcement. + - **stream events table**: `id BIGSERIAL` (the offset — engine- + assigned, monotone, immutable, never renumbered), stream name, + nullable `key` column (carried metadata, ADR-015 §1), payload, + `created_at` (unix-seconds, informational). The `(stream, id)` + index serving the `offset ASC` read paths. + - **offsets table**: (stream, consumer) → offset, the monotone + checkpoint upsert target. + - **schedule table** (ADR-009): name (unique), `@every` spec, target + queue, payload, the `ScheduleOpts` stamps, and the boundary-state + columns the tick math advances (the substrate's + `__alkstore_scheduler_tasks` shape is the reference — + `reference-honker-machinery.md`). + - **locks table**: name (unique), owner, expiry timestamp — the TTL + lock machinery's storage (the substrate's lock-table semantics, + re-derived; the locks task builds ops on it). +- **Payload columns store the exact encoded bytes** (`bytea`) — not + `JSONB`: JSONB normalization (key reordering, duplicate folding) + would break the contract suite's byte-identical-rows row (ADR-020 + §4). The engine stores exactly what `encode_payload` returned. +- **No notifications table** — pg notify is native (`pg_notify`); the + SQLite notify-table carriage has no pg counterpart. The 8000-byte + check is client-side (the seam task), not storage-shaped. +- **No migration machinery in v1** — tables are created fresh per + schema; append-column migrations are the substrate's evolution + discipline, not a greenfield v1 need (note the posture in the module + docs; wave 6's release pass revisits if the deployment story + demands in-place upgrades). + +Tests use a raw `tokio_postgres::Connection`/Client against the +harness server (no engine code yet): bootstrap runs twice cleanly +(idempotence), every table exists with its pinned columns, the +bigserial offset behaves (monotone, gap-on-delete), the claim-ordering +index exists. + +## Acceptance Criteria + +- [ ] `schema.rs` with an idempotent `bootstrap(conn, schema_name)` + creating the full table family + indexes in the one schema +- [ ] Column sets match the contract's value shapes (the constructors' + field lists; stamp columns per ADR-010 §3a; `claimed_at` per + ADR-021 §2) +- [ ] Payload columns are `bytea` (exact-bytes posture, JSONB + rejected — documented in the module docs with the reason) +- [ ] Schema name is a parameter (default `alkstore` decided at the + `PgOpts` layer, next task) +- [ ] Idempotence + table-shape tests green against the harness server +- [ ] `cargo test -p alkstore-postgres`, clippy `-D warnings`, fmt clean + +## References + +- docs/architecture/engine-postgres.md (Identity and posture; mapping table) +- docs/architecture/decisions/010-queue-semantics-depth.md §3a, §8 +- docs/architecture/decisions/015-streams-depth.md §1, §3 +- docs/architecture/decisions/009-scheduler-collapse.md §3 +- docs/architecture/deployment.md (co-tenancy) +- docs/research/reference-honker-machinery.md (schema reference read) +- alkstore/src/job.rs, stream.rs, schedule.rs (the field lists of record) + +## Notes + +> To be filled by implementation agent + +## Summary + +> To be filled on completion \ No newline at end of file diff --git a/tasks/pg-engine-seam-tx.md b/tasks/pg-engine-seam-tx.md new file mode 100644 index 0000000..ef1aa33 --- /dev/null +++ b/tasks/pg-engine-seam-tx.md @@ -0,0 +1,129 @@ +--- +id: pg-engine-seam-tx +name: Postgres engine — `begin_tx`, `TxHandle` (pooled-object handle), all eleven `*_tx` methods +status: pending +depends_on: [pg-engine-open-opts] +scope: moderate +risk: high +impact: component +level: implementation +tags: [wave-4, postgres-engine, seam] +--- + +## Description + +Implement the transactional seam — the pg arm of ADR-007, the shape +every other mechanism task rides, and the wave's highest-risk piece +(the wave-3 review's writer-slot findings show exactly the defect +class seam error arms breed): + +- **No bridge, no `spawn_blocking`** — the pg handle holds the pooled + object directly (tokio-postgres `Client` is `Send + Sync`, POC #2 + compile-probe; ADR-007's pg arm). `begin_tx`: pool checkout + + `BEGIN`, return the boxed `PgTxHandle` holding + `Option`; ops are straight `.await`s; commit/ + rollback return the object to the pool. +- **Drop = rollback** (ADR-021 §4): the handle's `Drop` issues + `ROLLBACK` and re-pools the object — panic-safe (no panic escapes + drop). A failed rollback's disposition is pinned (the object is + discarded, not re-pooled — its state is unknowable; the pool + accounting stays exact, deployment.md's ADR-021 note). Ops after + commit/consumption fail closed with `Error::Closed` (the wave-3 + fail-closed guard shape). +- **All eleven `*_tx` methods, for real** (nothing stubs — the schema + exists, the SQL is this task's to write): + - Writes: `enqueue_tx` (stamps the derived plain-queue defaults + 300/3/5/none — the no-handle-open shape, ADR-020 §3a), + `publish_tx`, `publish_with_key_tx` (empty-`Some`-key → + `InvalidName`, ADR-015 §2), `notify_tx` (`SELECT pg_notify($1, + $2)` inside the caller's tx — delivers at commit, native + commit-atomicity, ADR-007; the 8000-byte client-side check fires + here and on the auto-commit twin — typed `PayloadTooLarge` with + the limit in the variant, before any round trip, ADR-016 §5), + `save_offset_tx` (monotone upsert), `outbox_enqueue_tx` (derives + the `__alkstore_outbox:{name}` backing queue name engine-side, + stamps the outbox's 60/5/5 set — ADR-014 §1). + - Reads (ADR-021 §1, read-your-own-writes through the same held + tx): `get_job_tx` (queue-scoped — a found row's queue column + must match the argument, the wave-3 seam's scoping rule), + `get_offset_tx`, `read_since_tx`, `read_from_consumer_tx` + (offset ASC; the extent guard `limit <= 0` → empty `Vec` at + trait-impl entry, ADR-023 §2). + - `commit(self: Box)`: `COMMIT` + re-pool. A failed COMMIT + discards the object (unknowable state) — the wave-3 review's + finding (c), pre-empted here. +- **Name validation at every entry point** via core's `validation` + helpers (shared-namespace kinds vs consumer-local; ADR-008 §4) — + before any round trip. +- **Payload encoding**: core's fallible `encode_payload` at the seam; + `?` the `Codec` error (ADR-023 §1). Decode via the `#[doc(hidden)]` + `from_row` constructors; malformed rows → `Codec` (the posture + guard). +- **The engine-side arithmetic lives here first** (one owner per + formula, ADR-012 §2 — the pg engine's own `resolution.rs`, the + SQLite one is reference not dependency): `now` (the single clock + read — `now()` at the SQL layer or one `std::time` read; pick one, + document it), `plain_queue_default_stamps()` (300/3/5/none), + `outbox_default_stamps()` (60/5/5), `outbox_backing_queue_name()` + (the reserved-prefix derivation), `resolve_enqueue_opts` + (delay-over-`run_at`, relative-`expires`, one clock read per + enqueue). The auto-commit mechanism tasks reuse these directly — + same shape as the SQLite engine's `resolution.rs`. +- **`pg_notify` payload carriage**: the notify payload is the + serde_json serialization's UTF-8 text (≤ 8000 bytes checked + client-side). Large payloads do NOT ride a side-table in v1 — the + typed rejection is the contract's answer (ADR-016 §5's occurrence + asymmetry; the outbox shape is the documented large-payload + pattern, not engine machinery). + +The mechanism tasks then add what the tx path deliberately lacks: the +auto-commit handles, the forwarder-driven receivers, the backoff +curve, and the scheduler runner. + +## Acceptance Criteria + +- [ ] `begin_tx` checks out + `BEGIN`; concurrent `begin_tx` draws + distinct pooled objects (pool-bounded); pool exhaustion blocks + (deadpool's acquire semantics) — tested +- [ ] All eleven `*_tx` methods implemented with entry-point + validation; writes commit-atomic, reads see the tx's own writes + (read committed, both directions POC-pinned); opts resolution + (delay-over-run_at, relative-expires, 300/3/5/none and 60/5/5 + stamp sets) pinned by tests +- [ ] `commit` commits + re-pools; drop (without commit) rolls back + + re-pools — no-ghosts tested for job rows, stream events, + notifications (a listener hears nothing from a rolled-back + `notify_tx`), and offset saves +- [ ] Drop is panic-safe; failed COMMIT/ROLLBACK arms discard the + object (pool accounting exact); ops-after-consume → `Closed` +- [ ] `notify_tx`/`notify`'s 8000-byte check: typed + `PayloadTooLarge { limit: 8000 }` client-side, before any round + trip; ≤ 8000 crosses +- [ ] `with_tx` (core's provided method) works end-to-end on this + engine: Ok ⇒ commit, Err ⇒ rollback — tested +- [ ] `encode_payload`'s `Codec` error propagates; extent guard on the + tx stream reads; `get_job_tx` queue-scoped +- [ ] `cargo test -p alkstore-postgres` (harness server), clippy + `-D warnings`, fmt clean; gates green server-less + +## References + +- docs/architecture/decisions/007-transactional-seam.md (pg arm) +- docs/architecture/decisions/021-tx-reads-and-value-shape-fixes.md §1, §2, §4 +- docs/architecture/decisions/014-outbox-tx-enqueue.md §1 +- docs/architecture/decisions/020-enqueue-opt-semantics-and-bridges.md §1–§4 +- docs/architecture/decisions/016-deployment-honesty.md §5 +- docs/architecture/decisions/023-fourth-review-round.md §1, §2 +- docs/architecture/engine-postgres.md (Mapping the contract: begin_tx, handle ops) +- docs/research/poc-pg-posture-findings.md (Sub-module T — the shape + (a) evidence, read-your-writes, notify-at-commit) +- alkstore/src/tx.rs (the trait to implement) +- alkstore-sqlite/src/tx.rs, resolution.rs (the structural twin) + +## Notes + +> To be filled by implementation agent + +## Summary + +> To be filled on completion \ No newline at end of file diff --git a/tasks/pg-engine-streams.md b/tasks/pg-engine-streams.md new file mode 100644 index 0000000..967a170 --- /dev/null +++ b/tasks/pg-engine-streams.md @@ -0,0 +1,126 @@ +--- +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 \ No newline at end of file diff --git a/tasks/review-wave-4.md b/tasks/review-wave-4.md new file mode 100644 index 0000000..80d8848 --- /dev/null +++ b/tasks/review-wave-4.md @@ -0,0 +1,94 @@ +--- +id: review-wave-4 +name: Review gate — wave 4 (Postgres engine) +status: pending +depends_on: [pg-engine-integration] +scope: moderate +risk: low +impact: phase +level: review +tags: [wave-4, review] +--- + +## Description + +Review the Postgres engine before wave 5 decomposes. Primary lens per +the plan's review gates: **engine-vs-contract conformance** — the same +discipline as the wave-3 gate, now against the second engine, with +the pg-specific machinery added. Wave 5 (the contract suite) pins +*cross-engine equivalence*; this gate checks the pg implementation +against the pinned contract text line by line. + +Check: + +- **Contract conformance**: every trait method's behavior against + core-contract.md's per-mechanism text — validation at entry points + (auto-commit AND tx paths), error arms (`Database`-only receivers, + `Codec` at payload seams, `LeadershipLost` vs clean stop), value + shapes (`Job`/`StreamEvent` field-for-field), the no-work-is-a-value + vocabulary (`None`/`false`/empty-`Vec` where pinned, never errors). +- **ADR-023 follow-through**: extent guards at trait-impl entry + (claims + all four stream reads); duration guards on both lock call + sites; `encode_payload` consumed fallibly at every call site. +- **ADR-016 §5**: the 8000-byte check is client-side, typed, before + any round trip, on both `notify` and `notify_tx`; the limit constant + has one owner. +- **Forwarder integrity** (the pg engine's own high-risk machinery — + the analogue of the wave-3 gate's seam-integrity lens): the two + POC-pinned deadlock pitfalls structurally excluded (poll loop + before first query; `Client` alive for the listener's lifetime); + reconnect backoff + re-LISTEN from the live channel set; the + synthetic reconnect-wake on the reserved string broadcast to every + subscriber; lag surfaced, not silent; receiver close semantics + exactly the pg arm (open across reconnects, close at shutdown + only, terminal, never reopens). +- **Seam integrity**: drop = rollback under panic; failed + COMMIT/ROLLBACK arms discard (not re-pool) the object — pool + accounting exact under error paths (deployment.md's ADR-021 note); + ops-after-consume fail closed; no `spawn_blocking` anywhere (the + natively-async posture — a stray bridge would be a finding). +- **Backoff curve + boundary math**: the engine-side formulas match + ADR-010 §3's pinned range/cap and ADR-009 §4's boundary guarantee + exactly (wave 5 pins cross-engine equivalence; this gate checks + this implementation against the text). +- **Schema posture**: one engine-owned schema; payload columns + `bytea` (exact bytes — no JSONB normalization); no per-queue + tables; bootstrap idempotent. +- **Backlog column**: the engine's suite rows present, stamped, green + against the pg factory; the `PayloadTooLarge` pg arm row present. +- **Family standards**: no panics, no unwrap/expect outside tests, no + inline comments (doc comments fine), module-per-file. +- **Gates**: full workspace build/test/clippy/fmt; the pg suite green + against the harness server. + +## Acceptance Criteria + +- [ ] Contract conformance spot-checked against the spec text (not + just the tests) +- [ ] ADR-023's numeric-domain + typed-encoding follow-through + verified at their sites +- [ ] Forwarder + seam integrity verified (code-read, not + test-name-trust) +- [ ] Backoff curve + scheduler boundary math match the pinned ADR + text +- [ ] Backlog column present, stamped, green against the pg factory +- [ ] All gates green (incl. against the harness server) +- [ ] Findings recorded; wave 5 decomposition may proceed + +## References + +- docs/plans/implementation.md (Review gates) +- docs/architecture/core-contract.md +- docs/architecture/engine-postgres.md +- docs/architecture/decisions/004-postgres-driver.md +- docs/architecture/decisions/016-deployment-honesty.md §5 +- docs/architecture/decisions/023-fourth-review-round.md +- tasks/review-wave-3.md (the gate discipline's precedent) + +## Notes + +> To be filled by implementation agent + +## Summary + +> To be filled on completion \ No newline at end of file