diff --git a/alkstore-contract-suite/src/lib.rs b/alkstore-contract-suite/src/lib.rs index 689ee0e..251b8bd 100644 --- a/alkstore-contract-suite/src/lib.rs +++ b/alkstore-contract-suite/src/lib.rs @@ -38,9 +38,9 @@ pub mod version_stamp; pub use factory::StoreFactory; pub use properties::{ - concurrent_try_lock_loser_is_a_value, duration_refusal_on_non_positive_ttl, - enqueue_opts_resolution, extent_clamp_semantics, in_tx_reads_see_own_writes, - job_handle_validity_predicate, lock_ttl_expiry_and_reacquisition, + backoff_curve_equivalence, concurrent_try_lock_loser_is_a_value, + duration_refusal_on_non_positive_ttl, enqueue_opts_resolution, extent_clamp_semantics, + in_tx_reads_see_own_writes, job_handle_validity_predicate, lock_ttl_expiry_and_reacquisition, name_validation_rejects_empty_and_reserved, outbox_enqueue_tx_commit_atomicity, payload_round_trip_stores_exact_encoding, payload_too_large_never_produced_on_sqlite, payload_too_large_produced_on_pg, publish_with_key_tx_commit_atomicity, diff --git a/alkstore-contract-suite/src/properties.rs b/alkstore-contract-suite/src/properties.rs index d2b8a55..d9fb8ec 100644 --- a/alkstore-contract-suite/src/properties.rs +++ b/alkstore-contract-suite/src/properties.rs @@ -894,15 +894,22 @@ pub async fn in_tx_reads_see_own_writes(factory: &dyn StoreFactory) { /// `EnqueueOpts::max_attempts` override riding over them — `get_job` /// shows resolved values, never the raw opts. /// -/// The `run_at`-alone-literal and neither-field-resolves-to-now legs -/// are wall-clock-adjacent; they pin with wave 5's equivalence row -/// (this row's precedence/resolution legs are the engine-agnostic -/// claim). +/// Wave-5 completion — the two clock-adjacent legs the earlier note +/// deferred pin here: the `run_at`-alone literal leg (a future +/// `run_at` is the row's ready time *verbatim* — no arithmetic, a +/// deterministic equality; a past `run_at` is ready *now* — the +/// boundary-total rule, claimable through the `run_at <= now` +/// predicate within a bounded wait) and the neither-field leg (ready +/// now at the enqueue instant — tolerance-bounded proximity of the +/// stamp to the clock read, never a tight timing assert). /// -/// Contract stamp: ADR-020 §1 (delay wins over `run_at`); §2 +/// Contract stamp: ADR-020 §1 (delay wins over `run_at`; `run_at` +/// alone is literal; neither field resolves to now); §2 /// (relative `expires` → absolute row value); §3/§3a (the derived /// default stamps — plain queue 300/3/5/none, per-job override); -/// ADR-012 §2 (engine-side arithmetic pinned equivalent). +/// ADR-023 §2 (the boundary-total rule — a past ready time is ready +/// now, never rejected); ADR-012 §2 (engine-side arithmetic pinned +/// equivalent). pub async fn enqueue_opts_resolution(factory: &dyn StoreFactory) { let store = factory.open().await.expect("factory opens a store"); @@ -993,6 +1000,259 @@ pub async fn enqueue_opts_resolution(factory: &dyn StoreFactory) { "the per-job override rides over the derived defaults, others untouched" ); + // Wave-5 clock legs (the run_at-alone and neither-field pins). + // Future run_at alone: the ready time is the literal — stored + // verbatim, no arithmetic, no clock read involved (a deterministic + // equality, not a timing assert). + let literal = 1_800_000_000; + let literal_id = queue + .enqueue( + json!({"leg": "run-at-literal"}), + EnqueueOpts { + run_at: Some(literal), + ..Default::default() + }, + ) + .await + .unwrap(); + let literal_job = queue.get_job(literal_id).await.unwrap().unwrap(); + assert_eq!( + literal_job.run_at, literal, + "run_at alone is the row's ready time verbatim (delay absent)" + ); + + // Past run_at alone: ready now (the boundary-total rule) — + // claimable within a bounded wait. The claim ordering (run_at ASC) + // hands it out ahead of the every same-now row above, so the + // observation is unambiguous even against the row's other legs. + let past = unix_now() - 30; + let past_id = queue + .enqueue( + json!({"leg": "run-at-past"}), + EnqueueOpts { + run_at: Some(past), + ..Default::default() + }, + ) + .await + .unwrap(); + assert_eq!( + queue.get_job(past_id).await.unwrap().unwrap().run_at, + past, + "a past run_at stores the literal too — ready-now is the claim semantics" + ); + let claimed_past = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if let Some(h) = queue.claim_one("wave5-past").await.unwrap() { + let id = h.job().id; + if id == past_id { + break id; + } + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + }) + .await + .expect("a past run_at resolves ready-now: claimable within the bounded wait"); + assert_eq!( + claimed_past, past_id, + "the past-run_at row is the ready-now claim (ordering: run_at ASC)" + ); + + // Neither field: ready now at the enqueue instant — the stamp sits + // close to the clock read (tolerance-bounded proximity, never a + // tight timing assert). + let neither_id = queue + .enqueue(json!({"leg": "neither"}), EnqueueOpts::default()) + .await + .unwrap(); + let now = unix_now(); + assert!( + queue + .get_job(neither_id) + .await + .unwrap() + .unwrap() + .run_at + .abs_diff(now) + <= 5, + "neither field resolves to now: the stamp is within the tolerance \ + of the clock read" + ); + + factory.teardown().await.expect("factory teardown"); +} + +/// The backoff row's bounded-wait claim: poll `claim_one` on `queue` +/// until `id` comes out and return that claim's handle (claim +/// ordering keeps single-job queues unambiguous; the deadline is the +/// safety bound — the assertions are on state outcomes, and the claim +/// is consumed here, not re-taken afterwards). +async fn await_claimable( + queue: &dyn alkstore::Queue, + id: i64, + max_wait: std::time::Duration, + label: &str, +) -> Box { + let deadline = std::time::Instant::now() + max_wait; + loop { + match queue.claim_one("backoff-w").await.unwrap() { + Some(h) if h.job().id == id => return h, + _ => {} + } + assert!( + std::time::Instant::now() < deadline, + "{label}: the row never became ready within the bounded wait" + ); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } +} + +/// **Backoff-curve equivalence across engines** — the equal-jitter +/// exponential curve (ADR-010 §3): `retry(err, None)` computes +/// `delay ∈ [base·2^(a−1)/2, base·2^(a−1)]` from the job's stamped +/// `backoff_base_s` (the attempt index `a` the row's post-claim +/// `attempts`), capped at 1 hour — *the range, not a jitter label, is +/// the definition*, and identical bounds across engines is the +/// equivalence pin (ADR-012 §2: engine-side arithmetic, suite-pinned +/// identical outputs). The row drives claim → `retry(err, None)` +/// cycles on a queue with `backoff_base_s = 2` (integer-second +/// resolution: the ranges [1,2], [2,4], [4,8] land across the first +/// three attempts), reading each computed delay back through +/// `get_job`'s resolved `run_at` minus a clock read taken before the +/// retry, asserting range membership per attempt — with a +/// one-second straddle tolerance on the upper side only (the retry's +/// internal clock read and the suite's may cross a wall second; the +/// lower side is race-free, the retry's clock read cannot precede the +/// suite's). The inter-attempt waits are tolerance-bounded and +/// sequential (the row waits for each `run_at` to pass through +/// bounded-wait claims). +/// +/// The explicit-delay override legs: `retry(err, Some(d))` is honored +/// verbatim — `Some(5)` lands the row ready at the retry instant plus +/// exactly 5 s (the same straddle tolerance), and `Some(0)` resolves +/// ready-now (the boundary-total rule: a zero delay is claimable at +/// the retry's own second, within a bounded wait). +/// +/// Cap-leg disposition: the 1-hour cap is not suite-pinnable (reaching +/// capped attempts requires waiting out minute-scale delays); it stays +/// pinned engine-side — both engines' unit tests pin the curve against +/// the ADR text including the cap. +/// +/// Contract stamp: ADR-010 §3 (the equal-jitter exponential, the range +/// as definition, the 1-hour cap); ADR-020 §1 (the explicit-delay +/// override resolves to the row's ready time — `Some(0)` ready now); +/// ADR-012 §2 (engine-side arithmetic pinned identical); ADR-023 §2 +/// (the boundary-total rule — a zero delay is ready now, never +/// rejected). +pub async fn backoff_curve_equivalence(factory: &dyn StoreFactory) { + let store = factory.open().await.expect("factory opens a store"); + let queue = store + .queue( + "curve", + alkstore::QueueOpts { + max_attempts: 7, + backoff_base_s: 2, + ..Default::default() + }, + ) + .await + .unwrap(); + let id = queue + .enqueue(json!({"leg": "curve"}), EnqueueOpts::default()) + .await + .unwrap(); + + // Three curve cycles at the first three attempts (base 2 s: + // ranges [1,2], [2,4], [4,8] — the integer-second row resolution + // kept honest by the 1-second straddle tolerance on the upper + // side). The now-read before the retry makes the lower bound + // race-free: the retry's internal clock read cannot precede it, so + // the observed value over-reads the delay, never under-reads. + for (attempt, lower, upper) in [(1i64, 1i64, 2i64), (2, 2, 4), (3, 4, 8)] { + let held = await_claimable( + queue.as_ref(), + id, + std::time::Duration::from_secs(upper as u64 + 10), + &format!("attempt {attempt}"), + ) + .await; + assert_eq!( + held.job().attempts, + attempt, + "the attempt index is the claim count" + ); + let now = unix_now(); + assert!( + held.retry(Some("boom".to_string()), None) + .await + .expect("an in-window retry lands") + ); + let job = queue + .get_job(id) + .await + .unwrap() + .expect("the retried row lives"); + assert_eq!( + job.state, + JobState::Pending, + "the retry left the row pending" + ); + let observed = job.run_at - now; + assert!( + (lower..=upper + 1).contains(&observed), + "attempt {attempt}: the computed delay must land in \ + [{lower}, {upper}] (one-second straddle tolerance), got {observed}" + ); + } + + // Explicit-delay override: Some(5) is honored — ready at the retry + // instant plus exactly 5 s (the same straddle tolerance), not the + // attempt-4 curve range [8, 16]. + let held = await_claimable( + queue.as_ref(), + id, + std::time::Duration::from_secs(10), + "Some(5) leg", + ) + .await; + let now = unix_now(); + assert!( + held.retry(Some("boom".to_string()), Some(5)) + .await + .expect("an in-window retry lands") + ); + let job = queue.get_job(id).await.unwrap().unwrap(); + let observed = job.run_at - now; + assert!( + (5..=6).contains(&observed), + "retry(err, Some(5)) is honored verbatim (± one second straddle), \ + got {observed}" + ); + + // Some(0): the boundary-total rule — ready now. The retry's own + // second already satisfies the claim predicate, so the bounded + // wait is short and the row lands promptly. + let held = await_claimable( + queue.as_ref(), + id, + std::time::Duration::from_secs(8), + "Some(0) leg", + ) + .await; + assert!( + held.retry(Some("boom".to_string()), Some(0)) + .await + .expect("an in-window retry lands") + ); + await_claimable( + queue.as_ref(), + id, + std::time::Duration::from_secs(5), + "Some(0) resolves ready-now (the boundary-total rule)", + ) + .await; + factory.teardown().await.expect("factory teardown"); } diff --git a/alkstore-postgres/tests/contract_suite.rs b/alkstore-postgres/tests/contract_suite.rs index e188f4a..f6fd9a9 100644 --- a/alkstore-postgres/tests/contract_suite.rs +++ b/alkstore-postgres/tests/contract_suite.rs @@ -163,16 +163,16 @@ async fn harness_ready() -> bool { mod rows { use alkstore_contract_suite::properties::{ - concurrent_try_lock_loser_is_a_value, drop_rollback_leaves_no_ghosts, - duration_refusal_on_non_positive_ttl, enqueue_opts_resolution, extent_clamp_semantics, - in_tx_reads_see_own_writes, job_handle_validity_predicate, - lock_ttl_expiry_and_reacquisition, name_validation_rejects_empty_and_reserved, - outbox_enqueue_tx_commit_atomicity, payload_round_trip_stores_exact_encoding, - payload_too_large_produced_on_pg, publish_with_key_tx_commit_atomicity, - queue_depth_reclaim_and_dead_letter, receiver_close_and_save_arms, - scheduler_boundary_fires, scheduler_bounded_catchup, scheduler_leadership_discipline, - stream_ordering_equivalence, sweep_no_stranded_rows, trim_to_semantics, - wake_receiver_shapes, with_tx_panicking_closure_rolls_back, + backoff_curve_equivalence, concurrent_try_lock_loser_is_a_value, + drop_rollback_leaves_no_ghosts, duration_refusal_on_non_positive_ttl, + enqueue_opts_resolution, extent_clamp_semantics, in_tx_reads_see_own_writes, + job_handle_validity_predicate, lock_ttl_expiry_and_reacquisition, + name_validation_rejects_empty_and_reserved, outbox_enqueue_tx_commit_atomicity, + payload_round_trip_stores_exact_encoding, payload_too_large_produced_on_pg, + publish_with_key_tx_commit_atomicity, queue_depth_reclaim_and_dead_letter, + receiver_close_and_save_arms, scheduler_boundary_fires, scheduler_bounded_catchup, + scheduler_leadership_discipline, stream_ordering_equivalence, sweep_no_stranded_rows, + trim_to_semantics, wake_receiver_shapes, with_tx_panicking_closure_rolls_back, }; use super::PgFactory; @@ -243,6 +243,11 @@ mod rows { enqueue_opts_resolution, "row-opts-resolution" ); + harness_row!( + row_backoff_curve_equivalence, + backoff_curve_equivalence, + "row-backoff-curve" + ); harness_row!( row_receiver_close_and_save_arms, receiver_close_and_save_arms, diff --git a/alkstore-sqlite/tests/contract_suite.rs b/alkstore-sqlite/tests/contract_suite.rs index 6f74d16..bd3ed6c 100644 --- a/alkstore-sqlite/tests/contract_suite.rs +++ b/alkstore-sqlite/tests/contract_suite.rs @@ -67,16 +67,16 @@ impl StoreFactory for SqliteFactory { mod rows { use alkstore_contract_suite::properties::{ - concurrent_try_lock_loser_is_a_value, drop_rollback_leaves_no_ghosts, - duration_refusal_on_non_positive_ttl, enqueue_opts_resolution, extent_clamp_semantics, - in_tx_reads_see_own_writes, job_handle_validity_predicate, - lock_ttl_expiry_and_reacquisition, name_validation_rejects_empty_and_reserved, - outbox_enqueue_tx_commit_atomicity, payload_round_trip_stores_exact_encoding, - payload_too_large_never_produced_on_sqlite, publish_with_key_tx_commit_atomicity, - queue_depth_reclaim_and_dead_letter, receiver_close_and_save_arms, - scheduler_boundary_fires, scheduler_bounded_catchup, scheduler_leadership_discipline, - stream_ordering_equivalence, sweep_no_stranded_rows, trim_to_semantics, - wake_receiver_shapes, with_tx_panicking_closure_rolls_back, + backoff_curve_equivalence, concurrent_try_lock_loser_is_a_value, + drop_rollback_leaves_no_ghosts, duration_refusal_on_non_positive_ttl, + enqueue_opts_resolution, extent_clamp_semantics, in_tx_reads_see_own_writes, + job_handle_validity_predicate, lock_ttl_expiry_and_reacquisition, + name_validation_rejects_empty_and_reserved, outbox_enqueue_tx_commit_atomicity, + payload_round_trip_stores_exact_encoding, payload_too_large_never_produced_on_sqlite, + publish_with_key_tx_commit_atomicity, queue_depth_reclaim_and_dead_letter, + receiver_close_and_save_arms, scheduler_boundary_fires, scheduler_bounded_catchup, + scheduler_leadership_discipline, stream_ordering_equivalence, sweep_no_stranded_rows, + trim_to_semantics, wake_receiver_shapes, with_tx_panicking_closure_rolls_back, }; use super::SqliteFactory; @@ -135,6 +135,11 @@ mod rows { enqueue_opts_resolution(&factory("row-opts-resolution")).await; } + #[tokio::test(flavor = "multi_thread")] + async fn row_backoff_curve_equivalence() { + backoff_curve_equivalence(&factory("row-backoff-curve")).await; + } + #[tokio::test(flavor = "multi_thread")] async fn row_receiver_close_and_save_arms() { receiver_close_and_save_arms(&factory("row-receiver-arms")).await; diff --git a/tasks/suite-opts-backoff-rows.md b/tasks/suite-opts-backoff-rows.md index 9d3ac24..d252622 100644 --- a/tasks/suite-opts-backoff-rows.md +++ b/tasks/suite-opts-backoff-rows.md @@ -1,7 +1,7 @@ --- id: suite-opts-backoff-rows name: Contract-suite rows — backoff-curve equivalence + the deferred enqueue-opts clock legs -status: pending +status: completed depends_on: [] scope: narrow risk: low @@ -76,6 +76,80 @@ Work: > To be filled by implementation agent +- **The 1-hour cap leg is not suite-pinned** (as pre-decided): reaching + capped attempts requires waiting out minute-scale delays. The cap + stays pinned engine-side — both engines' unit tests pin the curve + against the ADR text including the cap (SQLite's + `backoff_curve_lands_in_the_pinned_range` and the pg twin, verified + in the wave-3/4 rounds); the disposition is recorded in the row's + doc comment for the review gate. +- **Observation tolerance convention for the curve row**: the + computed delay is read back as `get_job`'s `run_at` minus a clock + read taken *before* the `retry(err, None)` call. That direction is + deliberate and race-free on the lower side: the retry's internal + clock read (SQLite's `unixepoch()` in-statement, pg's Rust-side + engine read) cannot precede the suite's, so the observed value + *over-reads* the delay, never under-reads — the pinned lower bound + is exact. The upper bound carries a one-second straddle tolerance + (retry commit and suite read may cross a wall second, and the + engines' stamps are integer seconds, ADR-010 §3's integerized + resolution). Asserted as a `(lower..=upper+1)` range membership; the + *identical* assertion on both engines is the equivalence pin + (ADR-012 §2's identical-outputs rule) — the row body is shared, so + the bounds are identical by construction. +- **Base 2 s kept the integer-second row resolution honest** per the + task text: ranges [1,2], [2,4], [4,8] across attempts 1–3 (the a=1 + pin is inherently weak at one-second resolution — lower bound 1 with + upper ±1 — but the attempt-2/3 ranges are genuinely discriminating, + and the explicit-delay legs pin the exact-literal resolution). +- **Past-run_at leg stores the literal too**: the pin on the past leg + is the *claim semantics* (ready-now through the `run_at <= now` + predicate, bounded-wait claimable) and the stored literal (no + arithmetic on any form) — "resolves to ready now" is the claim- + predicate reading of the boundary-total rule (ADR-023 §2), not a + clamping of the stored value; both engines store verbatim + (`resolution.rs` twins verified). +- **Claim-polling observation posture**: both rows' ready-now waits + are bounded `claim_one` poll loops (100/50 ms cadence), deadline + asserts only — no timing-value assertions, single-store sequential + drive within the module doc's determinism posture. +- **One unreproducibility note on the helper**: the first draft made + `await_claimable` re-claim after the bounded wait, which found + nothing (the bounded wait consumed the claim); the helper returns + the claimed handle. Found on the first SQLite run, no engine-side + hazard. +- No ADR amendments needed; no divergences found (both engines' + arithmetic agreed out of the box — the equivalence pin held on the + first full run pairs). + ## Summary -> To be filled on completion \ No newline at end of file +> To be filled on completion + +`enqueue_opts_resolution` extended in place with the wave-5 +completion legs (its deferral note replaced by the completion +statement, ADR-023 §2 stamp accumulated): the `run_at`-alone literal +leg (future `run_at` = the row's ready time verbatim — a +deterministic equality, no clock read involved; past `run_at` stored +verbatim too and *claimable now* within a bounded-wait claim, the +run_at-ASC ordering making the observation unambiguous among the +row's other legs) and the neither-field leg (ready now at the +enqueue instant, `abs_diff(now) <= 5` tolerance-bounded proximity — +never a tight timing assert). New `backoff_curve_equivalence` row: +claim → `retry(err, None)` cycles on a `backoff_base_s = 2` queue +(ranges [1,2], [2,4], [4,8] per attempt, asserted as `(lower..=upper+1)` +range membership of `run_at − now_pre` with the race-free lower +bound and one-second straddle tolerance documented), reading each +computed delay back through `get_job`'s resolved `run_at`; the +explicit-delay override legs ride the same row (`Some(5)` honored +verbatim in [5, 6]; `Some(0)` resolving ready-now per the +boundary-total rule, claimed within a bounded wait). Stamped ADR-010 +§3, ADR-020 §1, ADR-012 §2, ADR-023 §2. Both rows wired into both +engines' `contract_suite.rs` targets (SQLite `#[tokio::test]`, pg +`harness_row!`) — 25 rows per column, up from 24. Verified: SQLite +column green server-less (25/25 suite, 189 lib), pg column green vs +the harness server (postgres/poc@:15432 — 25/25 suite, 121 lib + 9 +schema), 3 solo re-runs of each new/extended row per engine +(determinism), server-less pg column skips cleanly (25 "passed" via +the reachability gate), `cargo test -p alkstore-sqlite -p +alkstore-postgres` green, clippy `-D warnings`, fmt clean. \ No newline at end of file