Contract-suite rows (task suite-opts-backoff-rows): the backoff-curve equivalence row and the deferred enqueue-opts clock legs. backoff_curve_equivalence — the equal-jitter exponential pinned as the range, not a jitter label (ADR-010 §3): claim → retry(err, None) cycles on a backoff_base_s = 2 queue land the ranges [1,2], [2,4], [4,8] across attempts 1–3, each computed delay read back through get_job's resolved run_at minus a clock read taken before the retry — the lower bound race-free (the retry's internal clock read cannot precede the suite's, so the observed value over-reads the delay, never under-reads) and the upper bound carrying a one-second straddle tolerance for the integer-second stamp crossing a wall second; identical bounds on both engines is the equivalence pin (ADR-012 §2, the row body shared). The explicit-delay override legs ride the same row: retry(err, Some(5)) honored verbatim ([5, 6] under the same straddle tolerance) and retry(err, Some(0)) resolving ready-now per the boundary-total rule (ADR-023 §2), claimed within a bounded wait. enqueue_opts_resolution extended in place with the wave-5 legs its deferral note named — the run_at-alone literal leg (a future run_at is the row's ready time verbatim, a deterministic equality with no clock read involved; a past run_at stores the literal too and is claimable now through the run_at <= now predicate within a bounded wait, the run_at-ASC ordering making the observation unambiguous among the row's other legs) and the neither-field leg (ready at the enqueue instant, abs_diff(now) <= 5 tolerance-bounded proximity, never a tight timing assert); the deferral note replaced by the completion statement, ADR-023 §2 stamp accumulated per the convention (ADR-020 §1 governs the resolutions). Cap-leg disposition recorded in Notes for the review gate: the 1-hour cap is not suite-pinnable (capped attempts need minute-scale waits) and stays pinned engine-side on both engines' unit tests per the wave-3/4 reviews. Ready-now waits are bounded claim_one poll loops (deadline asserts only, sequential single-store drive), with the one draft defect the finding surfaced (a re-claim after the bounded wait consumed nothing; the helper returns the claimed handle) noted. Wired into both engines' suite 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 harness (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 skips cleanly via the reachability gate, cargo test -p alkstore-sqlite -p alkstore-postgres green, clippy -D warnings, fmt clean
This commit is contained in:
1 parent
2a2ad7e11f
commit
36023914b2
5 files changed
+375
-31
No files matched your search
@@ -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<dyn alkstore::JobHandle> {
|
||||
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");
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user