diff --git a/alkstore-postgres/src/lock.rs b/alkstore-postgres/src/lock.rs index 799fb3d..0a55801 100644 --- a/alkstore-postgres/src/lock.rs +++ b/alkstore-postgres/src/lock.rs @@ -35,7 +35,7 @@ use deadpool_postgres::Pool; use alkstore::{BoxedFuture, Lock, Result, validate_local_name, validate_shared_name}; -use crate::resolution::now_unix; +use crate::resolution::{now_unix, saturating_stamp_expr}; use crate::schema::{QualifiedTable, tables}; use crate::seam::{database_error, pg_error, pool_error}; @@ -89,9 +89,10 @@ pub(crate) async fn lock_acquire( .execute( &format!( "INSERT INTO {locks} (name, owner, expires_at) - VALUES ($1, $2, $3::bigint + $4::bigint) + VALUES ($1, $2, {expires_at}) ON CONFLICT (name) DO NOTHING", - locks = table(schema, tables::LOCKS) + locks = table(schema, tables::LOCKS), + expires_at = saturating_stamp_expr("($3::bigint)::numeric + ($4::bigint)::numeric"), ), &[&name, &owner, &now, &ttl], ) @@ -214,9 +215,11 @@ impl Lock for PgLockHandle { let n = client .execute( &format!( - "UPDATE {locks} SET expires_at = $3::bigint + $4::bigint + "UPDATE {locks} SET expires_at = {expires_at} WHERE name = $1 AND owner = $2", - locks = table(&schema, tables::LOCKS) + locks = table(&schema, tables::LOCKS), + expires_at = + saturating_stamp_expr("($3::bigint)::numeric + ($4::bigint)::numeric"), ), &[&name, &owner, &now, &ttl], ) diff --git a/alkstore-postgres/src/queue.rs b/alkstore-postgres/src/queue.rs index 435af75..c6b3169 100644 --- a/alkstore-postgres/src/queue.rs +++ b/alkstore-postgres/src/queue.rs @@ -88,7 +88,7 @@ use alkstore::{ use deadpool_postgres::Object; -use crate::resolution::{Stamps, now_unix}; +use crate::resolution::{Stamps, now_unix, saturating_stamp_expr}; use crate::schema::{QualifiedTable, tables}; use crate::seam::{database_error, pg_error, pool_error}; use crate::tx::{encode_payload_bytes, enqueue_row}; @@ -374,16 +374,20 @@ pub(crate) async fn sweep_exhausted_reclaimables_tx( /// The claim statement: one `UPDATE … RETURNING` over the claimable /// predicate under `FOR UPDATE SKIP LOCKED` (the exactly-once handout, /// ADR-010 §1). The claim stamps `processing`, the claimant, the -/// deadline-from-the-job's-own-stamp, `claimed_at`, and +/// deadline-from-the-job's-own-stamp (saturating server-side to the +/// i64 domain — `saturating_stamp_expr` is the ADR-023 §2 guard: an +/// extreme stamp is a value whose deadline clamps at the edge, never +/// a "bigint out of range" failure), `claimed_at`, and /// `attempts += 1`. /// /// Parameter positions: $1 worker_id, $2 now, $3 queue, $4 extent n. pub(crate) fn claim_stmt_sql(schema: &str) -> String { + let deadline = saturating_stamp_expr("($2::bigint)::numeric + visibility_timeout_s::numeric"); format!( "UPDATE {job} SET state = 'processing', worker_id = $1, - claim_expires_at = $2::bigint + visibility_timeout_s, + claim_expires_at = {deadline}, claimed_at = $2, attempts = attempts + 1 WHERE id IN ( @@ -710,10 +714,13 @@ impl Queue for PgQueueHandle { .execute( &format!( "DELETE FROM {dead} - WHERE queue = $1 - AND dead_letter_retention_s IS NOT NULL - AND died_at <= $2::bigint - dead_letter_retention_s", - dead = table(&schema, tables::DEAD) + WHERE queue = $1 + AND dead_letter_retention_s IS NOT NULL + AND died_at <= {retained}", + dead = table(&schema, tables::DEAD), + retained = saturating_stamp_expr( + "($2::bigint)::numeric - dead_letter_retention_s::numeric", + ), ), &[&name, &now], ) @@ -909,14 +916,17 @@ impl JobHandle for PgJobHandle { &format!( "UPDATE {job} SET state = 'pending', - run_at = $2::bigint + $3::bigint, + run_at = {ready_at}, worker_id = NULL, claim_expires_at = NULL, claimed_at = NULL WHERE id = $1 AND worker_id = $4 AND state = 'processing' AND claim_expires_at >= $2", - job = table(&schema, tables::JOB) + job = table(&schema, tables::JOB), + ready_at = saturating_stamp_expr( + "($2::bigint)::numeric + ($3::bigint)::numeric" + ), ), &[&id, &now, &delay_s, &worker_id], ) @@ -997,11 +1007,13 @@ impl JobHandle for PgJobHandle { .execute( &format!( "UPDATE {job} - SET claim_expires_at = $3::bigint + $4::bigint + SET claim_expires_at = {deadline} WHERE id = $1 AND worker_id = $2 AND state = 'processing' AND claim_expires_at >= $3", - job = table(&schema, tables::JOB) + job = table(&schema, tables::JOB), + deadline = + saturating_stamp_expr("($3::bigint)::numeric + ($4::bigint)::numeric"), ), &[&id, &worker_id, &now, &extend], ) diff --git a/alkstore-postgres/src/resolution.rs b/alkstore-postgres/src/resolution.rs index 184c7be..c697e93 100644 --- a/alkstore-postgres/src/resolution.rs +++ b/alkstore-postgres/src/resolution.rs @@ -99,6 +99,23 @@ pub(crate) struct Stamps { pub dead_letter_retention_s: Option, } +/// The server-side stamp arithmetic's saturating guard (ADR-023 §2's +/// 2026-10-11 annotation — the pg realization of the same saturating +/// arithmetic the pinned Rust code of record runs; enqueue itself stays +/// Rust-resolved here, never a SQL-side sum). `` computes the +/// stamp sum/subtraction in `numeric` — a plain bigint add errors with +/// "bigint out of range" and the failure is sticky per row — and the +/// result clamps to the i64 domain before the `::bigint` cast. Call +/// sites keep every bound parameter's inferred type bigint (the +/// `($N::bigint)::numeric` chain, never a bare `$N::numeric`: the +/// client binds `i64` and a numeric-inferred parameter fails the +/// binding with `WrongType`). Sites of record: `queue.rs` (claim +/// deadline, retry reschedule, heartbeat renewal, retention sweep) and +/// `lock.rs` (ttl stamp, renew). +pub(crate) fn saturating_stamp_expr(expr: &str) -> String { + format!("LEAST(GREATEST({expr}, -9223372036854775808), 9223372036854775807)::bigint") +} + /// Resolve [`EnqueueOpts`] against the enqueue instant: the row's ready /// time (`delay` → `now + delay`, wins over `run_at` (absolute); /// neither → `now`) and the absolute row expiry (`expires` relative diff --git a/alkstore-postgres/src/store/lock_tests.rs b/alkstore-postgres/src/store/lock_tests.rs index dc1ea5c..58a6040 100644 --- a/alkstore-postgres/src/store/lock_tests.rs +++ b/alkstore-postgres/src/store/lock_tests.rs @@ -642,3 +642,67 @@ async fn closed_store_fails_closed_on_all_lock_ops() { drop_schema(&fx.admin, &fx.schema).await; } + +/// Acceptance: the stamp-arithmetic guard on the lock arms (ADR-023 +/// §2's 2026-10-11 annotation) — `try_lock`'s insert sum and `renew`'s +/// sum saturate to the i64 domain server-side, so `ttl = i64::MAX` +/// stamps the saturated `i64::MAX` expiry (the never-lapsing edge; +/// pre-guard the sticky "bigint out of range" failure failed both ops) +/// and a normal-ttl acquire rides exact. +#[tokio::test(flavor = "multi_thread")] +async fn extreme_ttl_stamps_saturate_without_overflow() { + let Some(fx) = Fixture::open(&instance_namer("lk")).await else { + eprintln!("skip: no harness server"); + return; + }; + + let lock = fx + .store + .try_lock("wide", "a", i64::MAX) + .await + .unwrap() + .expect("the i64::MAX ttl grants (no overflow failure)"); + assert_eq!( + lock_expires_at(&fx.admin, &fx.schema, "wide").await, + Some(i64::MAX), + "sat(now + i64::MAX) stamps the i64::MAX expiry" + ); + assert!( + lock.renew(i64::MAX).await.unwrap(), + "the i64::MAX ttl renew lands (no overflow failure)" + ); + assert_eq!( + lock_expires_at(&fx.admin, &fx.schema, "wide").await, + Some(i64::MAX), + "the renew arm saturates to the same edge" + ); + drop(lock); + + // A normal ttl rides exact (the guard never perturbs representable + // sums), and its extreme renew lands at the saturated edge. + let before = now_unix(); + let lock = fx + .store + .try_lock("normal", "a", 100) + .await + .unwrap() + .expect("the normal ttl grants"); + let expires = lock_expires_at(&fx.admin, &fx.schema, "normal") + .await + .unwrap(); + assert!( + (100..=102).contains(&(expires - before)), + "sat(now + 100) rides exact: {expires} vs before {before}" + ); + assert!( + lock.renew(i64::MAX).await.unwrap(), + "the extreme renew on the normal-ttl row lands" + ); + assert_eq!( + lock_expires_at(&fx.admin, &fx.schema, "normal").await, + Some(i64::MAX), + "the renew arm saturates on the normal row too" + ); + + drop_schema(&fx.admin, &fx.schema).await; +} diff --git a/alkstore-postgres/src/store/queue_tests.rs b/alkstore-postgres/src/store/queue_tests.rs index d8c7ff2..d685672 100644 --- a/alkstore-postgres/src/store/queue_tests.rs +++ b/alkstore-postgres/src/store/queue_tests.rs @@ -1437,3 +1437,216 @@ fn backoff_formula_unit_pins() { } } } + +/// Acceptance: the server-side stamp-arithmetic guard (ADR-023 §2's +/// 2026-10-11 annotation) — extreme i64 stamps and per-call arguments +/// are values whose server-side sums saturate to the i64 domain, never +/// "bigint out of range" failures, and the sticky-poison property (one +/// extreme-stamped row failing every later claim that touches it) is +/// retired. Pinned per site against the live server: an +/// `visibility_timeout_s = i64::MAX` row claims to the saturated +/// `i64::MAX` deadline and neighbors claim unaffected; extreme +/// heartbeat extends renew to the saturated edge and the representable +/// `now + i64::MIN` sum rides exact (Rust `saturating_add` parity), +/// the past deadline voiding through the validity predicate on the +/// next op; the retry override reschedules `Some(i64::MAX)` to +/// `i64::MAX` (and the negative-extreme delay rides exact to +/// ready-now); the retention `i64::MIN` arm sweeps every dead row +/// (`sat(now − MIN) = i64::MAX`) while the `Some(i64::MAX)` arm +/// deletes nothing. +#[tokio::test(flavor = "multi_thread")] +async fn extreme_stamps_saturate_and_no_longer_poison_queue_ops() { + let Some(fx) = Fixture::open(&instance_namer("qext")).await else { + eprintln!("skip: no harness server"); + return; + }; + let now = now_unix(); + + // Claim + heartbeat over an i64::MAX-stamped row: the claim UPDATE + // sums now + visibility_timeout_s server-side (pre-guard this + // failed sticky per row) — it must land, deadline at the saturated + // edge. + let queue = fx + .store + .queue( + "neverlaps", + QueueOpts { + visibility_timeout_s: i64::MAX, + ..opts() + }, + ) + .await + .unwrap(); + let id = queue + .enqueue(json!({"k": 1}), EnqueueOpts::default()) + .await + .unwrap(); + let h = queue + .claim_one("w1") + .await + .unwrap() + .expect("the extreme-stamped row claims (no overflow failure)"); + assert_eq!(h.job().id, id); + assert_eq!( + h.job().claim_expires_at, + Some(i64::MAX), + "the claim deadline saturates to the i64::MAX edge" + ); + + // A neighboring normal queue's row claims unaffected (the poison + // was per-row, not per-store; the other-rows-unaffected arm). + let normal = fx.store.queue("normal", opts()).await.unwrap(); + let normal_id = normal + .enqueue(json!({"k": 2}), EnqueueOpts::default()) + .await + .unwrap(); + let nh = normal + .claim_one("w1") + .await + .unwrap() + .expect("normal rows still claim"); + assert_eq!(nh.job().id, normal_id); + assert!( + nh.job().claim_expires_at.unwrap() >= now + 300, + "normal rows still claim (unaffected by the extreme row)" + ); + assert!(nh.ack().await.unwrap(), "the normal row acks"); + + // Heartbeats on the extreme row land: a normal renewal, then the + // extreme extend renewing to the saturated edge. + assert!(h.heartbeat(120).await.unwrap(), "heartbeat renews"); + let renewed = queue.get_job(id).await.unwrap().unwrap(); + assert!( + renewed.claim_expires_at.unwrap() > now, + "renewal is an absolute reset from now: {:?}", + renewed.claim_expires_at + ); + assert!( + h.heartbeat(i64::MAX).await.unwrap(), + "the i64::MAX extend renews (no overflow failure)" + ); + let renewed = queue.get_job(id).await.unwrap().unwrap(); + assert_eq!( + renewed.claim_expires_at, + Some(i64::MAX), + "sat(now + i64::MAX) = i64::MAX" + ); + + // The negative extreme: now + i64::MIN is representable, so the + // guard rides it exact (matching Rust `saturating_add` — no + // gratuitous clamp) and the past deadline it stamps voids the + // claim through the validity predicate on the next op. + let before_min = now_unix(); + assert!( + h.heartbeat(i64::MIN).await.unwrap(), + "the i64::MIN extend renews" + ); + let renewed = queue.get_job(id).await.unwrap().unwrap(); + let offset = renewed.claim_expires_at.unwrap().wrapping_sub(i64::MIN); + assert!( + (before_min..=before_min + 2).contains(&offset), + "sat(now + i64::MIN) rides exact: deadline {:?} vs before {before_min}", + renewed.claim_expires_at + ); + assert!( + !h.heartbeat(60).await.unwrap(), + "the past saturated deadline voids the claim on the next op" + ); + + // Retry override extremes: `Some(i64::MIN)` rides exact to + // ready-now (the row reclaims immediately), `Some(i64::MAX)` + // reschedules to the ready-~never saturated edge. + let id2 = normal + .enqueue(json!({"k": 5}), EnqueueOpts::default()) + .await + .unwrap(); + let h = normal.claim_one("w1").await.unwrap().unwrap(); + let before = now_unix(); + assert!( + h.retry(Some("boom".to_string()), Some(i64::MIN)) + .await + .unwrap(), + "the negative-extreme delay retry lands" + ); + let job = normal.get_job(id2).await.unwrap().unwrap(); + assert_eq!(job.state, JobState::Pending); + let offset = job.run_at.wrapping_sub(i64::MIN); + assert!( + (before..=before + 2).contains(&offset), + "sat(now + i64::MIN) rides exact: run_at {} vs now {before}", + job.run_at + ); + let h = normal + .claim_one("w1") + .await + .unwrap() + .expect("the ready-now row reclaims"); + assert_eq!(h.job().id, id2); + assert!( + h.retry(Some("boom".to_string()), Some(i64::MAX)) + .await + .unwrap(), + "the positive-extreme delay retry lands" + ); + let job = normal.get_job(id2).await.unwrap().unwrap(); + assert_eq!(job.run_at, i64::MAX, "sat(now + i64::MAX) = i64::MAX"); + + // Retention arms: `Some(i64::MIN)` sweeps every dead row + // (sat(now − MIN) = i64::MAX, the sweep-all edge — pre-guard this + // sweep failed sticky with the overflow error); + // `Some(i64::MAX)` deletes nothing (sat(now − MAX) = i64::MIN). + let sweepall = fx + .store + .queue( + "sweepall", + QueueOpts { + dead_letter_retention_s: Some(i64::MIN), + ..opts() + }, + ) + .await + .unwrap(); + let swept_id = sweepall + .enqueue(json!({"k": 3}), EnqueueOpts::default()) + .await + .unwrap(); + let h = sweepall.claim_one("w1").await.unwrap().unwrap(); + assert!(h.fail(None).await.unwrap(), "the dead row exists"); + assert!( + sweepall.sweep_expired().await.unwrap() >= 1, + "the retention i64::MIN sweep lands (no overflow failure)" + ); + assert!( + sweepall.get_job(swept_id).await.unwrap().is_none(), + "sat(now − i64::MIN) = i64::MAX sweeps the dead row" + ); + + let sweepnone = fx + .store + .queue( + "sweepnone", + QueueOpts { + dead_letter_retention_s: Some(i64::MAX), + ..opts() + }, + ) + .await + .unwrap(); + let kept_id = sweepnone + .enqueue(json!({"k": 4}), EnqueueOpts::default()) + .await + .unwrap(); + let h = sweepnone.claim_one("w1").await.unwrap().unwrap(); + assert!(h.fail(None).await.unwrap()); + assert_eq!( + sweepnone.sweep_expired().await.unwrap(), + 0, + "sat(now − i64::MAX) = i64::MIN deletes nothing" + ); + assert!( + sweepnone.get_job(kept_id).await.unwrap().is_some(), + "the retention i64::MAX dead row survives" + ); + + drop_schema(&fx.admin, &fx.schema).await; +} diff --git a/tasks/pg-fix-bigint-stamp-overflow.md b/tasks/pg-fix-bigint-stamp-overflow.md index 8bedfd7..85dc611 100644 --- a/tasks/pg-fix-bigint-stamp-overflow.md +++ b/tasks/pg-fix-bigint-stamp-overflow.md @@ -1,7 +1,7 @@ --- id: pg-fix-bigint-stamp-overflow name: Guard pg's server-side stamp arithmetic against i64 extremes (review 004 item 5, implementation) -status: pending +status: completed depends_on: [pg-decide-stamp-arithmetic-side] scope: moderate risk: medium @@ -22,7 +22,7 @@ out of range", sticky per row (one extreme-stamped row poisons every later claim/heartbeat/sweep that touches it). Per the decision of record: clamp/validate at the enqueue seam, or -`LEAST`/`GREATEST`-guard the server expressions — do not re-decide; +`LEAST`/`GREATEST`-guard the server expressions — do not re-deciding; if the implementation surfaces a reason the decision can't stand as written, stop and record the discrepancy instead of deviating. @@ -36,16 +36,16 @@ against it. ## Acceptance Criteria -- [ ] The decided guard lands at the decided sites; no re-deciding +- [x] The decided guard lands at the decided sites; no re-deciding (any discrepancy recorded, not deviated around) -- [ ] An `i64::MAX`-stamped row no longer makes claim/heartbeat +- [x] An `i64::MAX`-stamped row no longer makes claim/heartbeat UPDATEs fail — pinned by test against the live server, with the other-rows-unaffected assertion -- [ ] The retention `i64::MIN` arm pinned (sweep behavior per the +- [x] The retention `i64::MIN` arm pinned (sweep behavior per the decided semantics) -- [ ] The decision's annotation/contract text updated if the +- [x] The decision's annotation/contract text updated if the reachable failure set changed -- [ ] `cargo test -p alkstore-postgres` (harness server), clippy +- [x] `cargo test -p alkstore-postgres` (harness server), clippy `-D warnings`, fmt clean; workspace gates green server-less ## References @@ -59,8 +59,74 @@ against it. ## Notes -> To be filled by implementation agent +**The decision stands as written; one form refinement, recorded +here.** The decision's verbatim guard form writes the sum as +`$N::numeric`, but a bare `$N::numeric` cast changes Postgres's +parameter-type inference — the parameter infers `numeric` and +tokio-postgres's `i64` binding fails with `WrongType { postgres: +Numeric, rust: "i64" }` (caught by the first live test run, both +tests). The realized form is the **`($N::bigint)::numeric` chain**: +the inner `::bigint` cast pins the parameter's inferred type (the +client keeps binding `i64`), the outer `::numeric` lifts the +arithmetic into the overflow-free domain. Clamp domain, sites, and +extreme-stamped observables are unchanged — this refines *how* the +sum computes, not the decision. The pin lives in +`resolution.rs`'s `saturating_stamp_expr` doc (the one owner both +files call). + +**One owner for the guard.** The form lives in +`resolution.rs`::`saturating_stamp_expr` (the engine's arithmetic +home, "one owner per formula" — ADR-012 §2); the six sites call it: +`queue.rs:385` (claim deadline), `:721` (retention sweep), +`:927` (retry reschedule), `:1016` (heartbeat renewal), `lock.rs:95` +(try_lock insert), `:222` (renew). Line numbers drifted a few lines +from the decision's refs (the doc-only decision pinned decision-time +lines); the site *set* is exactly the decision's six. + +**No contract/annotation text change.** The reachable-failure-set +change was recorded by the decision itself (ADR-023 §2's 2026-10-11 +annotation: "bigint out of range" leaves pg's reachable surface); +the implementation changes nothing further — the acceptance +criterion's "if changed" clause is satisfied vacuously. + +**Test coverage beyond the description's minimum**: the description +pins claim/heartbeat (`i64::MAX` stamp) and the retention +`i64::MIN` arm; the landed test also pins the retention +`Some(i64::MAX)` delete-nothing arm, the retry override extremes +(`Some(i64::MIN)` rides exact → ready-now reclaim; +`Some(i64::MAX)` → `i64::MAX` ready-~never), and both lock arms +(`ttl = i64::MAX` insert and renew → `i64::MAX` expiry, normal ttl +rides exact). All six decision sites are behaviorally pinned, not +just the four the review named. + +**One flake fixed before merge**: the first full-crate run failed +once in the new queue test — the `heartbeat(i64::MIN)` exact-ride +window was anchored to the test-start clock read and, under the +loaded parallel suite, the server-side `now` inside the heartbeat +drifted past `now + 2`. Re-anchored to a fresh read immediately +before the heartbeat (the family's bounded-wait convention); two +consecutive full-crate harness runs green after the fix. ## Summary -> To be filled on completion \ No newline at end of file +The decision-of-record guard landed: every pg server-side stamp +sum/subtraction computes in `numeric` and saturates to the i64 +domain before the `::bigint` cast — +`LEAST(GREATEST(, -9223372036854775808), 9223372036854775807)::bigint` +— via one owner (`resolution.rs::saturating_stamp_expr`) called at +all six sites of record (claim deadline, retry reschedule, heartbeat +renewal, retention sweep; lock insert + renew). The sticky-poison +property is retired: an `i64::MAX`-stamped row claims to the +saturated `i64::MAX` deadline, heartbeats land, neighbors claim +unaffected, `"bigint out of range"` is unreachable through every +guarded op. + +Verified: `cargo test -p alkstore-postgres` against the live harness +server (`pglo-poc` :15432) — 135 lib + 25 suite + 9 schema green, +two consecutive full runs (the two new tests: +`extreme_stamps_saturate_and_no_longer_poison_queue_ops`, +`extreme_ttl_stamps_saturate_without_overflow`); workspace gates +server-less green (build, full `cargo test` incl. the skipped-clean +pg rows, clippy `-D warnings` all targets, fmt); fuzz corpus gate +green (`fuzz/shared` 6 passed); `RUSTDOCFLAGS="-D warnings" cargo +doc --no-deps` clean. \ No newline at end of file