SQLite engine: scheduler + outbox — schedule/unschedule/run_schedules leader loop, outbox enqueue/run_once (task sqlite-engine-scheduler-outbox)

alkstore-sqlite/src/store/outbox_tests.rs

@@ -0,0 +1,376 @@
+//! The outbox mechanism's acceptance tests (`sqlite-engine-scheduler-outbox`):
+//! the derived backing queue is unreachable by `enqueue_tx` (the
+//! reserved-prefix rejection — ADR-014's stated guarantee), the
+//! outbox's `enqueue` stamps the 60/5/5 derived set, `run_once` acks
+//! on `Ok`, retries with the curve on `Err`, and returns `false` on an
+//! empty backing queue; worker_id validation (ADR-019 §2). The
+//! commit-atomic tx half pins under tx_tests (already landed with the
+//! seam task).
+
+use std::path::PathBuf;
+use std::time::Duration;
+
+use alkstore::{EnqueueOpts, JobState, Store};
+use serde_json::json;
+
+use crate::store::open_store;
+
+fn temp_dir(tag: &str) -> PathBuf {
+    let d = std::env::temp_dir().join(format!(
+        "alkstore-outbox-{tag}-{}-{:?}",
+        std::process::id(),
+        std::thread::current().id()
+    ));
+    let _ = std::fs::remove_dir_all(&d);
+    std::fs::create_dir_all(&d).unwrap();
+    d
+}
+
+fn cleanup(dir: &PathBuf) {
+    let _ = std::fs::remove_dir_all(dir);
+}
+
+fn open_test_store(tag: &str) -> (PathBuf, Box<dyn Store>) {
+    let dir = temp_dir(tag);
+    let path = dir.join("store.db");
+    let store: Box<dyn Store> =
+        Box::new(open_store(path.to_str().unwrap(), Default::default()).unwrap());
+    (dir, store)
+}
+
+/// Count rows in the outbox's backing queue directly (the reserved
+/// name is unreachable by the handle surface by design).
+fn backing_count(dir: &std::path::Path, outbox: &str) -> i64 {
+    let conn = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+    conn.query_row(
+        "SELECT count(*) FROM __alkstore_live WHERE queue = ?1",
+        [format!("__alkstore_outbox:{outbox}")],
+        |r| r.get(0),
+    )
+    .unwrap()
+}
+
+/// A minimal delivery impl driving `run_once` from the test side. The
+/// error it feeds is replayed as the exhaust string (`retry(Some(e),
+/// None)` carries the delivery error's `Display`), so the test holds
+/// both ends of that mapping.
+struct Scripted {
+    outcome: Result<(), String>,
+    seen_payloads: Vec<Vec<u8>>,
+}
+
+impl Scripted {
+    fn error(&self) -> alkstore::Error {
+        alkstore::Error::Codec(self.outcome.clone().unwrap_err())
+    }
+}
+
+impl alkstore::Delivery for Scripted {
+    fn deliver<'a>(
+        &'a mut self,
+        job: Box<dyn alkstore::JobHandle>,
+    ) -> alkstore::BoxedFuture<'a, alkstore::Result<()>> {
+        self.seen_payloads.push(job.job().payload.clone());
+        let outcome = self.outcome.clone().map_err(alkstore::Error::Codec);
+        Box::pin(async move { outcome })
+    }
+}
+
+/// Acceptance: the constructor validates (empty → `InvalidName`,
+/// reserved → `ReservedName`) and the handle name reads back.
+#[tokio::test(flavor = "multi_thread")]
+async fn outbox_constructor_validates() {
+    let (dir, store) = open_test_store("outbox-construction");
+
+    let check = |err: alkstore::Error, expect: &str, name: &str| match err {
+        alkstore::Error::InvalidName { .. } => assert_eq!(expect, "empty", "{name}"),
+        alkstore::Error::ReservedName { .. } => assert_eq!(expect, "reserved", "{name}"),
+        other => panic!("{name} got the wrong error: {other}"),
+    };
+    let empty_err = match store.outbox("").await {
+        Err(e) => e,
+        Ok(_) => panic!("outbox(\"\") must fail entry-point validation"),
+    };
+    check(empty_err, "empty", "empty outbox name");
+    let reserved_err = match store.outbox("__alkstore_outbox:steal").await {
+        Err(e) => e,
+        Ok(_) => panic!("outbox(reserved) must fail entry-point validation"),
+    };
+    check(reserved_err, "reserved", "reserved outbox name");
+
+    let outbox = store.outbox("billing").await.unwrap();
+    assert_eq!(outbox.name(), "billing");
+    cleanup(&dir);
+}
+
+/// Acceptance: the derived backing queue is a reserved name —
+/// `enqueue_tx` into it rejects (`ReservedName`; ADR-014 §1's "no
+/// third path" guarantee) and `enqueue` stamps the 60/5/5 derived set
+/// with the one `max_attempts` override (raw row probes: the backing
+/// queue has no `Queue` handle surface).
+#[tokio::test(flavor = "multi_thread")]
+async fn outbox_enqueue_stamps_the_derived_set_and_backing_is_unreachable() {
+    let (dir, store) = open_test_store("outbox-stamps");
+
+    let err = store
+        .begin_tx()
+        .await
+        .unwrap()
+        .enqueue_tx("__alkstore_outbox:billing", Default::default(), json!({}))
+        .await
+        .unwrap_err();
+    assert!(
+        matches!(err, alkstore::Error::ReservedName { .. }),
+        "the derived backing queue name must be unreachable by enqueue_tx: {err:?}"
+    );
+
+    let outbox = store.outbox("billing").await.unwrap();
+    assert_eq!(outbox.name(), "billing");
+
+    let id = outbox
+        .enqueue(json!({"email": "x@y.z"}), EnqueueOpts::default())
+        .await
+        .unwrap();
+
+    let stamps = {
+        let conn = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+        conn.query_row(
+            "SELECT visibility_timeout_s, max_attempts, backoff_base_s, \
+                    dead_letter_retention_s, payload \
+             FROM __alkstore_live WHERE id = ?1",
+            [id],
+            |r| {
+                Ok((
+                    r.get::<_, i64>(0)?,
+                    r.get::<_, i64>(1)?,
+                    r.get::<_, i64>(2)?,
+                    r.get::<_, Option<i64>>(3)?,
+                    r.get::<_, String>(4)?,
+                ))
+            },
+        )
+        .unwrap()
+    };
+    assert_eq!((stamps.0, stamps.1, stamps.2, stamps.3), (60, 5, 5, None));
+    assert_eq!(stamps.4, json!({"email": "x@y.z"}).to_string());
+
+    // The one per-job override resolves over the outbox set.
+    let id2 = outbox
+        .enqueue(
+            json!({}),
+            EnqueueOpts {
+                max_attempts: Some(2),
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap();
+    let max2: i64 = {
+        let conn = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+        conn.query_row(
+            "SELECT max_attempts FROM __alkstore_live WHERE id = ?1",
+            [id2],
+            |r| r.get(0),
+        )
+        .unwrap()
+    };
+    assert_eq!(max2, 2);
+
+    assert_eq!(backing_count(&dir, "billing"), 2);
+    cleanup(&dir);
+}
+
+/// Acceptance: `run_once` — `false` on an empty backing queue; ack on
+/// `Ok` (the row is gone from live); `retry(Some(e), None)` on `Err`
+/// (the row returns to pending with the queue's curve stamped; the
+/// delivery's error string would surface on a later dead row).
+#[tokio::test(flavor = "multi_thread")]
+async fn run_once_acks_on_ok_and_retries_on_err() {
+    let (dir, store) = open_test_store("outbox-runonce");
+
+    let mut outbox = store.outbox("mail").await.unwrap();
+    let id_ok = outbox
+        .enqueue(json!({"n": 1}), Default::default())
+        .await
+        .unwrap();
+    let id_err = outbox
+        .enqueue(json!({"n": 2}), Default::default())
+        .await
+        .unwrap();
+
+    // Empty-behavior first on a different outbox: no work → `false`.
+    let mut quiet = store.outbox("quiet").await.unwrap();
+    let mut noop = Scripted {
+        outcome: Ok(()),
+        seen_payloads: Vec::new(),
+    };
+    assert!(!quiet.run_once("w0", &mut noop).await.unwrap());
+
+    // Ok ⇒ ack: the claim is processed and the row leaves live.
+    let mut ok_delivery = Scripted {
+        outcome: Ok(()),
+        seen_payloads: Vec::new(),
+    };
+    assert!(outbox.run_once("w1", &mut ok_delivery).await.unwrap());
+    assert_eq!(ok_delivery.seen_payloads.len(), 1);
+    assert_eq!(
+        ok_delivery.seen_payloads[0],
+        json!({"n": 1}).to_string().into_bytes()
+    );
+
+    // Err ⇒ retry with the curve: the row returns to pending, the
+    // backoff curve's draw must land ≥ base/2 (base 5 → draw ∈ [3, 5]).
+    let mut err_delivery = Scripted {
+        outcome: Err(" smtp down".to_string()),
+        seen_payloads: Vec::new(),
+    };
+    assert!(outbox.run_once("w2", &mut err_delivery).await.unwrap());
+    assert_eq!(err_delivery.seen_payloads.len(), 1);
+    assert_eq!(
+        err_delivery.seen_payloads[0],
+        json!({"n": 2}).to_string().into_bytes()
+    );
+
+    // The backing queue is reserved: get job state through the raw row.
+    let (state, run_at_gap, attempts): (String, i64, i64) = {
+        let conn = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+        let now: i64 = conn
+            .query_row("SELECT unixepoch()", [], |r| r.get(0))
+            .unwrap();
+        conn.query_row(
+            "SELECT state, run_at - ?2, attempts FROM __alkstore_live WHERE id = ?1",
+            rusqlite::params![id_err, now],
+            |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
+        )
+        .unwrap()
+    };
+    assert_eq!(state, "pending", "the Err outcome retries into pending");
+    assert!(
+        (1..=6).contains(&run_at_gap),
+        "the queue curve draw ∈ [2, 5] (+1 clk): got gap {run_at_gap}"
+    );
+    assert_eq!(attempts, 1, "the first claim counts an attempt");
+
+    assert_eq!(backing_count(&dir, "mail"), 1, "the acked row left live");
+    let _ = id_ok;
+    cleanup(&dir);
+}
+
+/// Acceptance: `run_once`'s no-heartbeat posture — the engine issues
+/// no heartbeat during delivery; the boxed handle's own `heartbeat` is
+/// the consumer's renewal path (ADR-019 §5). Exercised as the reach of
+/// the boxed handle: the consumer closure heartbeats inside delivery
+/// and the row's claim deadline resets forward absolutely (inspected
+/// through the raw row — the claimed `Job` snapshot keeps the
+/// claim-time deadline by design).
+#[tokio::test(flavor = "multi_thread")]
+async fn run_once_delivery_can_heartbeat_the_claim_itself() {
+    let (dir, store) = open_test_store("outbox-heartbeat");
+
+    let mut outbox = store.outbox("hb").await.unwrap();
+    outbox
+        .enqueue(json!("work"), Default::default())
+        .await
+        .unwrap();
+
+    struct Renew {
+        db_path: std::path::PathBuf,
+    }
+    impl alkstore::Delivery for Renew {
+        fn deliver<'a>(
+            &'a mut self,
+            job: Box<dyn alkstore::JobHandle>,
+        ) -> alkstore::BoxedFuture<'a, alkstore::Result<()>> {
+            let db_path = self.db_path.clone();
+            Box::pin(async move {
+                assert!(job.heartbeat(3000).await.unwrap(), "renewal lands");
+                assert_eq!(job.job().state, JobState::Processing);
+                // Inspect the row while the claim is held (a separate
+                // WAL reader connection): the deadline moved to
+                // ≈ now + 3000 — the heartbeat's absolute reset, not
+                // the stamped 60 s visibility.
+                let conn = rusqlite::Connection::open(&db_path).unwrap();
+                let (claim_expires_at, db_now): (i64, i64) = conn
+                    .query_row(
+                        "SELECT claim_expires_at, unixepoch() FROM __alkstore_live WHERE id = ?1",
+                        [job.job().id],
+                        |r| Ok((r.get(0)?, r.get(1)?)),
+                    )
+                    .unwrap();
+                assert!(
+                    claim_expires_at > db_now + 2000,
+                    "the consumer's heartbeat renewed the deadline forward: \
+                     row {claim_expires_at} vs now {db_now}"
+                );
+                Ok(())
+            })
+        }
+    }
+    let mut d = Renew {
+        db_path: dir.join("store.db"),
+    };
+    assert!(outbox.run_once("w-hb", &mut d).await.unwrap());
+    cleanup(&dir);
+}
+
+/// Acceptance: `run_once`'s exhaustion path — a delivery that keeps
+/// failing past the outbox set's max_attempts (5) dead-letters the row
+/// with the delivery's error string carried (`retry(Some(e), None)`'s
+/// exhaust string) — driven here to the dead row via repeated
+/// `run_once` calls with deadline waits.
+#[tokio::test(flavor = "multi_thread")]
+async fn run_once_exhaustion_dead_letters_with_the_delivery_error() {
+    let (dir, store) = open_test_store("outbox-dead");
+
+    let mut outbox = store.outbox("doom").await.unwrap();
+    let id = outbox
+        .enqueue(
+            json!("boom"),
+            EnqueueOpts {
+                max_attempts: Some(2),
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap();
+
+    let mut failing = Scripted {
+        outcome: Err("always fails".to_string()),
+        seen_payloads: Vec::new(),
+    };
+
+    // max_attempts = 2 (the per-job override over the outbox set) → two
+    // claim+fail cycles; the curve's draw ∈ [2, 5]+ with base 5.
+    for round in 1..=2 {
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(15);
+        loop {
+            if outbox.run_once("w-doom", &mut failing).await.unwrap() {
+                break;
+            }
+            assert!(
+                tokio::time::Instant::now() < deadline,
+                "round {round}: retry never came claimable"
+            );
+            tokio::time::sleep(Duration::from_millis(150)).await;
+        }
+    }
+    // The second retry exhausts the budget: the row is dead with the
+    // delivery's error string carried (`retry(Some(e), None)`'s exhaust
+    // string — ADR-019 §5's outcome posture through the engine).
+    let (last_error, died): (String, i64) = {
+        let conn = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+        conn.query_row(
+            "SELECT last_error, died_at FROM __alkstore_dead WHERE id = ?1",
+            [id],
+            |r| Ok((r.get(0)?, r.get(1)?)),
+        )
+        .unwrap()
+    };
+    assert_eq!(last_error, failing.error().to_string());
+    assert!(died > 0);
+
+    // Nothing left to do: run_once returns false on the drained queue.
+    assert!(!outbox.run_once("w-doom", &mut failing).await.unwrap());
+    assert_eq!(failing.seen_payloads.len(), 2);
+    cleanup(&dir);
+}

alkstore-sqlite/src/store/scheduler_tests.rs

@@ -0,0 +1,545 @@
+//! The scheduler mechanism's acceptance tests (`sqlite-engine-scheduler-outbox`):
+//! upsert + read-back with the entry-point validation triad pinned
+//! (`InvalidSpec` pre-storage with no round trip, `InvalidName`/
+//! `ReservedName` on both name and queue argument), `unschedule`'s
+//! true/false, the leader loop firing due boundaries into the named
+//! queue (claimable via the ordinary queue surface), the leadership
+//! lock held and one-firer-per-boundary under a second concurrent
+//! runner, `Err(LeadershipLost)` vs clean `Ok(())` stop, fired jobs'
+//! stamp source (the plain defaults with `ScheduleOpts` applied,
+//! ADR-020 §3), and catch-up-cap/skip-forward observability through
+//! the pinned substrate behavior (ADR-009, ADR-019 §3/§4).
+
+use std::path::PathBuf;
+use std::time::Duration;
+
+use alkstore::{Error, JobState, QueueOpts, ScheduleOpts, StopToken, Store};
+use serde_json::json;
+
+use crate::store::open_store;
+
+fn temp_dir(tag: &str) -> PathBuf {
+    let d = std::env::temp_dir().join(format!(
+        "alkstore-sched-{tag}-{}-{:?}",
+        std::process::id(),
+        std::thread::current().id()
+    ));
+    let _ = std::fs::remove_dir_all(&d);
+    std::fs::create_dir_all(&d).unwrap();
+    d
+}
+
+fn cleanup(dir: &PathBuf) {
+    let _ = std::fs::remove_dir_all(dir);
+}
+
+fn open_test_store(tag: &str) -> (PathBuf, Box<dyn Store>) {
+    let dir = temp_dir(tag);
+    let path = dir.join("store.db");
+    let store: Box<dyn Store> =
+        Box::new(open_store(path.to_str().unwrap(), Default::default()).unwrap());
+    (dir, store)
+}
+
+/// Acceptance: `schedule` validates at the entry point on all three
+/// arguments — empty/reserved schedule name (`InvalidName` /
+/// `ReservedName`), empty/reserved queue argument (same, ADR-021 §3),
+/// and a non-`@every` or malformed spec → `InvalidSpec` **before
+/// storage** (no round trip; a failing register observable via the
+/// absent registry row, and `@every 0s` rejected as non-positive).
+#[tokio::test(flavor = "multi_thread")]
+async fn schedule_entry_point_validation_is_pinned_on_all_arguments() {
+    let (dir, store) = open_test_store("sched-validation");
+
+    let check = |err: Error, expect: &str, name: &str| match err {
+        Error::InvalidName { .. } => assert_eq!(expect, "empty", "{name}"),
+        Error::ReservedName { .. } => assert_eq!(expect, "reserved", "{name}"),
+        Error::InvalidSpec { spec } => {
+            assert_eq!(expect, "spec", "{name}");
+            assert!(
+                !spec.is_empty(),
+                "{name}: InvalidSpec carries the spec string"
+            );
+        }
+        other => panic!("{name} got the wrong error: {other}"),
+    };
+
+    check(
+        store
+            .schedule("", "@every 1s", "work", json!(null), Default::default())
+            .await
+            .unwrap_err(),
+        "empty",
+        "empty schedule name",
+    );
+    check(
+        store
+            .schedule(
+                "__alkstore_rogue",
+                "@every 1s",
+                "work",
+                json!(null),
+                Default::default(),
+            )
+            .await
+            .unwrap_err(),
+        "reserved",
+        "reserved schedule name",
+    );
+    check(
+        store
+            .schedule("tick", "@every 1s", "", json!(null), Default::default())
+            .await
+            .unwrap_err(),
+        "empty",
+        "empty queue argument",
+    );
+    check(
+        store
+            .schedule(
+                "tick",
+                "@every 1s",
+                "__alkstore_outbox:steal",
+                json!(null),
+                Default::default(),
+            )
+            .await
+            .unwrap_err(),
+        "reserved",
+        "reserved queue argument",
+    );
+    for bad in [
+        "0 9 * * *",
+        "@every 1x",
+        "@every 0s",
+        "every 5s",
+        "@every s",
+        "@every 1.5s",
+    ] {
+        check(
+            store
+                .schedule("bad-spec", bad, "work", json!(null), Default::default())
+                .await
+                .unwrap_err(),
+            "spec",
+            &(format!("spec {bad:?}")),
+        );
+    }
+
+    // The rejected calls never reached storage (pre-storage, no round
+    // trip): the good name registers fresh and the registry holds only
+    // that one row.
+    let sched = store
+        .schedule(
+            "tick",
+            "@every 1s",
+            "work",
+            json!({"a": 1}),
+            Default::default(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(sched.name, "tick");
+    assert_eq!(sched.spec, "@every 1s");
+    assert_eq!(sched.queue, "work");
+    assert_eq!(sched.opts, ScheduleOpts::default());
+
+    cleanup(&dir);
+}
+
+/// Acceptance: `schedule` upserts by name — a re-register replaces the
+/// whole row (spec, queue, payload, opts and the next-fire reset),
+/// read-back reflects the replacement.
+#[tokio::test(flavor = "multi_thread")]
+async fn schedule_upsert_replaces_the_whole_row() {
+    let (dir, store) = open_test_store("sched-upsert");
+
+    let first = store
+        .schedule(
+            "sweep",
+            "@every 10s",
+            "old-queue",
+            json!({"v": 1}),
+            ScheduleOpts {
+                priority: 3,
+                max_attempts: Some(7),
+                expires: Some(120),
+            },
+        )
+        .await
+        .unwrap();
+    assert_eq!(first.queue, "old-queue");
+    assert_eq!(first.opts.priority, 3);
+
+    let second = store
+        .schedule(
+            "sweep",
+            "@every 20m",
+            "new-queue",
+            json!({"v": 2}),
+            ScheduleOpts {
+                priority: 0,
+                max_attempts: None,
+                expires: None,
+            },
+        )
+        .await
+        .unwrap();
+    assert_eq!(second.spec, "@every 20m");
+    assert_eq!(second.queue, "new-queue");
+    assert_eq!(second.opts, ScheduleOpts::default());
+
+    // unregister + re-register is the pause shape — the upsert makes
+    // update = re-register honest (ADR-009 §1).
+    assert!(store.unschedule("sweep").await.unwrap());
+    assert!(!store.unschedule("sweep").await.unwrap());
+
+    cleanup(&dir);
+}
+
+/// Acceptance: the leader loop fires due boundaries into the named
+/// queue, the fired jobs are ordinary queue work (`get_job`-visible
+/// with the plain-queue defaults + `ScheduleOpts` applied over them,
+/// ADR-020 §3; claimable via the queue surface), the leadership lock
+/// is held (`__alkstore_scheduler`), and a clean stop returns `Ok(())`.
+#[tokio::test(flavor = "multi_thread")]
+async fn run_schedules_fires_boundaries_and_stop_is_clean() {
+    let (dir, store) = open_test_store("sched-fire");
+
+    store
+        .schedule(
+            "fast",
+            "@every 1s",
+            "work",
+            json!({"kind": "sweep"}),
+            ScheduleOpts {
+                priority: 5,
+                max_attempts: Some(9),
+                expires: None,
+            },
+        )
+        .await
+        .unwrap();
+
+    let stop = StopToken::new();
+    let runner_store = {
+        // `run_schedules` borrows the store: spawn on a clone-shaped
+        // handle — the store is behind `Arc`-free trait object; run the
+        // loop in this task and stop it from this same task's timer.
+        // (A second store open on the same path shares the db.)
+        let path = dir.join("store.db");
+        open_store(path.to_str().unwrap(), Default::default()).unwrap()
+    };
+    let runner_stop = stop.clone();
+    let runner = tokio::spawn(async move { runner_store.run_schedules(runner_stop).await });
+
+    // Two boundaries at 1 s cadence (+ registration's first-fire grace
+    // of exactly one interval): wait for two fired jobs, bounded.
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(25);
+    let mut fired_ids = Vec::new();
+    loop {
+        let q = store.queue("work", QueueOpts::default()).await.unwrap();
+        let page = q.claim_batch("probe", 10).await.unwrap();
+        for h in page {
+            fired_ids.push(h.job().id);
+        }
+        if fired_ids.len() >= 2 {
+            break;
+        }
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "no two boundary fires within the window"
+        );
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    }
+
+    // The stamp source pin (ADR-020 §3): the plain defaults 300/3/5/none
+    // with ScheduleOpts applied — priority 5 and max_attempts 9 override,
+    // visibility/backoff/retention take the defaults.
+    let got = store.queue("work", QueueOpts::default()).await.unwrap();
+    let job = got.get_job(fired_ids[0]).await.unwrap().unwrap();
+    assert_eq!(job.state, JobState::Processing, "our probe claimed it");
+    assert_eq!(job.priority, 5);
+    assert_eq!(job.max_attempts, 9);
+    assert_eq!(job.visibility_timeout_s, 300);
+    assert_eq!(job.backoff_base_s, 5);
+    assert_eq!(job.dead_letter_retention_s, None);
+    assert_eq!(
+        job.payload,
+        json!({"kind": "sweep"}).to_string().into_bytes()
+    );
+    assert_eq!(job.worker_id.as_deref(), Some("probe"));
+
+    // The leadership lock row exists under the reserved name.
+    let leader = crate::lock::SCHEDULER_LEADER_LOCK;
+    let (count,) = {
+        // Raw probe through a fresh store connection: a reserved name
+        // is unreachable by try_lock (Shared validation would reject),
+        // so query the substrate table directly via a tx read on
+        // get_job-shaped machinery is wrong namespace; use the reader
+        // path with plain SQL through the engine's substrate module.
+        let conn = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+        conn.query_row(
+            "SELECT count(*) FROM __alkstore_locks WHERE name = ?1 AND expires_at > unixepoch()",
+            [leader],
+            |r| Ok((r.get::<_, i64>(0)?,)),
+        )
+        .unwrap()
+    };
+    assert_eq!(count, 1, "the leadership lock row must be live");
+
+    // Clean stop: `Ok(())`, promptly (not after the full sleep).
+    stop.cancel();
+    let outcome = tokio::time::timeout(Duration::from_secs(10), runner)
+        .await
+        .expect("clean stop must terminate promptly")
+        .unwrap();
+    assert!(
+        matches!(outcome, Ok(())),
+        "clean stop returns Ok(()): {outcome:?}"
+    );
+
+    drop(store);
+    cleanup(&dir);
+}
+
+/// Acceptance: a second concurrent runner does not double-fire — the
+/// leadership lock admits one leader; the loser returns
+/// `Err(LeadershipLost)` (no loop, no ticks), the winner fires each
+/// boundary exactly once (at-least-once per boundary with one firer,
+/// ADR-009 §4's writer-serialization floor).
+#[tokio::test(flavor = "multi_thread")]
+async fn second_runner_is_refused_and_boundaries_fire_once() {
+    let (dir, store) = open_test_store("sched-single");
+
+    store
+        .schedule("beat", "@every 1s", "single", json!(1), Default::default())
+        .await
+        .unwrap();
+
+    // Runner A takes leadership first.
+    let stop_a = StopToken::new();
+    let runner_a_store =
+        open_store(dir.join("store.db").to_str().unwrap(), Default::default()).unwrap();
+    let runner_a = tokio::spawn({
+        let stop = stop_a.clone();
+        let store = runner_a_store;
+        async move { store.run_schedules(stop).await }
+    });
+
+    // Runner B comes second: it must be refused (locked out) with
+    // `Err(LeadershipLost)` — start it once A has demonstrably taken
+    // the lock (wait for the first fired job).
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(20);
+    loop {
+        let q = store.queue("single", QueueOpts::default()).await.unwrap();
+        if !q.claim_batch("probe", 10).await.unwrap().is_empty() {
+            break;
+        }
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "runner A never fired a boundary"
+        );
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    }
+
+    let runner_b_store =
+        open_store(dir.join("store.db").to_str().unwrap(), Default::default()).unwrap();
+    let outcome_b = runner_b_store
+        .run_schedules(StopToken::new())
+        .await
+        .unwrap_err();
+    assert!(
+        matches!(outcome_b, Error::LeadershipLost),
+        "runner B must be refused with LeadershipLost: {outcome_b:?}"
+    );
+
+    // Let the leader run several more boundaries; the total fired count
+    // matches the elapsed boundaries within fire-window tolerance —
+    // with a second non-firing runner, one firer per boundary.
+    tokio::time::sleep(Duration::from_millis(2500)).await;
+    let fired_total: i64 = {
+        let conn = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+        conn.query_row(
+            "SELECT count(*) FROM __alkstore_live WHERE queue = 'single'",
+            [],
+            |r| r.get(0),
+        )
+        .unwrap()
+    };
+    assert!(
+        (2..=5).contains(&fired_total),
+        "boundary fires must stay within the elapsed-boundary band, got {fired_total}"
+    );
+
+    stop_a.cancel();
+    let outcome_a = tokio::time::timeout(Duration::from_secs(10), runner_a)
+        .await
+        .expect("clean stop must terminate promptly")
+        .unwrap();
+    assert!(matches!(outcome_a, Ok(())), "{outcome_a:?}");
+
+    drop(store);
+    cleanup(&dir);
+}
+
+/// Acceptance: losing leadership mid-run returns `Err(LeadershipLost)`
+/// *before any tick* — here exercised at the substrate boundary the
+/// engine's renew path rides (`lock_renew` refuses a foreign/expired
+/// owner row): a runner whose lease row was deleted (its TTL lapsed)
+/// fails its next renew and returns the loss, never a tick.
+#[tokio::test(flavor = "multi_thread")]
+async fn leadership_loss_returns_the_loss_error_before_ticking() {
+    let (dir, store) = open_test_store("sched-loss");
+
+    store
+        .schedule(
+            "slow",
+            "@every 3600s",
+            "lossq",
+            json!(1),
+            Default::default(),
+        )
+        .await
+        .unwrap();
+
+    // Steal the lock from under no one: a rogue operator (bypassing
+    // the engine surface — try_lock itself rejects the reserved name)
+    // inserts the leadership row directly; the runner starts, must
+    // observe LeadershipLost at acquire.
+    {
+        let conn = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+        conn.execute(
+            "INSERT INTO __alkstore_locks (name, owner, expires_at)
+             VALUES (?1, 'rogue-operator', unixepoch() + 30)",
+            [crate::lock::SCHEDULER_LEADER_LOCK],
+        )
+        .unwrap();
+    }
+
+    let runner_store =
+        open_store(dir.join("store.db").to_str().unwrap(), Default::default()).unwrap();
+    let outcome = runner_store.run_schedules(StopToken::new()).await;
+    assert!(
+        matches!(outcome, Err(Error::LeadershipLost)),
+        "a runner refused at acquire returns LeadershipLost: {outcome:?}"
+    );
+
+    // No tick ever ran: the 1-hour schedule's queue is untouched.
+    let fired: i64 = {
+        let conn = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+        conn.query_row(
+            "SELECT count(*) FROM __alkstore_live WHERE queue = 'lossq'",
+            [],
+            |r| r.get(0),
+        )
+        .unwrap()
+    };
+    assert_eq!(fired, 0, "a refused runner must never tick");
+
+    drop(store);
+    cleanup(&dir);
+}
+
+/// Acceptance: catch-up cap + skip-forward are observable through the
+/// runner (the substrate's pinned per-tick behavior, ADR-009 §4): a
+/// schedule with its fire horizon pushed far into the past replays at
+/// most the 64-fires cap and skips forward strictly past now.
+#[tokio::test(flavor = "multi_thread")]
+async fn catch_up_replays_up_to_the_cap_then_skips_forward() {
+    let (dir, store) = open_test_store("sched-catchup");
+
+    store
+        .schedule(
+            "backlog",
+            "@every 1h",
+            "catchq",
+            json!("w"),
+            Default::default(),
+        )
+        .await
+        .unwrap();
+
+    // Push the fire horizon 200 boundary-widths into the past (200
+    // elapsed boundaries > the 64 cap) before the runner ever starts.
+    // `@every 1h` keeps the post-cap boundary beyond the test window:
+    // the only fires possible here are the capped replay's own.
+    {
+        let conn = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+        let now: i64 = conn
+            .query_row("SELECT unixepoch()", [], |r| r.get(0))
+            .unwrap();
+        conn.execute(
+            "UPDATE __alkstore_scheduler_tasks SET next_fire_at = ?2 WHERE name = 'backlog'",
+            rusqlite::params!["backlog", now - 200 * 3600],
+        )
+        .unwrap();
+    }
+
+    let store2 = open_store(dir.join("store.db").to_str().unwrap(), Default::default()).unwrap();
+    let stop = StopToken::new();
+    let runner_stop = stop.clone();
+    let runner = tokio::spawn(async move { store2.run_schedules(runner_stop).await });
+
+    // The first tick fires exactly 64, advances past now (the
+    // skip-forward lands at the next 1 h boundary — far beyond this
+    // window), and the skipped boundaries never enqueue.
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(20);
+    let fired: i64 = loop {
+        let c: i64 = {
+            let conn = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+            conn.query_row(
+                "SELECT count(*) FROM __alkstore_live WHERE queue = 'catchq'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap()
+        };
+        let next: i64 = {
+            let conn = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+            conn.query_row(
+                "SELECT next_fire_at FROM __alkstore_scheduler_tasks WHERE name = 'backlog'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap()
+        };
+        let now: i64 = {
+            let conn = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+            conn.query_row("SELECT unixepoch()", [], |r| r.get(0))
+                .unwrap()
+        };
+        if c == 64 && next > now {
+            break c;
+        }
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "the cap never landed (fires={c}, next={next}, now={now})"
+        );
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    };
+    assert_eq!(fired, 64, "exactly the contract-constant cap replays");
+
+    stop.cancel();
+    let outcome = tokio::time::timeout(Duration::from_secs(10), runner)
+        .await
+        .expect("clean stop must terminate promptly")
+        .unwrap();
+    assert!(matches!(outcome, Ok(())), "{outcome:?}");
+
+    // After the stop: the skipped boundaries never enqueue — and the
+    // skip-forward's next 1 h boundary cannot have fired in-window.
+    let fired_after: i64 = {
+        let conn = rusqlite::Connection::open(dir.join("store.db")).unwrap();
+        conn.query_row(
+            "SELECT count(*) FROM __alkstore_live WHERE queue = 'catchq'",
+            [],
+            |r| r.get(0),
+        )
+        .unwrap()
+    };
+    assert_eq!(fired_after, 64, "the skipped boundaries never enqueue");
+
+    drop(store);
+    cleanup(&dir);
+}