From 7f749ac6736f13f49ba3bd54eca1dbec84eeeccb Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sat, 10 Oct 2026 06:04:30 +0000 Subject: [PATCH] =?UTF-8?q?Contract-suite=20scheduler=20rows=20(task=20sui?= =?UTF-8?q?te-scheduler-rows):=20the=20determinism-posture=20extension=20i?= =?UTF-8?q?n=20properties.rs's=20module=20doc=20(runner-driving=20rows=20a?= =?UTF-8?q?dmitted=20=E2=80=94=20spawned=20run=5Fschedules(stop)=20tasks?= =?UTF-8?q?=20on=20a=20core=20StopToken=20against=20state=20outcomes=20onl?= =?UTF-8?q?y,=20elapsed-boundary-band=20tolerances,=20never=20timing-value?= =?UTF-8?q?=20assertions,=20never=20two-task=20race=20windows;=20ADR-017?= =?UTF-8?q?=20=C2=A72=20class=204)=20and=20three=20version-stamped=20rows:?= =?UTF-8?q?=20scheduler=5Fboundary=5Ffires=20(fired=20jobs=20are=20ordinar?= =?UTF-8?q?y=20claimable=20work=20with=20ScheduleOpts=20over=20the=20plain?= =?UTF-8?q?-queue=20derived=20defaults=20300/9/5/none=20+=20payload=20exac?= =?UTF-8?q?t,=20clean-stop=20Ok(()),=20fired=20count=20inside=20the=20elap?= =?UTF-8?q?sed-boundary=20band=20=E2=80=94=20no=20double-fire=20per=20boun?= =?UTF-8?q?dary=20while=20one=20leader=20runs;=20ADR-009=20=C2=A73/=C2=A74?= =?UTF-8?q?,=20ADR-020=20=C2=A73,=20ADR-019=20=C2=A74),=20scheduler=5Fboun?= =?UTF-8?q?ded=5Fcatchup=20(runner-less=20downtime=20proves=20no=20fire=20?= =?UTF-8?q?without=20a=20runner,=20=E2=89=A53=20elapsed=20boundaries=20rep?= =?UTF-8?q?lay=20boundary-by-boundary=20bounded=20below=20the=2064-cap=20a?= =?UTF-8?q?nd=20inside=20the=20band;=20ADR-009=20=C2=A74),=20scheduler=5Fl?= =?UTF-8?q?eadership=5Fdiscipline=20(two=20spawned=20runners=20on=20one=20?= =?UTF-8?q?store:=20exactly=20one=20Ok(())/Err(LeadershipLost)=20pair=20by?= =?UTF-8?q?=20value,=20no=20duplicated=20fires=20inside=20the=20band;=20AD?= =?UTF-8?q?R-009=20=C2=A71/=C2=A76,=20ADR-019=20=C2=A74)=20=E2=80=94=20wir?= =?UTF-8?q?ed=20into=20both=20engines'=20suite=20targets=20(SQLite=20tests?= =?UTF-8?q?,=20pg=20harness=5Frow!s),=20suite=20tokio=20dep=20added=20for?= =?UTF-8?q?=20the=20runner=20rows.=20Dispositions=20recorded=20in=20Notes:?= =?UTF-8?q?=20the=20beyond-cap=20skip-forward=20leg=20stays=20pinned=20eng?= =?UTF-8?q?ine-side=20(both=20engines'=20scheduler=20tests=20already=20bac?= =?UTF-8?q?kdate=20next=5Ffire=5Fat=20directly=20=E2=80=94=20catch=5Fup=5F?= =?UTF-8?q?replays=5Fup=5Fto=5Fthe=5Fcap=5Fthen=5Fskips=5Fforward=20twins)?= =?UTF-8?q?,=20and=20the=20pg=20fire-wake=20parity=20gap=20is=20not=20dema?= =?UTF-8?q?nded=20by=20these=20rows'=20shapes=20(claim-polling=20observati?= =?UTF-8?q?on=20only;=20the=20one-call=20wake=5Ftx=20disposition=20recorde?= =?UTF-8?q?d=20for=20a=20later=20task).=20Verified:=20sqlite=20suite=2016/?= =?UTF-8?q?16,=20pg=20suite=2016/16=20vs=20harness=20twice=20+=20solo=20re?= =?UTF-8?q?-runs=20of=20the=20three=20rows=20per=20engine=20(determinism),?= =?UTF-8?q?=20workspace=20build/test=20green,=20clippy=20-D=20warnings,=20?= =?UTF-8?q?fmt=20clean?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- alkstore-contract-suite/Cargo.toml | 4 +- alkstore-contract-suite/src/lib.rs | 3 +- alkstore-contract-suite/src/properties.rs | 403 +++++++++++++++++++++- alkstore-postgres/tests/contract_suite.rs | 19 +- alkstore-sqlite/tests/contract_suite.rs | 19 +- tasks/suite-scheduler-rows.md | 91 ++++- 6 files changed, 523 insertions(+), 16 deletions(-) diff --git a/alkstore-contract-suite/Cargo.toml b/alkstore-contract-suite/Cargo.toml index 134f7a9..6c56487 100644 --- a/alkstore-contract-suite/Cargo.toml +++ b/alkstore-contract-suite/Cargo.toml @@ -9,6 +9,4 @@ publish = false [dependencies] alkstore = { version = "0.1", path = "../alkstore" } serde_json = "1" - -[dev-dependencies] -tokio = { version = "1", features = ["macros", "rt"] } \ No newline at end of file +tokio = { version = "1", features = ["rt", "time"] } \ No newline at end of file diff --git a/alkstore-contract-suite/src/lib.rs b/alkstore-contract-suite/src/lib.rs index 7d2d38f..a0947d2 100644 --- a/alkstore-contract-suite/src/lib.rs +++ b/alkstore-contract-suite/src/lib.rs @@ -42,5 +42,6 @@ pub use properties::{ 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, + queue_depth_reclaim_and_dead_letter, receiver_close_and_save_arms, scheduler_boundary_fires, + scheduler_bounded_catchup, scheduler_leadership_discipline, sweep_no_stranded_rows, }; diff --git a/alkstore-contract-suite/src/properties.rs b/alkstore-contract-suite/src/properties.rs index a9857f7..1d735e2 100644 --- a/alkstore-contract-suite/src/properties.rs +++ b/alkstore-contract-suite/src/properties.rs @@ -7,21 +7,36 @@ //! is the suite's purpose as a test-side artifact (relaxing the //! family-wide no-panic rule here only). //! -//! Determinism: no wall-clock-timing assertions and no two-task race -//! windows — orchestration that would need one (runner spawning, -//! cross-connection wakes) stays engine-test-side; rows keep to -//! single-store sequential drives and the factory's isolation -//! guarantee. Rows exercise a *close/dispose* arm by dropping the +//! Determinism: no wall-clock-timing-value assertions and no two-task +//! race windows; rows keep to single-store sequential drives and the +//! factory's isolation guarantee. +//! +//! Orchestration posture: cross-connection wake orchestration stays +//! engine-test-side. **Runner-driving rows are admitted as an +//! extension** (the backlog's scheduler row demands the tick machinery +//! in the suite, and it cannot be pinned sequentially — +//! `run_schedules` is the drive): a spawned `run_schedules(stop)` +//! task with a core [`StopToken`], driven against state outcomes only +//! (jobs fired, leadership returned), with tolerances bounded well +//! above the boundary resolution — elapsed-boundary bands, never +//! timing-value assertions, never two-task race windows. The wave-5 +//! scheduler rows are the extension's riders; this is a suite-side +//! posture change (ADR-017 §2 class 4 — the suite tests the contract, +//! it is not the contract). +//! +//! 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 //! rows would be factory-contract tests (ADR-022), not contract rows, //! so none exist. +use std::sync::Arc; + use serde_json::json; use alkstore::{ - BoxedFuture, Delivery, EnqueueOpts, Error, JobHandle, JobState, RESERVED_PREFIX, Result, - StreamEvent, encode_payload, validate_shared_name, + BoxedFuture, Delivery, EnqueueOpts, Error, JobHandle, JobState, QueueOpts, RESERVED_PREFIX, + Result, ScheduleOpts, StopToken, Store, StreamEvent, encode_payload, validate_shared_name, }; use crate::factory::StoreFactory; @@ -1547,3 +1562,377 @@ fn unix_now() -> i64 { .map(|d| d.as_secs() as i64) .unwrap_or(0) } + +type RunnerJoin = tokio::task::JoinHandle>; + +/// Spawn a `run_schedules(stop)` runner driving off one shared store +/// handle (the runner-driving posture: the row holds the observing +/// Arc, the spawned task holds its clone; leadership is storage-level, +/// not handle-level). The token clone returned cancels only this +/// runner's stop. +fn spawn_runner(store: Arc) -> (StopToken, RunnerJoin) { + let stop = StopToken::new(); + let owned = store.clone(); + let token = stop.clone(); + let handle = tokio::spawn(async move { owned.run_schedules(token).await }); + (stop, handle) +} + +/// Cancel and join a runner: `Ok` of the runner's own return shape +/// (clean stop `Ok(())` — ADR-019 §4 — or `LeadershipLost`), or a +/// panic if the join itself fails or the stop is not prompt. +async fn join_runner(stop: StopToken, runner: RunnerJoin) -> Result<()> { + stop.cancel(); + match tokio::time::timeout(std::time::Duration::from_secs(10), runner).await { + Ok(joined) => match joined { + Ok(outcome) => outcome, + Err(e) => panic!("the runner task join-failed: {e}"), + }, + Err(_) => panic!("the stopped runner must terminate promptly, not hang"), + } +} + +/// Claim everything currently claimable from `queue` (one worker: +/// "suite-probe") and append the new fire ids to `fired`. Claiming is +/// the observation surface only — the fire rows are ordinary queue +/// work. +async fn collect_fires(store: &dyn Store, queue: &str, fired: &mut Vec) { + let q = store + .queue(queue, QueueOpts::default()) + .await + .expect("queue handle for the fire probe"); + loop { + let page = q + .claim_batch("suite-probe", 50) + .await + .expect("the fire probe claims"); + let got = page.len() as i64; + for handle in page { + let id = handle.job().id; + if !fired.contains(&id) { + fired.push(id); + } + } + if got < 50 { + break; + } + } +} + +/// Poll for fires until at least `min` distinct ids have landed, or +/// panic well-labelled after `deadline` (the tolerance-bounded wait: +/// the deadline is a safety bound, the assertions are on state +/// outcomes). +async fn await_fires( + store: &dyn Store, + queue: &str, + fired: &mut Vec, + min: usize, + deadline: std::time::Duration, + label: &str, +) { + let end = std::time::Instant::now() + deadline; + loop { + collect_fires(store, queue, fired).await; + if fired.len() >= min { + return; + } + assert!( + std::time::Instant::now() < end, + "{label}: only {} boundary fires within the wait window", + fired.len() + ); + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + } +} + +/// The elapsed-boundary band's upper bound for a run started at +/// `started`: boundaries elapsed since start, plus slack for the +/// unix-second boundary arithmetic and the in-progress tick at +/// cancel (the engine tests' fire-window tolerance). +fn elapsed_boundary_band(started: &std::time::Instant) -> i64 { + started.elapsed().as_secs() as i64 + 2 +} + +/// **Scheduler boundary fires** — the registered tick machinery, run: +/// a spawned runner fires due `@every 1s` boundaries into the named +/// queue as ordinary claimable jobs stamped with the `ScheduleOpts` +/// over the plain-queue derived defaults (`priority`/`max_attempts` +/// ride; visibility/backoff/retention take the engine defaults +/// 300/5/none), the fired rows are `get_job`-visible queue work with +/// the schedule's payload exactly, and a cancelled stop terminates the +/// runner promptly with the clean `Ok(())` shape. No double-fire per +/// boundary while one leader runs (the fire + boundary-advance commit +/// atomically — ADR-009 §4's one-firer discipline): the total fired +/// count stays inside the elapsed-boundary band above. +/// +/// Contract stamp: ADR-009 §3 (boundary fires enqueue the schedule's +/// payload with `ScheduleOpts` into the named queue — ordinary queue +/// work); ADR-009 §4 (one firer per boundary, fire + advance atomic); +/// ADR-020 §3 (the stamp source: `ScheduleOpts` over the engine's +/// plain-queue derived defaults); ADR-019 §4 (the clean-stop `Ok(())` +/// return shape). +pub async fn scheduler_boundary_fires(factory: &dyn StoreFactory) { + const LABEL: &str = "scheduler_boundary_fires"; + let store: Arc = Arc::from(factory.open().await.expect("factory opens a store")); + let started = std::time::Instant::now(); + + store + .schedule( + "suite-tick", + "@every 1s", + "work", + json!({"kind": "tick"}), + ScheduleOpts { + priority: 5, + max_attempts: Some(9), + expires: None, + }, + ) + .await + .expect("schedule registers"); + + let (stop, runner) = spawn_runner(store.clone()); + let mut fired: Vec = Vec::new(); + await_fires( + store.as_ref(), + "work", + &mut fired, + 2, + std::time::Duration::from_secs(25), + LABEL, + ) + .await; + + // ADR-020 §3's stamp source: ScheduleOpts applied over the + // plain-queue derived defaults (the fire resolves engine-side; the + // queue was never opened before the fire). + let q = store + .queue("work", QueueOpts::default()) + .await + .expect("queue handle"); + let job = q + .get_job(fired[0]) + .await + .expect("get_job reads the fire row") + .expect("the fire row exists"); + assert_eq!( + job.state, + JobState::Processing, + "the fire probe claimed this row" + ); + assert_eq!(job.priority, 5, "ScheduleOpts priority rides the fire"); + assert_eq!( + job.max_attempts, 9, + "ScheduleOpts max_attempts overrides the default" + ); + assert_eq!( + job.visibility_timeout_s, 300, + "visibility stamps the engine default" + ); + assert_eq!(job.backoff_base_s, 5, "backoff stamps the engine default"); + assert_eq!( + job.dead_letter_retention_s, None, + "retention stamps the engine default" + ); + assert_eq!( + job.payload, + json!({"kind": "tick"}).to_string().into_bytes(), + "the schedule's payload is the fire's payload, exactly" + ); + + let outcome = join_runner(stop, runner).await; + match outcome { + Ok(()) => {} + other => panic!("{LABEL}: a cancelled stop returns Ok(()), got {other:?}"), + } + + // One firer per boundary: the total fired count stays inside the + // elapsed-boundary band (a double-fire per boundary would blow + // past it). + collect_fires(store.as_ref(), "work", &mut fired).await; + let band = elapsed_boundary_band(&started); + assert!( + fired.len() >= 2, + "{LABEL}: two boundaries must have fired, got {}", + fired.len() + ); + assert!( + (fired.len() as i64) <= band, + "{LABEL}: no double-fire per boundary while one leader runs — \ + {} fires vs the elapsed-boundary band {band}", + fired.len() + ); + + factory.teardown().await.expect("factory teardown"); +} + +/// **Scheduler bounded catch-up** — downtime replay is at-least-once +/// per elapsed boundary, bounded by the 64-cap: register, let several +/// boundaries elapse with no runner in sight (schedules never fire +/// without one — ADR-009 §1's opt-in posture), then start one: the +/// missed boundaries replay boundary-by-boundary — strictly more fires +/// land than a coalescing single-fire would leave — the count stays +/// bounded by the 64-cap and inside the elapsed-boundary band, and the +/// runner stops clean. +/// +/// The beyond-cap skip-forward leg (>64 elapsed boundaries ≈ 65+ s of +/// downtime, plus a row-backdate the public surface does not express — +/// the schedule row's `next_fire_at` is engine-internal) pins +/// engine-side, where both engines' scheduler tests manipulate +/// `next_fire_at` directly +/// (`catch_up_replays_up_to_the_cap_then_skips_forward` on each); the +/// suite pins the ≤64 catch-up leg, the suite-observable one. +/// +/// Contract stamp: ADR-009 §4 (at-least-once per elapsed boundary, +/// bounded catch-up — the fixed 64-per-tick contract constant, beyond +/// which missed boundaries skip forward); ADR-019 §4 (the clean-stop +/// `Ok(())` return shape). +pub async fn scheduler_bounded_catchup(factory: &dyn StoreFactory) { + const LABEL: &str = "scheduler_bounded_catchup"; + let store: Arc = Arc::from(factory.open().await.expect("factory opens a store")); + let started = std::time::Instant::now(); + + store + .schedule( + "suite-backlog", + "@every 1s", + "catchup", + json!("w"), + Default::default(), + ) + .await + .expect("schedule registers"); + + // Downtime posture: no runner exists and nothing fires for it. + let mut fired: Vec = Vec::new(); + collect_fires(store.as_ref(), "catchup", &mut fired).await; + assert!( + fired.is_empty(), + "schedules never fire without a runner, got {} fires", + fired.len() + ); + // Several boundaries elapse (the first sits one interval past + // registration; a 4 s runner-less pause leaves at least three + // elapsed) — still nothing, no runner in sight. + past_stamp_sleep(4_000); + collect_fires(store.as_ref(), "catchup", &mut fired).await; + assert!( + fired.is_empty(), + "elapsed boundaries pile up as schedule-row state, not fires, got {}", + fired.len() + ); + + // Now run: the first tick replays the elapsed boundaries + // boundary-by-boundary. + let (stop, runner) = spawn_runner(store.clone()); + await_fires( + store.as_ref(), + "catchup", + &mut fired, + 3, + std::time::Duration::from_secs(25), + LABEL, + ) + .await; + let outcome = join_runner(stop, runner).await; + match outcome { + Ok(()) => {} + other => panic!("{LABEL}: a cancelled stop returns Ok(()), got {other:?}"), + } + collect_fires(store.as_ref(), "catchup", &mut fired).await; + + // The bounded-catch-up property: at-least-once per elapsed + // boundary (strictly more than a coalescing fire), bounded by the + // 64-cap (a multi-second downtime stays far below it). + let count = fired.len() as i64; + assert!( + count >= 3, + "{LABEL}: multi-boundary downtime must replay boundary-by-boundary \ + (at-least-once per elapsed boundary), got {count} fires" + ); + assert!( + count <= 64, + "{LABEL}: the replay is bounded by the 64-cap, got {count}" + ); + let band = elapsed_boundary_band(&started); + assert!( + count <= band, + "{LABEL}: fires stay inside the elapsed-boundary band ({count} vs {band})" + ); + + factory.teardown().await.expect("factory teardown"); +} + +/// **Scheduler leadership discipline** — two concurrent runners on one +/// store: exactly one leader, the other surfaces `LeadershipLost` +/// rather than silently co-ticking (the acquire-time refusal is the +/// same return-shape outcome as a mid-run loss), the fired count stays +/// inside the elapsed-boundary band with both runners alive (fires are +/// not duplicated), and the winning leader's cancelled stop returns +/// the clean `Ok(())` shape (ADR-019 §4; the loser's return is the +/// matchable `Err(LeadershipLost)`, never an ordinary error or a +/// silent `Ok`). +/// +/// Contract stamp: ADR-009 §1 (one leader at a time; leadership loss +/// ends the runner); ADR-009 §6 (`LeadershipLost` — the matchable +/// return-shape variant of the runner's stop); ADR-019 §4 (the +/// clean-stop `Ok(())` return shape under a cancelled token). +pub async fn scheduler_leadership_discipline(factory: &dyn StoreFactory) { + const LABEL: &str = "scheduler_leadership_discipline"; + let store: Arc = Arc::from(factory.open().await.expect("factory opens a store")); + let started = std::time::Instant::now(); + + store + .schedule( + "suite-lead", + "@every 1s", + "single", + json!(1), + Default::default(), + ) + .await + .expect("schedule registers"); + + // Two concurrent runners on the same store: the leadership lock is + // storage-level, so exactly one is granted and drives the tick. + let (stop_a, runner_a) = spawn_runner(store.clone()); + let (stop_b, runner_b) = spawn_runner(store.clone()); + let mut fired: Vec = Vec::new(); + await_fires( + store.as_ref(), + "single", + &mut fired, + 1, + std::time::Duration::from_secs(25), + LABEL, + ) + .await; + // A few more boundaries with both runners alive. + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + + let outcome_a = join_runner(stop_a, runner_a).await; + let outcome_b = join_runner(stop_b, runner_b).await; + + let clean = |o: &Result<()>| matches!(o, Ok(())); + let lost = |o: &Result<()>| matches!(o, Err(Error::LeadershipLost)); + assert!( + (clean(&outcome_a) && lost(&outcome_b)) || (lost(&outcome_a) && clean(&outcome_b)), + "{LABEL}: exactly one leader — one clean stop and one \ + `LeadershipLost`, got A {outcome_a:?} and B {outcome_b:?}" + ); + + // One firer per boundary while both runners were alive: the fired + // count stays inside the elapsed-boundary band (duplicated fires + // would blow past it). + collect_fires(store.as_ref(), "single", &mut fired).await; + let count = fired.len() as i64; + let band = elapsed_boundary_band(&started); + assert!( + count >= 1 && count <= band, + "{LABEL}: fires not duplicated under two runners — {count} fires \ + vs the elapsed-boundary band {band}" + ); + + factory.teardown().await.expect("factory teardown"); +} diff --git a/alkstore-postgres/tests/contract_suite.rs b/alkstore-postgres/tests/contract_suite.rs index ec18d00..584c479 100644 --- a/alkstore-postgres/tests/contract_suite.rs +++ b/alkstore-postgres/tests/contract_suite.rs @@ -167,7 +167,9 @@ mod rows { 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, - queue_depth_reclaim_and_dead_letter, receiver_close_and_save_arms, sweep_no_stranded_rows, + queue_depth_reclaim_and_dead_letter, receiver_close_and_save_arms, + scheduler_boundary_fires, scheduler_bounded_catchup, scheduler_leadership_discipline, + sweep_no_stranded_rows, }; use super::PgFactory; @@ -248,6 +250,21 @@ mod rows { sweep_no_stranded_rows, "row-sweep-stranded" ); + harness_row!( + row_scheduler_boundary_fires, + scheduler_boundary_fires, + "row-scheduler-boundary" + ); + harness_row!( + row_scheduler_bounded_catchup, + scheduler_bounded_catchup, + "row-scheduler-catchup" + ); + harness_row!( + row_scheduler_leadership_discipline, + scheduler_leadership_discipline, + "row-scheduler-leadership" + ); } mod factory_shape { diff --git a/alkstore-sqlite/tests/contract_suite.rs b/alkstore-sqlite/tests/contract_suite.rs index 9d16ac5..b4e0b72 100644 --- a/alkstore-sqlite/tests/contract_suite.rs +++ b/alkstore-sqlite/tests/contract_suite.rs @@ -71,7 +71,9 @@ mod rows { 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, - queue_depth_reclaim_and_dead_letter, receiver_close_and_save_arms, sweep_no_stranded_rows, + queue_depth_reclaim_and_dead_letter, receiver_close_and_save_arms, + scheduler_boundary_fires, scheduler_bounded_catchup, scheduler_leadership_discipline, + sweep_no_stranded_rows, }; use super::SqliteFactory; @@ -139,6 +141,21 @@ mod rows { async fn row_sweep_no_stranded_rows() { sweep_no_stranded_rows(&factory("row-sweep-stranded")).await; } + + #[tokio::test(flavor = "multi_thread")] + async fn row_scheduler_boundary_fires() { + scheduler_boundary_fires(&factory("row-scheduler-boundary")).await; + } + + #[tokio::test(flavor = "multi_thread")] + async fn row_scheduler_bounded_catchup() { + scheduler_bounded_catchup(&factory("row-scheduler-catchup")).await; + } + + #[tokio::test(flavor = "multi_thread")] + async fn row_scheduler_leadership_discipline() { + scheduler_leadership_discipline(&factory("row-scheduler-leadership")).await; + } } mod factory_shape { diff --git a/tasks/suite-scheduler-rows.md b/tasks/suite-scheduler-rows.md index 89088ca..5ac5640 100644 --- a/tasks/suite-scheduler-rows.md +++ b/tasks/suite-scheduler-rows.md @@ -1,7 +1,7 @@ --- id: suite-scheduler-rows name: Contract-suite rows — scheduler boundary/catch-up/leadership (+ determinism-posture extension) -status: pending +status: completed depends_on: [] scope: moderate risk: medium @@ -86,8 +86,93 @@ latency-posture in Notes and leave the engines as they are. ## Notes -> To be filled by implementation agent +- **Beyond-cap skip-forward leg: pinned engine-side, already done in + both engines.** The leg needs >64 elapsed boundaries (~65+ s of + downtime) and a direct schedule-row backdate the public surface does + not express (`next_fire_at` is engine-internal). Disposition: pinned + where `next_fire_at` is directly manipulable — + `catch_up_replays_up_to_the_cap_then_skips_forward` exists in + `alkstore-sqlite/src/store/scheduler_tests.rs` (substrate row + backdate, exactly 64 fires, skip-forward past now, skipped + boundaries never enqueue) and the twin in + `alkstore-postgres/src/store/scheduler_tests.rs` (admin-SQL backdate, + same shape). The suite row (`scheduler_bounded_catchup`) pins the + ≤64 in-band catch-up leg — the suite-observable one — and states the + engine-side disposition in its doc text. +- **Fire-wake parity: not demanded by this row set; engines left + as-is.** The suite rows observe fire outcomes through ordinary + claim polling, never through wake latency, so the row shapes demand + no wake. The recorded cross-engine latency-parity gap stands (the pg + tick's in-tx enqueue fires no `pg_notify`; SQLite's watcher wakes on + those commits — `tasks/review-wave-4-fixes.md`'s recorded-for-later + note): correctness is covered by the queues row's pinned re-poll + safety net on both engines; the row set here pins state outcomes + with tolerances bounded well above the boundary resolution, so wake + latency is out of scope. If a later task demands the parity, the + one-call `wake_tx` inside the pg tick's tx is the shape to ride. +- **Runner-driving posture details** (the extension's riders, beyond + what the description pinned): the spawned runners share the row's + single store handle via `Arc` clones — leadership is + storage-level, not handle-level, so two concurrent runners on one + handle still contend on the leadership lock (the SQLite engine's own + tests open a second handle on the same path; the suite's factory + gives one isolated store per row, and the shared-handle shape + exercises the same storage-level discipline). A leadership-lock + acquire through the writer slot is a short lease on both engines, so + the shared handle neither deadlocks nor serializes the tick away. + The two-concurrent-runners row takes whichever runner wins (no + identity assumption): exactly one `Ok(())` and one + `Err(LeadershipLost)` pair, checked by value. +- **Band tolerances** (the determinism posture's "bounded well above + the boundary resolution"): every fired-count assertion is bounded by + an elapsed-boundary band measured from just before registration + (`unix-second boundary arithmetic` slack + one in-progress tick at + cancel, +2 s — the engine tests' fire-window tolerance), never a + timing value. The catch-up row's lower bound (≥ 3 fires) is the + boundary-by-boundary pin — strictly more than a coalescing + single-fire would leave; its registration's first-fire grace of one + interval is counted in the band. ## Summary -> To be filled on completion \ No newline at end of file +**Landed:** the determinism-posture extension in +`alkstore-contract-suite/src/properties.rs`'s module doc (runner-driving +rows admitted — spawned `run_schedules(stop)` tasks with a core +`StopToken`, state outcomes only, elapsed-boundary-band tolerances, +never timing-value assertions, never two-task race windows; ADR-017 §2 +class 4 posture note, cross-connection wake orchestration still +engine-test-side) and the three scheduler rows, version-stamped and +re-exported from the suite lib: + +- `scheduler_boundary_fires` — stamps ADR-009 §3/§4, ADR-020 §3, + ADR-019 §4; spawned runner, ≥2 boundary fires observed via ordinary + claim polling, the fire row's stamp source pinned (ScheduleOpts + priority/max_attempts over the plain-queue derived defaults + 300/5/none, payload exact), clean-stop `Ok(())` prompt, fired count + inside the elapsed-boundary band (no double-fire per boundary under + one leader). +- `scheduler_bounded_catchup` — stamp ADR-009 §4 (plus ADR-019 §4); + runner-less downtime proves schedules never fire without a runner, + a 4 s pause yields ≥3 replayed fires (boundary-by-boundary, not + coalesced) bounded below the 64-cap and inside the band; the + beyond-cap skip-forward disposition recorded in Notes (pinned + engine-side on both engines, already implemented there). +- `scheduler_leadership_discipline` — stamps ADR-009 §1/§6, + ADR-019 §4; two spawned runners on one store, exactly one + `Ok(())`/`Err(LeadershipLost)` pair by value, fired count inside the + elapsed-boundary band with both alive (no duplicated fires). + +Wired into `alkstore-sqlite/tests/contract_suite.rs` (three direct +tests) and `alkstore-postgres/tests/contract_suite.rs` (three +`harness_row!` entries). Suite crate gained `tokio` (rt + time) as a +regular dependency for the runner-driving rows (the unused +dev-dependency dropped). + +**Gates:** `cargo build`; `cargo test -p alkstore-sqlite` (189 lib + +16 suite rows green); `cargo test -p alkstore-postgres` serverless +green (pg rows skip cleanly); the full pg column green against the +harness server (`pglo-poc` :15432 — 121 lib incl. the previously +skipped rows + 16 suite rows + 9 schema); the three new rows re-run +twice solo on each engine (determinism check, green both columns both +runs); `cargo clippy --all-targets -- -D warnings`; `cargo fmt --check` +(clean after fmt). \ No newline at end of file