diff --git a/alkstore-contract-suite/src/lib.rs b/alkstore-contract-suite/src/lib.rs index 6d829b8..7d2d38f 100644 --- a/alkstore-contract-suite/src/lib.rs +++ b/alkstore-contract-suite/src/lib.rs @@ -39,7 +39,8 @@ pub mod version_stamp; pub use factory::StoreFactory; pub use properties::{ duration_refusal_on_non_positive_ttl, enqueue_opts_resolution, extent_clamp_semantics, - in_tx_reads_see_own_writes, name_validation_rejects_empty_and_reserved, - payload_round_trip_stores_exact_encoding, payload_too_large_never_produced_on_sqlite, - payload_too_large_produced_on_pg, receiver_close_and_save_arms, + in_tx_reads_see_own_writes, job_handle_validity_predicate, + name_validation_rejects_empty_and_reserved, payload_round_trip_stores_exact_encoding, + payload_too_large_never_produced_on_sqlite, payload_too_large_produced_on_pg, + queue_depth_reclaim_and_dead_letter, receiver_close_and_save_arms, sweep_no_stranded_rows, }; diff --git a/alkstore-contract-suite/src/properties.rs b/alkstore-contract-suite/src/properties.rs index 1361770..a9857f7 100644 --- a/alkstore-contract-suite/src/properties.rs +++ b/alkstore-contract-suite/src/properties.rs @@ -996,6 +996,548 @@ pub async fn payload_too_large_produced_on_pg(factory: &dyn StoreFactory) { factory.teardown().await.expect("factory teardown"); } +/// The suite's lapse-tolerance sleep: bounded well past a 1-second +/// stamp's resolution (a stamp set on second `T` has lapsed for any +/// clock read `> T + 1`; 2.5 s guarantees it, 3.5 s the `expires: 2` +/// stamps). Timing posture: state outcomes are asserted, never +/// durations. +fn past_stamp_sleep(millis: u64) { + std::thread::sleep(std::time::Duration::from_millis(millis)); +} + +/// **Job-handle validity predicate** — the uniform rule (processing +/// state + unexpired claim deadline) on every handle op: an in-window +/// heartbeat lands (renewal) and an in-window ack deletes +/// (delete-on-ack); the late heartbeat is refused exactly at the +/// lapsed claim — the short-stamp row claims with a 1 s visibility +/// stamp and tolerance-sleeps past it (the sleep-then-assert-state +/// shape is sequential single-store and fits the suite's determinism +/// posture); the post-lapse ack does not land (at-least-once: the +/// worker's ack does not land after lapse), and `retry`'s refusal is +/// `false` with the row untouched (state, attempts, claimant, and +/// deadline all as the claim left them). `fail(None)` after lapse is +/// refused identically — deadline lapse refuses *all* ops, not just +/// heartbeat. `ack_batch` applies the predicate per id: a lapsed id +/// and a never-claimed id are silently not counted — the returned +/// count reflects only the live acks. +/// +/// The engine-default strings land identically: `fail(None)` records +/// `"failed"`, and retry-at-exhausted-budget dead-letters with +/// `"max attempts exceeded"` (the reclaim-exhaustion arm of that +/// string pins in the queue-depth row's exhaustion leg). +/// +/// Contract stamp: ADR-010 §2 (the uniform validity predicate — the +/// late-heartbeat refusal, post-lapse ack refusal); ADR-010 §1 +/// (delete-on-ack; `ack_batch`'s per-id predicate, partial success +/// ordinary); ADR-010 §4 (the engine-default strings); ADR-019 §3 +/// (the claim handle's op surface — one-shot consumables, repeatable +/// heartbeat); ADR-008 §5 (refusals are `false`, never errors). +pub async fn job_handle_validity_predicate(factory: &dyn StoreFactory) { + let store = factory.open().await.expect("factory opens a store"); + + let slow = store + .queue("handles", alkstore::QueueOpts::default()) + .await + .unwrap(); + let fast = store + .queue( + "lapse", + alkstore::QueueOpts { + visibility_timeout_s: 1, + ..Default::default() + }, + ) + .await + .unwrap(); + + // In-window ops land: renewal + delete-on-ack under the live + // claim. + let live_id = fast + .enqueue(json!({"leg": "in-window"}), EnqueueOpts::default()) + .await + .unwrap(); + let held = fast + .claim_one("w1") + .await + .unwrap() + .expect("a ready row claims"); + assert_eq!(held.job().id, live_id); + assert_eq!(held.job().state, JobState::Processing); + assert!( + held.heartbeat(600).await.unwrap(), + "an in-window heartbeat lands (renewal)" + ); + let mid = fast + .get_job(live_id) + .await + .unwrap() + .expect("the row still lives mid-claim"); + assert_eq!(mid.state, JobState::Processing); + assert!(held.ack().await.unwrap(), "an in-window ack lands"); + assert!( + fast.get_job(live_id).await.unwrap().is_none(), + "delete-on-ack completed work leaves no row" + ); + + // Lapse legs: one claim per refusal op, all stamped with the 1 s + // visibility, one tolerance sleep past every deadline at once. + let ack_refused = fast + .enqueue(json!({"leg": "ack-refused"}), EnqueueOpts::default()) + .await + .unwrap(); + let retry_refused = fast + .enqueue(json!({"leg": "retry-refused"}), EnqueueOpts::default()) + .await + .unwrap(); + let fail_refused = fast + .enqueue(json!({"leg": "fail-refused"}), EnqueueOpts::default()) + .await + .unwrap(); + let lapsed_ack = fast.claim_one("w2").await.unwrap().unwrap(); + let lapsed_retry = fast.claim_one("w2").await.unwrap().unwrap(); + let lapsed_fail = fast.claim_one("w2").await.unwrap().unwrap(); + assert_eq!( + ( + lapsed_ack.job().id, + lapsed_retry.job().id, + lapsed_fail.job().id + ), + (ack_refused, retry_refused, fail_refused), + "claims hand out in enqueue order (equal priority)" + ); + let retry_before = lapsed_retry.job().clone(); + past_stamp_sleep(2_500); + assert!( + !lapsed_ack.heartbeat(600).await.unwrap(), + "late heartbeat must be refused: the deadline has lapsed" + ); + assert!( + !lapsed_retry + .retry(Some("boom".to_string()), Some(5)) + .await + .unwrap(), + "retry's refusal under the lapsed claim is false" + ); + assert!( + !lapsed_fail.fail(None).await.unwrap(), + "deadline lapse refuses fail identically — the predicate is uniform" + ); + let retry_after = fast + .get_job(retry_refused) + .await + .unwrap() + .expect("the retried-at row survived"); + assert_eq!( + retry_after, retry_before, + "retry's refusal leaves the row untouched: the claim keeps pre-call state" + ); + + // ack_batch applies the predicate per id: the live id counts, the + // lapsed id and a never-claimed id silently do not. (Own queue — + // the lapsed legs above left reclaimable rows in `lapse`, and any + // later claim there would reclaim FIFO-first. The live id survives + // the 1 s stamp the same way the lapsed one lapsed: a renewal.) + let batch = store + .queue( + "batch", + alkstore::QueueOpts { + visibility_timeout_s: 1, + ..Default::default() + }, + ) + .await + .unwrap(); + let batch_live_id = batch + .enqueue(json!({"leg": "batch-live"}), EnqueueOpts::default()) + .await + .unwrap(); + let batch_lapse_id = batch + .enqueue(json!({"leg": "batch-lapsed"}), EnqueueOpts::default()) + .await + .unwrap(); + let batch_live = batch.claim_one("w3").await.unwrap().unwrap(); + let batch_lapsed = batch.claim_one("w3").await.unwrap().unwrap(); + assert_eq!( + (batch_live.job().id, batch_lapsed.job().id), + (batch_live_id, batch_lapse_id), + "claims hand out in enqueue order (equal priority)" + ); + assert!( + batch_live.heartbeat(600).await.unwrap(), + "the renewal keeps this id's claim live through the batch" + ); + past_stamp_sleep(2_500); + let counted = batch + .ack_batch(&[batch_live_id, batch_lapse_id, 987_654_321]) + .await + .unwrap(); + assert_eq!( + counted, 1, + "the returned count reflects only the live acks (lapsed and nonexistent ids not counted)" + ); + assert!( + batch.get_job(batch_live_id).await.unwrap().is_none(), + "the counted id acked (deleted)" + ); + let lapsed_alive = batch + .get_job(batch_lapse_id) + .await + .unwrap() + .expect("the lapsed id's row survives the batch"); + assert_eq!( + lapsed_alive.state, + JobState::Processing, + "the lapsed id was silently not counted, not deleted" + ); + + // The engine-default strings (fail(None); retry at exhausted + // budget) on a long-visibility queue — no lapse interference. + let failed = slow + .enqueue( + json!({"leg": "fail-default"}), + EnqueueOpts { + max_attempts: Some(1), + ..Default::default() + }, + ) + .await + .unwrap(); + let held_fail = slow.claim_one("w4").await.unwrap().unwrap(); + assert_eq!(held_fail.job().id, failed); + assert!(held_fail.fail(None).await.unwrap()); + let dead_failed = slow + .get_job(failed) + .await + .unwrap() + .expect("fail(None) dead-letters"); + assert_eq!(dead_failed.state, JobState::Dead); + assert_eq!(dead_failed.last_error.as_deref(), Some("failed")); + assert!(dead_failed.died_at.is_some()); + + let exhausting = slow + .enqueue( + json!({"leg": "exhaustion"}), + EnqueueOpts { + max_attempts: Some(1), + ..Default::default() + }, + ) + .await + .unwrap(); + let held_retry = slow.claim_one("w5").await.unwrap().unwrap(); + assert_eq!(held_retry.job().id, exhausting); + assert!(held_retry.retry(None, Some(5)).await.unwrap()); + let dead_exhausted = slow + .get_job(exhausting) + .await + .unwrap() + .expect("retry at exhausted budget dead-letters"); + assert_eq!(dead_exhausted.state, JobState::Dead); + assert_eq!( + dead_exhausted.last_error.as_deref(), + Some("max attempts exceeded"), + "the engine-default exhaustion string lands identically" + ); + assert!(dead_exhausted.died_at.is_some()); + + factory.teardown().await.expect("factory teardown"); +} + +/// **Queue-depth reclaim and dead-letter** — the ADR-010 depth +/// properties over a 1 s-visibility queue: a visibility reclaim +/// consumes an attempt (the expired claim reclaims to a second +/// claimant with `attempts` incremented and `claimed_at` refreshed — +/// the original holder's ops refuse after the reclaim), exhaustion +/// dead-letters into get_job-visible dead storage (`"max attempts +/// exceeded"` + `died_at` — the pre-claim sweep shape: a row whose +/// reclaim consumed its budget dead-letters at the next claim), and +/// `cancel` is an unconditional delete of either state — not an +/// interrupt: the row is gone (live and dead alike refuse `cancel` +/// with `false`), and the holder's next op refuses with the same +/// shape as a lapse. +/// +/// Contract stamp: ADR-010 §2 (a reclaim consumes an attempt; the +/// original holder refuses post-reclaim); ADR-010 §3 (exhaustion via +/// reclaim dead-letters); ADR-010 §1 (cancel — unconditional, not an +/// interrupt); ADR-010 §4 (dead-visible `get_job` with +/// `last_error`/`died_at`; the exhaustion string). +pub async fn queue_depth_reclaim_and_dead_letter(factory: &dyn StoreFactory) { + let store = factory.open().await.expect("factory opens a store"); + let depth = store + .queue( + "depth", + alkstore::QueueOpts { + visibility_timeout_s: 1, + max_attempts: 2, + ..Default::default() + }, + ) + .await + .unwrap(); + + // Visibility reclaim consumes an attempt. + let reclaimed = depth + .enqueue(json!({"leg": "reclaim"}), EnqueueOpts::default()) + .await + .unwrap(); + let first = depth.claim_one("w1").await.unwrap().unwrap(); + assert_eq!(first.job().id, reclaimed); + assert_eq!(first.job().attempts, 1); + let t1 = first.job().claimed_at.expect("claim stamps claimed_at"); + let e1 = first + .job() + .claim_expires_at + .expect("claim stamps the deadline"); + past_stamp_sleep(2_500); + + let second = depth + .claim_one("w2") + .await + .unwrap() + .expect("the lapsed claim reclaims"); + assert_eq!(second.job().id, reclaimed, "the same row reclaims"); + assert_eq!(second.job().attempts, 2, "a reclaim consumes an attempt"); + assert_eq!(second.job().state, JobState::Processing); + assert_eq!(second.job().worker_id.as_deref(), Some("w2")); + let t2 = second + .job() + .claimed_at + .expect("the reclaim re-stamps claimed_at"); + let e2 = second + .job() + .claim_expires_at + .expect("the reclaim re-stamps the deadline"); + assert!(t2 > t1, "claimed_at is refreshed by the reclaim"); + assert!(e2 > e1, "the deadline is refreshed, not inherited"); + assert!( + !first.heartbeat(600).await.unwrap(), + "the original holder refuses after the reclaim" + ); + assert!(second.ack().await.unwrap(), "the reclaiming holder acks"); + assert!(depth.get_job(reclaimed).await.unwrap().is_none()); + + // Exhaustion: the reclaim that spends the budget, then the + // next claim's pre-claim sweep dead-letters the row. + let exhausted = depth + .enqueue(json!({"leg": "exhausted"}), EnqueueOpts::default()) + .await + .unwrap(); + depth.claim_one("w1").await.unwrap().unwrap(); + past_stamp_sleep(2_500); + let at_budget = depth.claim_one("w2").await.unwrap().unwrap(); + assert_eq!(at_budget.job().id, exhausted); + assert_eq!( + at_budget.job().attempts, + at_budget.job().max_attempts, + "the reclaim spent the budget" + ); + past_stamp_sleep(2_500); + assert!( + depth.claim_one("w3").await.unwrap().is_none(), + "the budget-spent row dead-letters instead of reclaiming" + ); + let dead = depth + .get_job(exhausted) + .await + .unwrap() + .expect("the dead row is get_job-visible"); + assert_eq!( + dead.state, + JobState::Dead, + "exhaustion moves to dead storage" + ); + assert_eq!(dead.last_error.as_deref(), Some("max attempts exceeded")); + assert!(dead.died_at.is_some()); + + // Cancel is an unconditional delete — either state, any holder — + // and not an interrupt: the holder's next op refuses the same + // shape as a lapse, and nothing re-enters the claim path. + let cancelled_live = depth + .enqueue(json!({"leg": "cancel-live"}), EnqueueOpts::default()) + .await + .unwrap(); + let holder = depth.claim_one("w1").await.unwrap().unwrap(); + assert_eq!(holder.job().id, cancelled_live); + assert!(depth.cancel(cancelled_live).await.unwrap()); + assert!( + depth.get_job(cancelled_live).await.unwrap().is_none(), + "cancel deletes the row whole: no re-claim, no dead row" + ); + assert!( + !holder.heartbeat(600).await.unwrap(), + "not an interrupt — the holder's next op refuses identically" + ); + + let cancelled_pending = depth + .enqueue(json!({"leg": "cancel-pending"}), EnqueueOpts::default()) + .await + .unwrap(); + assert!(depth.cancel(cancelled_pending).await.unwrap()); + assert!( + depth.claim_one("w1").await.unwrap().is_none(), + "a cancelled pending row never becomes claimable" + ); + + assert!( + !depth.cancel(987_654_321).await.unwrap(), + "no missing live row cancels" + ); + assert!( + !depth.cancel(exhausted).await.unwrap(), + "a dead row is outside cancel's live-row reach" + ); + assert!( + depth.get_job(exhausted).await.unwrap().is_some(), + "cancel on a dead id deletes nothing" + ); + + factory.teardown().await.expect("factory teardown"); +} + +/// **Sweep: no stranded rows** — `sweep_expired()` moves *every* +/// past-expiry row (any state: the pending-expired and the +/// processing-expired alike — an enqueued job with an `expires` +/// deadline is eventually in exactly one of pending, processing, or +/// dead, never stuck unreachable), stamps the `"expired"` string, and +/// enforces dead-letter retention (rows past a set +/// `dead_letter_retention_s` delete; retention `None` = forever) — +/// the count being the moved + retention-deleted sum. Unexpired and +/// never-expiring rows stay untouched. +/// +/// The M-1 follow-up (the sweep's retention half rolling back with +/// the sweep on a retention-DELETE failure) is pinned engine-side — +/// the failure is storage-level, not injectable through the contract +/// surface; see the SQLite substrate's +/// `sweep_rolls_back_the_retention_half_with_the_move` and this +/// task's Notes for the disposition. +/// +/// Contract stamp: ADR-010 §5 (the no-stranded-rows property, both +/// states, and the retention enforcement); ADR-010 §4 (the +/// `"expired"` string; retention default = forever). +pub async fn sweep_no_stranded_rows(factory: &dyn StoreFactory) { + let store = factory.open().await.expect("factory opens a store"); + + let sweeping = store + .queue("sweepme", alkstore::QueueOpts::default()) + .await + .unwrap(); + // The processing-expired row is claimed first (it is the only row + // at claim time) with a 2 s expiry stamp; the pending-expired row + // enqueues after the claim so FIFO never hands it out. + let processing = sweeping + .enqueue( + json!({"leg": "expired-processing"}), + EnqueueOpts { + expires: Some(2), + ..Default::default() + }, + ) + .await + .unwrap(); + let held = sweeping.claim_one("w1").await.unwrap().unwrap(); + assert_eq!(held.job().id, processing, "the expiry-stamped row claims"); + let pending = sweeping + .enqueue( + json!({"leg": "expired-pending"}), + EnqueueOpts { + expires: Some(2), + ..Default::default() + }, + ) + .await + .unwrap(); + let unexpired = sweeping + .enqueue( + json!({"leg": "unexpired"}), + EnqueueOpts { + expires: Some(3600), + ..Default::default() + }, + ) + .await + .unwrap(); + let never = sweeping + .enqueue(json!({"leg": "never-expiring"}), EnqueueOpts::default()) + .await + .unwrap(); + past_stamp_sleep(3_500); + + let moved = sweeping.sweep_expired().await.unwrap(); + assert_eq!( + moved, 2, + "every past-expiry row moves — both states, exactly the expired ones" + ); + let dead_processing = sweeping + .get_job(processing) + .await + .unwrap() + .expect("the processing-expired row moved to dead"); + assert_eq!(dead_processing.state, JobState::Dead); + assert_eq!(dead_processing.last_error.as_deref(), Some("expired")); + assert!(dead_processing.died_at.is_some()); + let dead_pending = sweeping + .get_job(pending) + .await + .unwrap() + .expect("the pending-expired row moved to dead"); + assert_eq!(dead_pending.state, JobState::Dead); + assert_eq!(dead_pending.last_error.as_deref(), Some("expired")); + for (id, label) in [(unexpired, "unexpired"), (never, "never-expiring")] { + let live = sweeping.get_job(id).await.unwrap().unwrap(); + assert_eq!(live.state, JobState::Pending, "{label} row untouched"); + } + drop(held); + + // Retention enforcement: a dead row past its stamp deletes; the + // same-shaped row on a retention-None queue lives forever. + let retained = store + .queue( + "retained-1s", + alkstore::QueueOpts { + dead_letter_retention_s: Some(1), + ..Default::default() + }, + ) + .await + .unwrap(); + let keeping = store + .queue("keeper", alkstore::QueueOpts::default()) + .await + .unwrap(); + let expires_1s = retained + .enqueue(json!({"leg": "aged-out"}), EnqueueOpts::default()) + .await + .unwrap(); + let held_dies = retained.claim_one("w1").await.unwrap().unwrap(); + assert!(held_dies.fail(None).await.unwrap()); + let expires_never = keeping + .enqueue(json!({"leg": "kept"}), EnqueueOpts::default()) + .await + .unwrap(); + let kept_dies = keeping.claim_one("w1").await.unwrap().unwrap(); + assert!(kept_dies.fail(None).await.unwrap()); + past_stamp_sleep(2_500); + + let deleted = retained.sweep_expired().await.unwrap(); + assert_eq!( + deleted, 1, + "the retention deletion reports through the sweep count" + ); + assert!( + retained.get_job(expires_1s).await.unwrap().is_none(), + "the dead row past its retention stamp is gone" + ); + assert_eq!(keeping.sweep_expired().await.unwrap(), 0); + let kept = keeping.get_job(expires_never).await.unwrap().unwrap(); + assert_eq!( + kept.state, + JobState::Dead, + "retention None = dead rows live forever" + ); + + factory.teardown().await.expect("factory teardown"); +} + /// Shared clock helper for the suite's clock-adjacent rows: unix /// seconds now (rows never assert tight timings — the tolerance is /// the stamp's resolution). diff --git a/alkstore-postgres/tests/contract_suite.rs b/alkstore-postgres/tests/contract_suite.rs index f4b8e85..ec18d00 100644 --- a/alkstore-postgres/tests/contract_suite.rs +++ b/alkstore-postgres/tests/contract_suite.rs @@ -165,8 +165,9 @@ mod rows { use alkstore_contract_suite::properties::{ drop_rollback_leaves_no_ghosts, duration_refusal_on_non_positive_ttl, enqueue_opts_resolution, extent_clamp_semantics, in_tx_reads_see_own_writes, - name_validation_rejects_empty_and_reserved, payload_round_trip_stores_exact_encoding, - payload_too_large_produced_on_pg, receiver_close_and_save_arms, + job_handle_validity_predicate, name_validation_rejects_empty_and_reserved, + payload_round_trip_stores_exact_encoding, payload_too_large_produced_on_pg, + queue_depth_reclaim_and_dead_letter, receiver_close_and_save_arms, sweep_no_stranded_rows, }; use super::PgFactory; @@ -232,6 +233,21 @@ mod rows { receiver_close_and_save_arms, "row-receiver-arms" ); + harness_row!( + row_job_handle_validity_predicate, + job_handle_validity_predicate, + "row-handle-predicate" + ); + harness_row!( + row_queue_depth_reclaim_and_dead_letter, + queue_depth_reclaim_and_dead_letter, + "row-queue-depth" + ); + harness_row!( + row_sweep_no_stranded_rows, + sweep_no_stranded_rows, + "row-sweep-stranded" + ); } mod factory_shape { diff --git a/alkstore-sqlite/src/substrate/queue_ops.rs b/alkstore-sqlite/src/substrate/queue_ops.rs index aeffeb9..e249784 100644 --- a/alkstore-sqlite/src/substrate/queue_ops.rs +++ b/alkstore-sqlite/src/substrate/queue_ops.rs @@ -795,6 +795,14 @@ mod queue_tests { .unwrap(); } + fn block_dead_deletes(conn: &Connection) { + conn.execute_batch( + "CREATE TRIGGER test_block_dead_delete BEFORE DELETE ON __alkstore_dead + BEGIN SELECT RAISE(ABORT, 'dead delete blocked'); END", + ) + .unwrap(); + } + fn undecodable_pending_job(conn: &Connection) -> i64 { let id = put(conn, "emails", "{}", now_unix(conn).unwrap(), 0, 3, None); conn.execute( @@ -1532,6 +1540,69 @@ mod queue_tests { assert!(conn.is_autocommit()); } + // The M-1 follow-up (the waves-1–2 general review's §4 suggested add, + // task suite-queue-depth-rows): the retention half's rollback. The + // retention DELETE runs inside the same savepoint as the move half — + // a failure in it rolls the live→dead move back too (no effects + // partially applied by an Err sweep). Not injectable through the + // contract surface (storage-level failure, no fault seam) — pinned + // here at the substrate seam with a trigger-raised ABORT. + #[test] + fn sweep_rolls_back_the_retention_half_with_the_move() { + let conn = db(); + let now: i64 = now_unix(&conn).unwrap(); + let aged = enqueue( + &conn, + "q", + "{}", + now, + 0, + Some(now - 1), + Stamps { + max_attempts: 5, + visibility_timeout_s: 300, + backoff_base_s: 5, + dead_letter_retention_s: Some(60), + }, + ) + .unwrap(); + assert_eq!(sweep_expired(&conn, "q", EXPIRED).unwrap(), 1); + let aged_out = now - 61; + conn.execute_batch(&format!( + "UPDATE __alkstore_dead SET died_at = {aged_out} WHERE id = {aged}" + )) + .unwrap(); + let fresh = put(&conn, "q", "{}", now, 0, 5, Some(now - 1)); + + block_dead_deletes(&conn); + let err = sweep_expired(&conn, "q", EXPIRED) + .expect_err("the failing retention DELETE must reach the caller"); + assert!( + err.to_string().contains("dead delete blocked"), + "got: {err}" + ); + assert_eq!( + (live_count(&conn, fresh), dead_count(&conn, fresh)), + (1, 0), + "the move half rolled back with the retention half" + ); + assert_eq!( + dead_count(&conn, aged), + 1, + "the retention DELETE rolled back with it" + ); + assert!(conn.is_autocommit()); + + conn.execute_batch("DROP TRIGGER test_block_dead_delete") + .unwrap(); + assert_eq!( + sweep_expired(&conn, "q", EXPIRED).unwrap(), + 2, + "unblocked, both halves land: one moved + one retention-deleted" + ); + assert_eq!((live_count(&conn, fresh), dead_count(&conn, aged)), (0, 0)); + } + // ---------------- no-stranded-rows property ---------------- #[test] diff --git a/alkstore-sqlite/tests/contract_suite.rs b/alkstore-sqlite/tests/contract_suite.rs index 393e182..9d16ac5 100644 --- a/alkstore-sqlite/tests/contract_suite.rs +++ b/alkstore-sqlite/tests/contract_suite.rs @@ -69,8 +69,9 @@ mod rows { use alkstore_contract_suite::properties::{ drop_rollback_leaves_no_ghosts, duration_refusal_on_non_positive_ttl, enqueue_opts_resolution, extent_clamp_semantics, in_tx_reads_see_own_writes, - name_validation_rejects_empty_and_reserved, payload_round_trip_stores_exact_encoding, - payload_too_large_never_produced_on_sqlite, receiver_close_and_save_arms, + job_handle_validity_predicate, name_validation_rejects_empty_and_reserved, + payload_round_trip_stores_exact_encoding, payload_too_large_never_produced_on_sqlite, + queue_depth_reclaim_and_dead_letter, receiver_close_and_save_arms, sweep_no_stranded_rows, }; use super::SqliteFactory; @@ -123,6 +124,21 @@ mod rows { async fn row_receiver_close_and_save_arms() { receiver_close_and_save_arms(&factory("row-receiver-arms")).await; } + + #[tokio::test(flavor = "multi_thread")] + async fn row_job_handle_validity_predicate() { + job_handle_validity_predicate(&factory("row-handle-predicate")).await; + } + + #[tokio::test(flavor = "multi_thread")] + async fn row_queue_depth_reclaim_and_dead_letter() { + queue_depth_reclaim_and_dead_letter(&factory("row-queue-depth")).await; + } + + #[tokio::test(flavor = "multi_thread")] + async fn row_sweep_no_stranded_rows() { + sweep_no_stranded_rows(&factory("row-sweep-stranded")).await; + } } mod factory_shape { diff --git a/tasks/suite-queue-depth-rows.md b/tasks/suite-queue-depth-rows.md index 7ab8abf..9e9caa0 100644 --- a/tasks/suite-queue-depth-rows.md +++ b/tasks/suite-queue-depth-rows.md @@ -1,7 +1,7 @@ --- id: suite-queue-depth-rows name: Contract-suite rows — job-handle validity predicate + ADR-010 queue depth -status: pending +status: completed depends_on: [] scope: moderate risk: medium @@ -63,16 +63,16 @@ outcomes never timing values. No two-task races. ## Acceptance Criteria -- [ ] Three rows exist in `properties.rs`, `pub async fn(&dyn +- [x] Three rows exist in `properties.rs`, `pub async fn(&dyn StoreFactory)`, version-stamped (ADR-010 §2/§5, ADR-019 §3, ADR-008 §5 as applicable; stamps cite ADR § per the `version_stamp.rs` convention) -- [ ] Rows wired into both engines' `contract_suite.rs` targets -- [ ] SQLite column green server-less; pg column green against the +- [x] Rows wired into both engines' `contract_suite.rs` targets +- [x] SQLite column green server-less; pg column green against the harness server -- [ ] The M-1 retention-failure pin's location decided and recorded in +- [x] The M-1 retention-failure pin's location decided and recorded in Notes (suite row or engine-side, with the reason) -- [ ] `cargo test -p alkstore-sqlite -p alkstore-postgres` green +- [x] `cargo test -p alkstore-sqlite -p alkstore-postgres` green (pg rows skip cleanly server-less); clippy `-D warnings`; fmt clean ## References @@ -87,8 +87,93 @@ outcomes never timing values. No two-task races. ## Notes -> To be filled by implementation agent +> Decisions of record the description didn't pin: + +- **M-1 retention-failure disposition: pinned engine-side (SQLite + substrate), not as a suite row.** The retention-DELETE failure is a + storage-level fault with no injection seam through the contract + surface (`StoreFactory` + the contract traits offer no way to fail + one statement of one engine's sweep while the store stays live) — so + the pin landed as `sweep_rolls_back_the_retention_half_with_the_move` + in the SQLite substrate's queue test seam (a trigger-raised ABORT on + `__alkstore_dead` DELETE — the exact mechanism review 001 §4 + sketched): a sweep with both an expired live row and an aged-out + dead row fails entire (the move half's effects roll back with the + retention half; unblocked, the re-sweep accounts 1 moved + + 1 retention-deleted — the sum). The pg twin was **not** given a + live-injected leg: the engine has no client-side fault seam either, + and its sweep frame is structurally the same guarantee verified by + code-read — both halves (move + retention) run inside one `in_tx` + `BEGIN…COMMIT/ROLLBACK` frame (queue.rs `sweep_expired`), so a + retention-DELETE failure rolls the move back via the frame's error + arm. Review gate should audit this disposition: pg's leg rests on + frame structure, not a live fault test. +- **The exhaustion-string pin is split across the two rows by + trigger**: `fail(None)` → `"failed"` and retry-at-exhausted-budget → + `"max attempts exceeded"` pin in `job_handle_validity_predicate` + (both reachable without a lapse); *reclaim*-exhaustion → + `"max attempts exceeded"` pins in `queue_depth_reclaim_and_dead_letter` + where that trigger actually occurs (the pre-claim sweep at the + budget-spent claim). The row-1 text's single-line mention of + exhaustion thus rides the row that owns the exercising path. +- **The `ack_batch` leg carries its own queue and a renewal, not a + post-lapse fresh claim** — any later claim in a queue holding lapsed + rows reclaims them FIFO-first (the predicate makes a lapsed claim + reclaimable, so a "fresh" claim in the same queue picks the lapsed + row ahead of the fresh one; found live on the first run against the + real engine — the wave-3 lesson repeating). The leg therefore keeps + one id's claim live through a heartbeat renewal while its sibling + lapses on the 1 s stamp, in a dedicated short-visibility queue: three + ids (live / lapsed / nonexistent), count 1, the lapsed row still + `processing` afterward. +- **cancel legs extend the task's list slightly in the same row**: + cancel of a *dead* id returns `false` and deletes nothing (outside + cancel's live-row reach), alongside the pinned processing-cancel + (holder's next op refuses the lapse shape), pending-cancel + (never claimable), and missing-id `false` legs. +- **Timing posture realization**: sleeps are 2.5 s for 1 s + visibility/claim stamps and 3.5 s for `expires: Some(2)` stamps (the + extra second keeps the claim window — `expires_at > now` at claim — + deterministic against second-resolution clocks); every assertion is + a state outcome, none a duration. The deadline-boundary legs pin + refusal-state both sides of the lapse (in-window ops land; + post-lapse ops refuse) without asserting the boundary instant. +- **Row 2's reclaim leg also pins the original holder refusing after + the reclaim** (heartbeat → `false` with the row now `w2`'s) — the + reclaim side of the same predicate, in the row that owns reclaims. ## Summary -> To be filled on completion \ No newline at end of file +> Landed: the three queue-depth contract rows in +> `alkstore-contract-suite/src/properties.rs` — +> `job_handle_validity_predicate` (in-window heartbeat/ack landing; +> late-heartbeat, post-lapse ack, retry, and fail refusals past a 1 s +> stamp with the row untouched; `ack_batch` per-id predicate — live +> 1, lapsed 0, nonexistent 0; `fail(None)` → `"failed"` and +> retry-at-budget → `"max attempts exceeded"`), +> `queue_depth_reclaim_and_dead_letter` (reclaim consumes an attempt +> with `claimed_at`/deadline refreshed; original holder refuses; +> reclaim-exhaustion dead-letters with the pre-claim sweep — dead row +> `get_job`-visible with `"max attempts exceeded"` + `died_at`; +> cancel unconditional delete with the not-an-interrupt refusal shape, +> pending/dead/missing arms), and `sweep_no_stranded_rows` +> (pending-expired + processing-expired both move with `"expired"`, +> unexpired/never-expiring untouched; retention TTL deletes with the +> moved+deleted sum; retention `None` = forever). All version-stamped +> (ADR-010 §1–§5, ADR-019 §3, ADR-008 §5), re-exported from the suite +> lib, and wired into both engines' `contract_suite.rs` targets +> (SQLite three new `#[tokio::test]`s; pg three `harness_row!`s). +> +> M-1 follow-up: the sweep's retention-half rollback pinned in the +> SQLite substrate (`block_dead_deletes` trigger seam + +> `sweep_rolls_back_the_retention_half_with_the_move`) — see Notes for +> the disposition. +> +> Verified: workspace `cargo build` and `cargo test` green (25 core + +> 189 + 121 SQLite + 121 pg lib + 9 schema + harness crates); +> SQLite contract suite 13/13 (three new rows, 8.5 s total); pg +> contract suite 13/13 against the harness server (`pglo-poc` :15432, +> `postgres/poc`/`blobs`) twice consecutively (15.3 s each — the sleep +> legs really ran, not skipped); the substrate rollback tests 4/4; +> clippy `--workspace --all-targets -D warnings` clean; +> `cargo fmt --check` clean. \ No newline at end of file