diff --git a/alkstore-postgres/src/notify.rs b/alkstore-postgres/src/notify.rs index 94dc058..0b89366 100644 --- a/alkstore-postgres/src/notify.rs +++ b/alkstore-postgres/src/notify.rs @@ -105,9 +105,7 @@ pub(crate) fn listen( } Box::pin(async move { if closed.load(Ordering::Acquire) { - return Err(Error::database(std::io::Error::other( - "the store is closed", - ))); + return Err(database_error("the store is closed")); } forwarder.register(&channel).await?; let broadcast_rx = forwarder.subscribe(); @@ -119,7 +117,7 @@ pub(crate) fn listen( forwarder.unregister(&channel); return Err(database_error("the store is closed")); }; - let (wake_tx, wake_rx) = tokio::sync::mpsc::channel(bridge_capacity()); + let (wake_tx, wake_rx) = tokio::sync::mpsc::channel(BRIDGE_CAPACITY); let bridge = tokio::spawn(bridge_loop(broadcast_rx, wake_tx, channel.clone())); Ok(Box::new(PgWakeReceiver { wake_rx, @@ -134,9 +132,7 @@ pub(crate) fn listen( /// broadcast → receiver). The wake rate is the notify rate (bursty at /// worst); 1024 matches the fanout's bounded shape, lag surfaced at /// the broadcast layer (not here). -fn bridge_capacity() -> usize { - 1024 -} +const BRIDGE_CAPACITY: usize = 1024; /// The broadcast bridge: loops the raw fanout, maps notifications on /// this receiver's channel (plus the reserved reconnect-wake — every diff --git a/alkstore-postgres/src/queue.rs b/alkstore-postgres/src/queue.rs index 0dbef29..435af75 100644 --- a/alkstore-postgres/src/queue.rs +++ b/alkstore-postgres/src/queue.rs @@ -68,8 +68,9 @@ //! ADR-020). //! - `sweep_expired()` — both-states no-stranded-rows move (any //! past-expiry row → dead with `"expired"`) + retention enforcement -//! (ADR-010 §5), single-statement atomic; returns the moved count. -//! No leader lock required. +//! (ADR-010 §5), in the [`in_tx`] multi-statement frame; returns the +//! moved + retention-deleted sum (the SQLite twin's same count). No +//! leader lock required. use std::collections::hash_map::RandomState; use std::hash::BuildHasher; @@ -192,7 +193,7 @@ fn table(schema: &str, name: &str) -> String { /// The live-row select's column list (the uniform 18-column shape the /// dead-table select mirrors with `last_error`/`died_at` in place of /// the NULL casts) — one shape, shared with the tx reads. -fn live_columns() -> &'static str { +pub(crate) fn live_columns() -> &'static str { "id, queue, state, payload, priority, run_at, attempts, max_attempts, worker_id, claimed_at, claim_expires_at, created_at, expires_at, visibility_timeout_s, @@ -202,7 +203,7 @@ fn live_columns() -> &'static str { /// The dead-row select's column list (the 18-column live shape with /// `last_error`/`died_at` riding the NULL casts' slots). -fn dead_columns() -> &'static str { +pub(crate) fn dead_columns() -> &'static str { "id, queue, 'dead' AS state, payload, priority, run_at, attempts, max_attempts, worker_id, claimed_at, claim_expires_at, created_at, expires_at, visibility_timeout_s, @@ -406,8 +407,9 @@ pub(crate) fn claim_stmt_sql(schema: &str) -> String { /// Decode an 18-column job row into the contract value /// ([`Job::from_row`] — the `#[doc(hidden)]` constructor's field list /// of record, ADR-021 §2). Malformed rows map to `Error::Codec` (the -/// decode-side posture guard). One decode owner: the tx reads and this -/// module's claim/get rows share the shape. +/// decode-side posture guard). The one decode owner: the tx reads +/// ([`crate::tx`]) and this module's claim/get rows import it, so a +/// contract-pinned `Job` shape edit has exactly one place to land. pub(crate) fn job_from_row(row: &tokio_postgres::Row) -> Result { let state = match row.get::<_, Option<&str>>(2) { Some("pending") => JobState::Pending, diff --git a/alkstore-postgres/src/tx.rs b/alkstore-postgres/src/tx.rs index 79d3d92..9780587 100644 --- a/alkstore-postgres/src/tx.rs +++ b/alkstore-postgres/src/tx.rs @@ -84,10 +84,11 @@ use serde_json::Value; use tokio::runtime::Handle; use alkstore::{ - BoxedFuture, EnqueueOpts, Error, Job, JobState, Result, StreamEvent, TxHandle, - validate_local_name, validate_shared_name, + BoxedFuture, EnqueueOpts, Error, Job, Result, StreamEvent, TxHandle, validate_local_name, + validate_shared_name, }; +use crate::queue::{dead_columns, job_from_row, live_columns}; use crate::resolution::{ ResolvedEnqueue, outbox_backing_queue_name, outbox_default_stamps, plain_queue_default_stamps, resolve_enqueue_opts, stamps_with_override, @@ -471,13 +472,9 @@ impl TxHandle for PgTxHandle { let row = client .query_opt( &format!( - "SELECT id, queue, 'dead', payload, priority, run_at, attempts, - max_attempts, worker_id, claimed_at, claim_expires_at, - created_at, expires_at, visibility_timeout_s, - backoff_base_s, dead_letter_retention_s, - last_error, died_at - FROM {} WHERE id = $1 AND queue = $2", - self.table(tables::DEAD) + "SELECT {dead} FROM {deadt} WHERE id = $1 AND queue = $2", + dead = dead_columns(), + deadt = self.table(tables::DEAD) ), &[&job_id, &queue], ) @@ -489,13 +486,9 @@ impl TxHandle for PgTxHandle { let row = client .query_opt( &format!( - "SELECT id, queue, state, payload, priority, run_at, attempts, - max_attempts, worker_id, claimed_at, claim_expires_at, - created_at, expires_at, visibility_timeout_s, - backoff_base_s, dead_letter_retention_s, - NULL::text, NULL::bigint - FROM {} WHERE id = $1 AND queue = $2", - self.table(tables::JOB) + "SELECT {live} FROM {jobt} WHERE id = $1 AND queue = $2", + live = live_columns(), + jobt = self.table(tables::JOB) ), &[&job_id, &queue], ) @@ -678,48 +671,8 @@ impl Drop for PgTxHandle { } } -/// Decode a job row (the uniform 18-column shape — the dead-table -/// select carries `last_error`/`died_at`; the live one binds NULLs) -/// into the contract value (`Job::from_row` — the `#[doc(hidden)]` -/// constructor, ADR-021 §2's field list of record). Malformed rows map -/// to `Error::Codec` (the decode-side posture guard). -fn job_from_row(row: &tokio_postgres::Row) -> Result { - let state = match row.get::<_, Option<&str>>(2) { - Some("pending") => JobState::Pending, - Some("processing") => JobState::Processing, - Some("dead") => JobState::Dead, - other => { - return Err(Error::Codec(format!( - "job row state must be pending|processing|dead, got {other:?}" - ))); - } - }; - let payload: Option> = row - .try_get(3) - .map_err(|e| Error::Codec(format!("job row payload must be bytea: {e}")))?; - Ok(Job::from_row( - row.get(0), - row.get(1), - state, - payload.unwrap_or_default(), - row.get(4), - row.get(5), - row.get(6), - row.get(7), - row.get(8), - row.get(9), - row.get(10), - row.get(11), - row.get(12), - row.get(13), - row.get(14), - row.get(15), - row.get(16), - row.get(17), - )) -} - -// The stream-page decode (`stream_events_from_rows`) lives in -// `stream.rs` — the one decode owner the tx reads and the auto-commit -// reads share (the `stream` field mapping and the `Codec` posture -// guard live once; ADR-012 §2's one-owner rule). +// The row decodes ride one owner each: the stream-page decode +// (`stream_events_from_rows`) lives in `stream.rs`, the job-row +// decode (`job_from_row` + the select column lists) in `queue.rs` — +// the one decode owners the tx reads and the auto-commit reads share +// (ADR-012 §2's one-owner rule). diff --git a/tasks/pg-fix-dedupe-cleanup.md b/tasks/pg-fix-dedupe-cleanup.md index 546a875..54211dc 100644 --- a/tasks/pg-fix-dedupe-cleanup.md +++ b/tasks/pg-fix-dedupe-cleanup.md @@ -1,7 +1,7 @@ --- id: pg-fix-dedupe-cleanup name: Decode dedupe + small cleanups in queue/tx/notify (review 002 minor notes) -status: pending +status: completed depends_on: [pg-fix-tx-wake] scope: narrow risk: low @@ -76,8 +76,41 @@ landing after avoids touching the same file regions blind. ## Notes -> To be filled by implementation agent +- The dedupe landed as the task pinned: `job_from_row` is now owned by + `queue.rs` (`pub(crate)`), `tx.rs` imports it; queue.rs's + `live_columns()`/`dead_columns()` went `pub(crate)` so `get_job_tx` + reuses them (the `format!`-bound `{dead}`/`{live}` select strings are + byte-identical to the inlined lists, so zero SQL change). Tx-side + imports ride the existing `crate::queue` module edge — fine + crate-internally (queue.rs already imports tx.rs's + `encode_payload_bytes`/`enqueue_row`; Rust module cycles are legal + within a crate). +- Doc truth-telling choices: queue.rs's `job_from_row` doc now states + the one owner explicitly ("the tx reads (`crate::tx`) ... import it"); + tx.rs's trailing decode-owner comment now names both decode owners + (stream decode in `stream.rs`, job decode in `queue.rs`) under + ADR-012 §2; the removed tx.rs duplicate carried `JobState`'s only + remaining use, so that import was dropped. queue.rs's `sweep_expired` + module-doc bullet now states the moved + retention-deleted sum and + the `in_tx` multi-statement frame (matching the SQLite twin's + semantics — the code already pinned the sum). +- `bridge_capacity()` → `const BRIDGE_CAPACITY` with the rationale doc + comment carried; the closed-store error in `listen` now routes + through `database_error` (it was the only inline duplicate; + import already present). +- No behavior change anywhere: the select SQL strings, error shapes, + and the bridge capacity value are identical. ## Summary -> To be filled on completion \ No newline at end of file +Deduped the pg engine's job-row decode and cleaned the wave-4 minor +notes (review 002): `job_from_row` has one owner (`queue.rs`), `get_job_tx` +reuses `live_columns()`/`dead_columns()` instead of inlining the +18-column lists, `sweep_expired`'s doc states the moved + +retention-deleted sum in the `in_tx` frame (the sum the SQLite twin +returns), `notify.rs`'s closed-store error routes through +`database_error`, and `bridge_capacity` is a `const`. Verified: pg +suite 140/140 green against the harness server (`pglo-poc` :15432) +and 140/140 server-less, unchanged assertions; workspace +`cargo build` / `cargo clippy --all-targets -- -D warnings` / +`cargo fmt --check` / `cargo test --workspace` all clean. \ No newline at end of file