Contract-suite tx commit-atomicity rows (task suite-tx-commit-atomicity-rows): the N-5 panic-probe admission in properties.rs's module doc (spawned with_tx task joined via JoinError::is_panic — catch_unwind around an async closure cannot see the panic point across an await; post-panic assertions read-only until convergence, single-task drive, no race window; ADR-017 §2 class 4) and three version-stamped rows: outbox_enqueue_tx_commit_atomicity (rollback drops the backing-queue job with the business write — get_job_tx-gone + run_once claims nothing; commit makes the job claimable exactly when the business write commits, real delivery through the RecordingDelivery closure with job-id identity, derived __alkstore_outbox:mail queue, exact payload, 60/5/5 stamps, consumed-after-ack; ADR-014 §1, ADR-010 §3a, ADR-021 §4, ADR-007), publish_with_key_tx_commit_atomicity (rollback drops the keyed event with the business write, key round-tripping inside the tx; commit surfaces it to read_since and a post-commit subscriber attach with the key round-tripping; ADR-015 §2/§4, ADR-021 §4, ADR-007), and with_tx_panicking_closure_rolls_back (N-5's probe against real engines: the panicking closure writes all four kinds then panics mid-flight; panic surfaces via the join; no-ghost reads converge with begin_tx granting every iteration; pre-panic listener silence on the notified channel; ghost never claimable; fresh with_tx commits through the same seam — both engines green; ADR-007, ADR-021 §4) — wired into both engines' suite targets (SQLite tokio tests, pg harness_row!s). Dispositions recorded in Notes: the 60/5/5 stamp inspection rides the delivery handle's job() because get_job cannot target the reserved derived backing queue through the contract surface (exemplar row pins the rejection; engine-side raw-row probes stayed put), the panic row's notify leg is a windowed silence (SQLite's wake overtriggers on any commit — the fully cross-engine notify-ghost pin rides the engines' rollback-ghosts twins, the suite's drop-rollback row made the same call), the closure tail routes through a #[cold] mid_flight_panic -> Error helper (a bare Err(panic!()) tail trips unreachable_code under -D warnings), and the post-panic convergence loop is the suite-side bounded wait (state outcomes only, begin_tx doubling as the seam-still-grants probe). Verified: sqlite suite 19/19, pg suite 19/19 vs harness twice (postgres/poc@:15432) + solo re-runs of each new row per engine (determinism), workspace build/test green, clippy -D warnings, fmt clean
This commit is contained in:
1 parent
7f749ac673
commit
360e71e51e
5 files changed
+567
-16
No files matched your search
@@ -40,8 +40,10 @@ 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, 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,
|
||||
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,
|
||||
queue_depth_reclaim_and_dead_letter, receiver_close_and_save_arms, scheduler_boundary_fires,
|
||||
scheduler_bounded_catchup, scheduler_leadership_discipline, sweep_no_stranded_rows,
|
||||
with_tx_panicking_closure_rolls_back,
|
||||
};
|
||||
@@ -24,6 +24,14 @@
|
||||
//! posture change (ADR-017 §2 class 4 — the suite tests the contract,
|
||||
//! it is not the contract).
|
||||
//!
|
||||
//! The N-5 panic probe is admitted by its own backlog row's mechanics
|
||||
//! for the same reason (the disposition demands a real engine's
|
||||
//! handle): a spawned `with_tx` task joined via
|
||||
//! `JoinError::is_panic()` — `catch_unwind` around an async closure
|
||||
//! cannot see the panic point across an await — with the post-panic
|
||||
//! assertions read-only until convergence, single-task driving, no
|
||||
//! two-task race window.
|
||||
//!
|
||||
//! Rows exercise a *close/dispose* arm by dropping the
|
||||
//! store handle (the dispose carrier every engine implements — the
|
||||
//! drop path closes receivers terminally); teardown-then-fresh-reopen
|
||||
@@ -35,8 +43,9 @@ use std::sync::Arc;
|
||||
use serde_json::json;
|
||||
|
||||
use alkstore::{
|
||||
BoxedFuture, Delivery, EnqueueOpts, Error, JobHandle, JobState, QueueOpts, RESERVED_PREFIX,
|
||||
Result, ScheduleOpts, StopToken, Store, StreamEvent, encode_payload, validate_shared_name,
|
||||
BoxedFuture, Delivery, EnqueueOpts, Error, Job, JobHandle, JobState, QueueOpts,
|
||||
RESERVED_PREFIX, Result, ScheduleOpts, StopToken, Store, StreamEvent, encode_payload,
|
||||
validate_shared_name,
|
||||
};
|
||||
|
||||
use crate::factory::StoreFactory;
|
||||
@@ -1936,3 +1945,435 @@ pub async fn scheduler_leadership_discipline(factory: &dyn StoreFactory) {
|
||||
|
||||
factory.teardown().await.expect("factory teardown");
|
||||
}
|
||||
|
||||
/// The delivery recorder for the tx-seam commit-atomicity rows: a
|
||||
/// real `Delivery` — every claimed job is captured (its `Job` snapshot,
|
||||
/// stamps included) and acked via the `Ok` disposition.
|
||||
struct RecordingDelivery {
|
||||
delivered: Vec<Job>,
|
||||
}
|
||||
|
||||
impl Delivery for RecordingDelivery {
|
||||
fn deliver<'a>(&'a mut self, job: Box<dyn JobHandle>) -> BoxedFuture<'a, Result<()>> {
|
||||
self.delivered.push(job.job().clone());
|
||||
Box::pin(std::future::ready(Ok(())))
|
||||
}
|
||||
}
|
||||
|
||||
/// The panic probe's typed diverger: a `#[cold]` call typed `Error` so
|
||||
/// the closure's tail stays a normal expression (a bare `Err(panic!())`
|
||||
/// tail is unreachable-call-shaped and trips `unreachable_code`).
|
||||
#[cold]
|
||||
fn mid_flight_panic(message: &str) -> Error {
|
||||
panic!("{message}")
|
||||
}
|
||||
|
||||
/// **`outbox_enqueue_tx` commit-atomicity** — the outbox send sits in
|
||||
/// the caller's transaction: rollback drops the backing-queue job row
|
||||
/// together with the business write (no ghost job — `run_once`
|
||||
/// afterwards claims nothing), and commit makes the job claimable by
|
||||
/// `run_once` exactly when the business write commits (a real delivery
|
||||
/// driven through the `Delivery` closure, the job consumed after its
|
||||
/// ack). The stamped opts are the outbox's derived 60/5/5 set,
|
||||
/// delivered on the job handle and identical on both engines — the
|
||||
/// derived backing queue is engine-internal (reserved-prefixed,
|
||||
/// unreachable through the contract surface: the exemplar row pins
|
||||
/// `get_job_tx`'s reserved rejection), so the delivery handle's
|
||||
/// `job()` is the contract-visible inspection of the stamps.
|
||||
///
|
||||
/// Contract stamp: ADR-014 §1 (`outbox_enqueue_tx` — commit makes the
|
||||
/// backing-queue job claimable exactly when the business write
|
||||
/// commits; rollback drops both; the derived-name-only path);
|
||||
/// ADR-010 §3a (the stamp point — `EnqueueOpts` resolved over the
|
||||
/// outbox's derived 60/5/5 set, visible on the delivered job);
|
||||
/// ADR-021 §4 (drop = rollback — the no-ghosts property the rollback
|
||||
/// leg pins); ADR-007 (the caller-held seam the pair commits on).
|
||||
pub async fn outbox_enqueue_tx_commit_atomicity(factory: &dyn StoreFactory) {
|
||||
let store = factory.open().await.expect("factory opens a store");
|
||||
|
||||
// Rollback leg: the backing-queue job row drops together with the
|
||||
// business write — nothing for `run_once` to claim afterwards.
|
||||
let mut tx = store.begin_tx().await.unwrap();
|
||||
let business = tx
|
||||
.enqueue_tx("ledger", EnqueueOpts::default(), json!({"leg": "rollback"}))
|
||||
.await
|
||||
.unwrap();
|
||||
tx.outbox_enqueue_tx("mail", EnqueueOpts::default(), json!({"send": "ghost"}))
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(
|
||||
tx.get_job_tx("ledger", business).await.unwrap().is_some(),
|
||||
"read-your-own-writes makes the rollback's drop meaningful"
|
||||
);
|
||||
drop(tx);
|
||||
|
||||
let mut tx = store.begin_tx().await.unwrap();
|
||||
let business_gone = tx.get_job_tx("ledger", business).await.unwrap();
|
||||
tx.commit().await.unwrap();
|
||||
assert_eq!(
|
||||
business_gone, None,
|
||||
"the rollback dropped the business write"
|
||||
);
|
||||
|
||||
let mut mail = store.outbox("mail").await.unwrap();
|
||||
let claimed = mail
|
||||
.run_once(
|
||||
"suite-probe",
|
||||
&mut RecordingDelivery {
|
||||
delivered: Vec::new(),
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(
|
||||
!claimed,
|
||||
"rollback leaves no ghost job for run_once to claim"
|
||||
);
|
||||
|
||||
// Commit leg: commit lands both halves together — the business
|
||||
// write becomes `get_job`-visible and the backing-queue job becomes
|
||||
// claimable by `run_once`, which delivers through the real
|
||||
// `Delivery` closure.
|
||||
let mut tx = store.begin_tx().await.unwrap();
|
||||
let business = tx
|
||||
.enqueue_tx("ledger", EnqueueOpts::default(), json!({"leg": "commit"}))
|
||||
.await
|
||||
.unwrap();
|
||||
let queued = tx
|
||||
.outbox_enqueue_tx("mail", EnqueueOpts::default(), json!({"send": "letter"}))
|
||||
.await
|
||||
.unwrap();
|
||||
tx.commit().await.unwrap();
|
||||
|
||||
let ledger = store.queue("ledger", QueueOpts::default()).await.unwrap();
|
||||
let business_row = ledger
|
||||
.get_job(business)
|
||||
.await
|
||||
.unwrap()
|
||||
.expect("the business write committed with the enqueue");
|
||||
assert_eq!(business_row.queue, "ledger");
|
||||
|
||||
let mut delivery = RecordingDelivery {
|
||||
delivered: Vec::new(),
|
||||
};
|
||||
let ran = mail
|
||||
.run_once("suite-probe", &mut delivery)
|
||||
.await
|
||||
.expect("run_once claims the committed job");
|
||||
assert!(
|
||||
ran,
|
||||
"commit makes the backing-queue job claimable by run_once"
|
||||
);
|
||||
assert_eq!(
|
||||
delivery.delivered.len(),
|
||||
1,
|
||||
"the delivery closure ran exactly once"
|
||||
);
|
||||
let job = &delivery.delivered[0];
|
||||
assert_eq!(job.id, queued, "the delivered job is the tx-enqueued one");
|
||||
assert_eq!(
|
||||
job.queue,
|
||||
format!("{RESERVED_PREFIX}outbox:mail"),
|
||||
"the enqueue landed in the engine-derived backing queue"
|
||||
);
|
||||
assert_eq!(
|
||||
job.payload,
|
||||
encode_payload(&json!({"send": "letter"})).unwrap(),
|
||||
"the payload is exact"
|
||||
);
|
||||
assert_eq!(
|
||||
(
|
||||
job.visibility_timeout_s,
|
||||
job.max_attempts,
|
||||
job.backoff_base_s,
|
||||
job.dead_letter_retention_s
|
||||
),
|
||||
(60, 5, 5, None),
|
||||
"the outbox's derived 60/5/5 stamp set rides the delivered job"
|
||||
);
|
||||
let again = mail
|
||||
.run_once("suite-probe", &mut delivery)
|
||||
.await
|
||||
.expect("run_once after the delivery");
|
||||
assert!(!again, "the delivered-then-acked job is consumed");
|
||||
|
||||
factory.teardown().await.expect("factory teardown");
|
||||
}
|
||||
|
||||
/// **`publish_with_key_tx` commit-atomicity** — the keyed publish
|
||||
/// rides the tx seam: rollback drops the keyed event row with the
|
||||
/// business write (no ghost event), and commit makes it visible to
|
||||
/// `read_since` and to a subscriber attach, the key round-tripping on
|
||||
/// the committed event. (The empty-`Some`-key `InvalidName` leg pins
|
||||
/// in the exemplar row — not duplicated here.)
|
||||
///
|
||||
/// Contract stamp: ADR-015 §2 (`publish_with_key_tx` — the tx seam
|
||||
/// carries the key; commit-atomic with the business write); ADR-015
|
||||
/// §4 (the committed event's read surfaces: offset ASC, the key as
|
||||
/// carried metadata); ADR-021 §4 (drop = rollback); ADR-007 (the
|
||||
/// caller-held seam).
|
||||
pub async fn publish_with_key_tx_commit_atomicity(factory: &dyn StoreFactory) {
|
||||
let store = factory.open().await.expect("factory opens a store");
|
||||
|
||||
// Rollback leg: the keyed event row drops together with the
|
||||
// business write — no ghost event; the key still round-tripped
|
||||
// inside the tx (read-your-own-writes makes the drop meaningful).
|
||||
let mut tx = store.begin_tx().await.unwrap();
|
||||
tx.publish_tx("ledger-events", json!({"leg": "rollback", "n": 1}))
|
||||
.await
|
||||
.unwrap();
|
||||
let keyed = tx
|
||||
.publish_with_key_tx(
|
||||
"events",
|
||||
Some("k".to_string()),
|
||||
json!({"leg": "rollback", "n": 1}),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let seen = tx.read_since_tx("events", 0, 100).await.unwrap();
|
||||
assert_eq!(seen.len(), 1, "the tx sees its own keyed publish pre-drop");
|
||||
assert_eq!(seen[0].offset, keyed);
|
||||
assert_eq!(
|
||||
seen[0].key.as_deref(),
|
||||
Some("k"),
|
||||
"the key round-trips inside the tx"
|
||||
);
|
||||
drop(tx);
|
||||
|
||||
let ledger_gone = store
|
||||
.stream("ledger-events")
|
||||
.await
|
||||
.unwrap()
|
||||
.read_since(0, 100)
|
||||
.await
|
||||
.unwrap();
|
||||
let events_gone = store
|
||||
.stream("events")
|
||||
.await
|
||||
.unwrap()
|
||||
.read_since(0, 100)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(
|
||||
ledger_gone.is_empty(),
|
||||
"the rollback dropped the business write"
|
||||
);
|
||||
assert!(
|
||||
events_gone.is_empty(),
|
||||
"no ghost keyed event after the rollback"
|
||||
);
|
||||
|
||||
// Commit leg: the keyed event lands with the business write —
|
||||
// visible to fresh reads with the key round-tripping, and to a
|
||||
// subscriber attaching after the commit.
|
||||
let mut tx = store.begin_tx().await.unwrap();
|
||||
let business = tx
|
||||
.publish_tx("ledger-events", json!({"leg": "commit", "n": 2}))
|
||||
.await
|
||||
.unwrap();
|
||||
let keyed = tx
|
||||
.publish_with_key_tx(
|
||||
"streamed",
|
||||
Some("k2".to_string()),
|
||||
json!({"leg": "commit", "n": 2}),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
tx.commit().await.unwrap();
|
||||
|
||||
let ledger = store.stream("ledger-events").await.unwrap();
|
||||
let keyed_stream = store.stream("streamed").await.unwrap();
|
||||
let ledger_page = ledger.read_since(0, 100).await.unwrap();
|
||||
assert_eq!(ledger_page.len(), 1, "the business write committed");
|
||||
assert_eq!(ledger_page[0].offset, business);
|
||||
let keyed_page = keyed_stream.read_since(0, 100).await.unwrap();
|
||||
assert_eq!(keyed_page.len(), 1, "the keyed event committed with it");
|
||||
assert_eq!(keyed_page[0].offset, keyed);
|
||||
assert_eq!(
|
||||
keyed_page[0].key.as_deref(),
|
||||
Some("k2"),
|
||||
"the key round-trips on the committed event"
|
||||
);
|
||||
assert_eq!(
|
||||
keyed_page[0].payload,
|
||||
encode_payload(&json!({"leg": "commit", "n": 2})).unwrap(),
|
||||
"the payload is exact"
|
||||
);
|
||||
|
||||
let mut rx = keyed_stream.subscribe("attach").await.unwrap();
|
||||
let yielded = rx
|
||||
.recv()
|
||||
.await
|
||||
.expect("the subscriber attaches and yields")
|
||||
.expect("the attach read yields the committed event");
|
||||
assert_eq!(
|
||||
yielded.offset, keyed,
|
||||
"the subscriber attach sees the committed keyed event"
|
||||
);
|
||||
assert_eq!(
|
||||
yielded.key.as_deref(),
|
||||
Some("k2"),
|
||||
"the key round-trips on the attach"
|
||||
);
|
||||
|
||||
factory.teardown().await.expect("factory teardown");
|
||||
}
|
||||
|
||||
/// **`with_tx` panicking closure rolls back** — a closure that panics
|
||||
/// mid-flight ⇒ rollback with no residue over the uniform no-ghosts
|
||||
/// list (job/event/notify/offset — ADR-021 §4's drop = rollback
|
||||
/// through the panic path), and the store remains usable afterward
|
||||
/// (the SQLite writer slot replenished — the wave-3 stranding fix's
|
||||
/// property, now pinned at contract level; the pg pooled object
|
||||
/// re-pooled). The panic crosses an await inside `with_tx`'s own
|
||||
/// future — a `catch_unwind` around an async closure cannot see the
|
||||
/// panic point — so the row drives a spawned task and asserts the
|
||||
/// panic surfaced through the `JoinError` (`is_panic`), then the
|
||||
/// post-panic state: the no-ghost reads converge (the rollback may
|
||||
/// land asynchronously engine-side — the loop is the bounded wait,
|
||||
/// state-outcome assertions only), the pre-panic listener hears
|
||||
/// nothing on the notified channel before any post-panic commit exists
|
||||
/// (the notify residue's cross-engine pin: SQLite's wake overtriggers
|
||||
/// on any commit past this window, so the window precedes the
|
||||
/// happy-path write; the fully-cross-engine notify-ghost pin rides
|
||||
/// the engines' own rollback-ghost tests), and a fresh `with_tx`
|
||||
/// commits through the same seam.
|
||||
///
|
||||
/// Contract stamp: ADR-007 (`with_tx` — the closure-scoped transaction
|
||||
/// and its panic disposition: `Ok` ⇒ commit, `Err`/drop/panic ⇒
|
||||
/// rollback); ADR-021 §4 (drop = rollback through the panic path — no
|
||||
/// residue, the store usable afterward; N-5's probe against real
|
||||
/// engines).
|
||||
pub async fn with_tx_panicking_closure_rolls_back(factory: &dyn StoreFactory) {
|
||||
let store: Arc<dyn Store> = Arc::from(factory.open().await.expect("factory opens a store"));
|
||||
|
||||
// The listener attaches before the panicking tx fires: a ghost
|
||||
// notify delivering at commit would reach it.
|
||||
let mut listener = store.listen("panic-chan").await.expect("listen attaches");
|
||||
|
||||
let ghost_ids: Arc<std::sync::Mutex<Vec<i64>>> = Arc::new(std::sync::Mutex::new(Vec::new()));
|
||||
let captured = ghost_ids.clone();
|
||||
let joined = tokio::spawn({
|
||||
let store = store.clone();
|
||||
async move {
|
||||
store
|
||||
.with_tx(Box::new(move |tx| {
|
||||
Box::pin(async move {
|
||||
let id = tx
|
||||
.enqueue_tx("panicked", EnqueueOpts::default(), json!({"panic": true}))
|
||||
.await
|
||||
.expect("the in-flight enqueue lands inside the tx");
|
||||
captured.lock().expect("capture mutex").push(id);
|
||||
tx.publish_tx("panicked-events", json!({"panic": true}))
|
||||
.await
|
||||
.expect("the in-flight publish lands inside the tx");
|
||||
tx.notify_tx("panic-chan", json!({"panic": true}))
|
||||
.await
|
||||
.expect("the in-flight notify lands inside the tx");
|
||||
tx.save_offset_tx("panicked-events", "ghostc", 7)
|
||||
.await
|
||||
.expect("the in-flight offset save lands inside the tx");
|
||||
Err::<(), Error>(mid_flight_panic(
|
||||
"mid-flight panic across with_tx's closure",
|
||||
))
|
||||
}) as BoxedFuture<'_, Result<()>>
|
||||
}))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
})
|
||||
.await;
|
||||
|
||||
assert!(
|
||||
joined.is_err(),
|
||||
"the panic must surface through the spawned task's join, not be swallowed"
|
||||
);
|
||||
let join_error = joined.unwrap_err();
|
||||
assert!(
|
||||
join_error.is_panic(),
|
||||
"the join must carry a panic, not a cancellation: {join_error}"
|
||||
);
|
||||
|
||||
// No residue: the no-ghost reads converge — the rollback may land
|
||||
// asynchronously (pg's detached rollback; SQLite's runs in the
|
||||
// unwind), so the loop is the bounded wait for a *state* outcome;
|
||||
// each iteration's begin_tx is also the seam-still-grants probe.
|
||||
let ghost_id = *ghost_ids
|
||||
.lock()
|
||||
.expect("capture mutex")
|
||||
.first()
|
||||
.expect("the enqueue's id was captured before the panic");
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15);
|
||||
loop {
|
||||
let granted = tokio::time::timeout(std::time::Duration::from_secs(5), store.begin_tx())
|
||||
.await
|
||||
.expect("begin_tx must not hang: the seam must replenish after the panic");
|
||||
let mut tx = granted.expect("begin_tx succeeds post-panic");
|
||||
let ghost_job = tx.get_job_tx("panicked", ghost_id).await.unwrap();
|
||||
let ghost_events = tx.read_since_tx("panicked-events", 0, 100).await.unwrap();
|
||||
let ghost_checkpoint = tx.get_offset_tx("panicked-events", "ghostc").await.unwrap();
|
||||
tx.commit().await.unwrap();
|
||||
if ghost_job.is_none() && ghost_events.is_empty() && ghost_checkpoint == 0 {
|
||||
break;
|
||||
}
|
||||
assert!(
|
||||
std::time::Instant::now() < deadline,
|
||||
"the panic's rollback never converged: job {ghost_job:?}, events {}, \
|
||||
checkpoint {ghost_checkpoint}",
|
||||
ghost_events.len()
|
||||
);
|
||||
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
||||
}
|
||||
|
||||
// No ghost notify: the pre-panic listener hears nothing on the
|
||||
// notified channel — this window precedes any post-panic commit,
|
||||
// so an engine may not have a wake source here (SQLite's watcher
|
||||
// overtriggers on any commit; a past-window wake is not this leg's
|
||||
// business). A delivery-shaped wake is a genuine defect.
|
||||
let budget = std::time::Instant::now() + std::time::Duration::from_secs(1);
|
||||
let heard = loop {
|
||||
let wake = listener
|
||||
.recv_timeout(std::time::Duration::from_millis(150))
|
||||
.await
|
||||
.expect("the silent wait is idle, never an error");
|
||||
match wake {
|
||||
None => break None,
|
||||
Some(w) if w.channel == "panic-chan" => break Some(w),
|
||||
Some(_) => {}
|
||||
}
|
||||
assert!(
|
||||
std::time::Instant::now() < budget,
|
||||
"the silence window overflowed with unrelated wakes"
|
||||
);
|
||||
};
|
||||
assert!(
|
||||
heard.is_none(),
|
||||
"no ghost notify reached the listener through the panic path: {heard:?}"
|
||||
);
|
||||
|
||||
// The store remains usable (no writer-slot stranding, no
|
||||
// pooled-object leak): the ghost never became claimable work, and
|
||||
// a fresh closure-scoped tx commits through the same seam.
|
||||
let panicked = store.queue("panicked", QueueOpts::default()).await.unwrap();
|
||||
let drained = panicked.claim_batch("probe", 10).await.unwrap();
|
||||
assert!(drained.is_empty(), "no ghost work became claimable");
|
||||
store
|
||||
.with_tx(Box::new(|tx| {
|
||||
Box::pin(async move {
|
||||
tx.enqueue_tx("after", EnqueueOpts::default(), json!({"after": true}))
|
||||
.await?;
|
||||
Ok(())
|
||||
}) as BoxedFuture<'_, Result<()>>
|
||||
}))
|
||||
.await
|
||||
.expect("the store survives the panic: a fresh with_tx commits");
|
||||
let after = store.queue("after", QueueOpts::default()).await.unwrap();
|
||||
let claimed = after.claim_batch("probe", 10).await.unwrap();
|
||||
assert_eq!(
|
||||
claimed.len(),
|
||||
1,
|
||||
"the post-panic commit's business write is live work"
|
||||
);
|
||||
|
||||
factory.teardown().await.expect("factory teardown");
|
||||
}
|
||||
@@ -166,10 +166,11 @@ mod rows {
|
||||
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, name_validation_rejects_empty_and_reserved,
|
||||
payload_round_trip_stores_exact_encoding, payload_too_large_produced_on_pg,
|
||||
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,
|
||||
sweep_no_stranded_rows,
|
||||
sweep_no_stranded_rows, with_tx_panicking_closure_rolls_back,
|
||||
};
|
||||
|
||||
use super::PgFactory;
|
||||
@@ -265,6 +266,21 @@ mod rows {
|
||||
scheduler_leadership_discipline,
|
||||
"row-scheduler-leadership"
|
||||
);
|
||||
harness_row!(
|
||||
row_outbox_enqueue_tx_commit_atomicity,
|
||||
outbox_enqueue_tx_commit_atomicity,
|
||||
"row-outbox-tx-atomicity"
|
||||
);
|
||||
harness_row!(
|
||||
row_publish_with_key_tx_commit_atomicity,
|
||||
publish_with_key_tx_commit_atomicity,
|
||||
"row-keyed-tx-atomicity"
|
||||
);
|
||||
harness_row!(
|
||||
row_with_tx_panicking_closure_rolls_back,
|
||||
with_tx_panicking_closure_rolls_back,
|
||||
"row-panic-rollback"
|
||||
);
|
||||
}
|
||||
|
||||
mod factory_shape {
|
||||
|
||||
@@ -70,10 +70,11 @@ mod rows {
|
||||
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, name_validation_rejects_empty_and_reserved,
|
||||
payload_round_trip_stores_exact_encoding, payload_too_large_never_produced_on_sqlite,
|
||||
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,
|
||||
sweep_no_stranded_rows,
|
||||
sweep_no_stranded_rows, with_tx_panicking_closure_rolls_back,
|
||||
};
|
||||
|
||||
use super::SqliteFactory;
|
||||
@@ -156,6 +157,21 @@ mod rows {
|
||||
async fn row_scheduler_leadership_discipline() {
|
||||
scheduler_leadership_discipline(&factory("row-scheduler-leadership")).await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn row_outbox_enqueue_tx_commit_atomicity() {
|
||||
outbox_enqueue_tx_commit_atomicity(&factory("row-outbox-tx-atomicity")).await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn row_publish_with_key_tx_commit_atomicity() {
|
||||
publish_with_key_tx_commit_atomicity(&factory("row-keyed-tx-atomicity")).await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn row_with_tx_panicking_closure_rolls_back() {
|
||||
with_tx_panicking_closure_rolls_back(&factory("row-panic-rollback")).await;
|
||||
}
|
||||
}
|
||||
|
||||
mod factory_shape {
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
---
|
||||
id: suite-tx-commit-atomicity-rows
|
||||
name: Contract-suite rows — outbox/keyed-publish tx commit-atomicity + with_tx panic probe
|
||||
status: pending
|
||||
status: completed
|
||||
depends_on: []
|
||||
scope: moderate
|
||||
risk: low
|
||||
@@ -53,15 +53,15 @@ Rows to add:
|
||||
|
||||
## Acceptance Criteria
|
||||
|
||||
- [ ] Three rows exist, version-stamped (ADR-014 §1, ADR-015 §2,
|
||||
- [x] Three rows exist, version-stamped (ADR-014 §1, ADR-015 §2,
|
||||
ADR-021 §4, ADR-007 as applicable)
|
||||
- [ ] 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 panic probe passes on both engines (no writer-slot stranding,
|
||||
- [x] The panic probe passes on both engines (no writer-slot stranding,
|
||||
no pooled-object leak) — if an engine fails here, that is a real
|
||||
defect: stop, record, fix engine-side before completing the task
|
||||
- [ ] `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
|
||||
@@ -76,8 +76,84 @@ Rows to add:
|
||||
|
||||
## Notes
|
||||
|
||||
> To be filled by implementation agent
|
||||
Decisions of record beyond the task description:
|
||||
|
||||
- **The 60/5/5 stamp inspection rides the delivery handle's `job()`.**
|
||||
ADR-014 §1's backlog wording says the stamped opts are
|
||||
"`get_job`-visible", but the derived backing queue is engine-internal:
|
||||
it is reserved-prefixed, so no `get_job`/`get_job_tx` call can target
|
||||
it through the contract surface (the exemplar row itself pins
|
||||
`get_job_tx(reserved queue)` → `ReservedName`; the engine-side stamp
|
||||
tests use raw row probes, which the suite may not). The row therefore
|
||||
asserts the 60/5/5 set on the `Delivery` closure's `JobHandle`
|
||||
(`job()` — `visibility_timeout_s`/`max_attempts`/`backoff_base_s`/
|
||||
`dead_letter_retention_s`) — the only contract-visible inspection of
|
||||
backing-queue rows. The delivery payload-exactness and the
|
||||
`__alkstore_outbox:{name}` derived-name pin ride the same snapshot.
|
||||
- **The panic row's notify leg is a windowed silence, not a full
|
||||
cross-engine pin.** The row drives all four write kinds
|
||||
(job/event/notify/offset) through the panicking tx, but a rolled-back
|
||||
notify's no-residue is only observable as listener silence, and
|
||||
SQLite's wake overtriggers on *any* commit (the notify module's
|
||||
documented coalescing posture), so silence is asserted in a window
|
||||
that precedes every post-panic committed write — a delivery-shaped
|
||||
wake there is a genuine defect. The fully cross-engine notify-ghost
|
||||
pin rides the engines' own rollback-ghosts tests (SQLite's raw
|
||||
notify-count probe, pg's listener-silence twin); the suite's existing
|
||||
`drop_rollback_leaves_no_ghosts` row makes the same shape of call
|
||||
(drives the observable three kinds).
|
||||
- **The panic-probe closure's tail routes through a `#[cold]`
|
||||
helper typed `Error`** (`mid_flight_panic`) — a bare
|
||||
`Err(panic!(...))` tail is unreachable-call-shaped and trips the
|
||||
`unreachable_code` lint under `-D warnings`; `catch_unwind` around an
|
||||
async closure cannot see the panic point anyway (the mechanics hint),
|
||||
so the typed diverger costs nothing.
|
||||
- **The post-panic convergence loop is the suite-side bounded wait**
|
||||
for the rollback to land (pg's detached rollback task; SQLite's rolls
|
||||
back synchronously in the unwind): state-outcome assertions only,
|
||||
each iteration's `begin_tx` doubling as the seam-still-grants probe
|
||||
(the writer-slot/pool replenish property), the deadline a safety
|
||||
bound like `await_fires`' — no timing-value assertion.
|
||||
|
||||
## Summary
|
||||
|
||||
> To be filled on completion
|
||||
Landed the three tx-seam commit-atomicity rows in
|
||||
`alkstore-contract-suite/src/properties.rs`, each version-stamped per
|
||||
the suite convention:
|
||||
|
||||
- `outbox_enqueue_tx_commit_atomicity` — rollback drops the
|
||||
backing-queue job with the business write (business id `get_job_tx`-
|
||||
gone in a fresh tx; `run_once` afterwards claims nothing); commit
|
||||
makes the job claimable by `run_once` exactly when the business
|
||||
write commits (real delivery through the `RecordingDelivery`
|
||||
closure: job id identity, derived `__alkstore_outbox:mail` queue,
|
||||
exact payload, 60/5/5 stamps, consumed-after-ack) (ADR-014 §1,
|
||||
ADR-010 §3a, ADR-021 §4, ADR-007).
|
||||
- `publish_with_key_tx_commit_atomicity` — rollback drops the keyed
|
||||
event with the business write (both streams empty on fresh reads,
|
||||
key round-tripping inside the tx pre-drop); commit makes it visible
|
||||
to `read_since` and to a post-commit subscriber attach with the key
|
||||
round-tripping on both (ADR-015 §2/§4, ADR-021 §4, ADR-007).
|
||||
- `with_tx_panicking_closure_rolls_back` — N-5's probe against real
|
||||
engines for the first time: spawned-task `with_tx` closure writes
|
||||
enqueue/publish/notify/offset then panics mid-flight; the panic
|
||||
surfaces via `JoinError::is_panic()`; the no-ghost reads converge
|
||||
with `begin_tx` granting every iteration; the pre-panic listener
|
||||
hears nothing on the notified channel; the ghost never becomes
|
||||
claimable work; a fresh `with_tx` commits through the same seam —
|
||||
both engines green (no writer-slot stranding, no pooled-object
|
||||
leak) (ADR-007, ADR-021 §4).
|
||||
|
||||
Also: the module-doc determinism/orchestration note gained the panic
|
||||
probe's admission sentence (spawned-task mechanics, `catch_unwind`
|
||||
asymmetry — the runner-driving extension's sibling; ADR-017 §2 class
|
||||
4), `RecordingDelivery` + the `#[cold]` diverger are suite-local
|
||||
helpers, and the rows are exported from the suite lib and wired into
|
||||
`alkstore-sqlite/tests/contract_suite.rs` (three direct tokio tests)
|
||||
and `alkstore-postgres/tests/contract_suite.rs` (three
|
||||
`harness_row!`s).
|
||||
|
||||
Verified: sqlite suite 19/19; pg suite 19/19 vs the harness server
|
||||
(`pglo-poc` :15432, `postgres/poc`/`blobs`) twice consecutively, plus
|
||||
solo re-runs of each new row per engine (determinism); workspace
|
||||
build/test green; clippy `-D warnings` clean; fmt clean.
|
||||
Reference in new issue
Block a user