feat(store): the sweep's durable form — staged candidates, next-sweep execution

The first durable delete window (single-node), beside the inline form:

- sweep_window branches on the delete window's host — the kv-engine
  enum (KvEngine::Sqlite -> durable; KvEngine::Mem -> inline body,
  extracted verbatim so the wave-2 mem/local suites stay green
  unchanged)
- the durable form, per sweep under the lease and after the shared
  protect consult: window_take(now) executes the overdue set with
  execution-time re-arbitration through the same arbitrated_delete
  path (consult_protect = false — the sweep's protect consult covers
  its execution phase; vanished candidates settle into the cancelled
  count), then marks unchanged, then stages the candidate-dead set at
  visible_after = mark_end + window (the constructor-tunable delay;
  ZERO = execute-on-next-sweep)
- report counts exact relative to the schedule (staged-this-sweep vs
  executed-this-sweep); aborts execute nothing, stage nothing, delete
  nothing
- unit tests: two-sweep exact-count shapes at delay 0, future-delay
  staging with idempotent re-stage, the race across the delay (put
  during the window survives; the unpinned twin deletes), the
  vanished-candidate form, and the three abort shapes on sqlite

Verification: cargo test (default 69 lib tests), --all-features (124),
--no-default-features --features mem (101), clippy -D warnings
(default and --all-features), cargo fmt --check — all green.

src/gc.rs

@@ -65,11 +65,10 @@
 //!   report data (`Ok(SweepReport)` — ADR-012 §1), nothing deleted.
 //!
 //! The `#![allow(dead_code)]` below covers the companion-seam scope:
-//! the window family has its first in-tree caller in wave 3 (store
-//! construction runs `ensure_gc_schema` on the SQL-backed kv engine),
-//! while `window_stage`/`window_take` wait for the sweep's durable
-//! form and the fleet family waits for postgres's wave. It shrinks as
-//! those land.
+//! the window family has its in-tree callers (store construction runs
+//! `ensure_gc_schema` on the SQL-backed kv engine; the sweep's durable
+//! form drives `window_stage`/`window_take`), while the fleet family
+//! waits for postgres's wave. It shrinks as that lands.
 
 #![allow(dead_code)]
 

src/store.rs

@@ -204,25 +204,26 @@
 //! `EngineWindowState` seam) at construction, dispatched on the
 //! kv-engine enum — a mem arm is structurally nothing, the in-process
 //! engines execute the ADR-005 window protocol inline, so there is no
-//! schema to ensure — and the SQL-backed sweep form (wave 3's
-//! sweep-durable-window task) stages its candidates on the kv engine
-//! the constructor built. The host branch keys off the engine
-//! representation — one more kv-engine-enum match arm — so wave 5's
+//! schema to ensure — and the sweep's SQL-backed form (**the durable
+//! form**, ADR-008 §2 staged re-arbitrated delete; ADR-012 §6.7 the
+//! executor is the next sweep) stages its candidates on the kv engine
+//! the constructor built. The form branch keys off the engine
+//! representation ([`Store::window_host`] — one kv-engine-enum match
+//! arm, never a flag): the durable form runs where the host is
+//! SQL-backed, the inline form verbatim on mem/local, so wave 5's
 //! pg-lo dormancy (the dual-SQL instance's large-engine window clauses
 //! are dormant when the kv engine already hosts) slots in without
 //! reshaping. Both sqlite constructors carry the delete-window delay
 //! (ADR-008 §2: default 5 min, constructor-tunable per deployment —
 //! [`DEFAULT_DELETE_WINDOW_DELAY`]; `0` is a legal degenerate tuning,
-//! execute-on-next-sweep) as the cadence knob the durable sweep form
-//! consumes; the mem/local constructors' inline form has no delay in
-//! effect.
+//! execute-on-next-sweep) as the cadence knob the durable form's
+//! staging consumes (`visible_after = mark_end + window`); the
+//! mem/local constructors' inline form has no delay in effect.
 //!
 //! The `#![allow(dead_code)]` below marks the work still ahead, not
-//! the put path (which is driven below): the GC home's fleet clauses
-//! and the durable window's cadence knob (until the sweep's durable
-//! form consumes it) hold the real scope. The kv engine set is no
-//! longer empty in default builds — sqlite landed in wave 3 — and the
-//! empty-set `allow`s came off with it.
+//! the GC protocol (which is driven below — the durable window's
+//! cadence knob is consumed by the durable form's staging). The fleet
+//! clauses hold the remaining scope until the fleet engine's wave.
 
 #![allow(dead_code)]
 
@@ -238,9 +239,7 @@ use crate::facade::{
     Batch, Blob, EntryMeta, GcAbortCause, LivenessSource, Pin, ProtectCallback, Registration,
     SweepReport,
 };
-#[cfg(feature = "sqlite")]
-use crate::gc::EngineWindowState;
-use crate::gc::{Arbitration, GcHome};
+use crate::gc::{Arbitration, EngineWindowState, GcHome, WindowCandidate};
 use crate::hashing::{Algorithm, BlobHasher};
 use crate::keys::Key;
 use crate::local::{LocalBackend, LocalCursor, LocalStaging};
@@ -1372,32 +1371,53 @@ impl Store {
     }
 
     /// **`sweep()`** — the explicit mark-and-sweep with the delete
-    /// window (ADR-005 §Sweep), in its non-SQL inline form (ADR-012 §5:
-    /// `local`/`mem` stage candidates in-process and execute the window
-    /// protocol inline in the same sweep). Sweeps are never errors:
+    /// window (ADR-005 §Sweep), in two forms under one signature: the
+    /// **durable form** runs where the kv engine is SQL-backed and hosts
+    /// the window (ADR-012 §5's one-window-host rule — sqlite in today's
+    /// set; the branch keys off the engine enum, never a flag), and the
+    /// **inline form** (ADR-012 §5: `local`/`mem` stage candidates
+    /// in-process and execute the window protocol inline in the same
+    /// sweep) runs verbatim elsewhere. Sweeps are never errors:
     /// every abort is report data in `SweepReport.aborted`
-    /// (`Ok(SweepReport)`), and a completed sweep's counts are exact:
-    /// `candidates_staged == deleted + cancelled_by_arbitration`
-    /// (`deleted`/`cancelled` are both 0 on any abort).
+    /// (`Ok(SweepReport)`), and a completed sweep's counts are exact
+    /// relative to its form's schedule — the inline form reports
+    /// `candidates_staged == deleted + cancelled_by_arbitration` over its
+    /// same-sweep staged set; the durable form reports
+    /// `candidates_staged` as the candidates staged *this* sweep and
+    /// `deleted`/`cancelled` as the *overdue* set executed this sweep
+    /// (staged and executed are different sets there — the delay
+    /// separates them, ADR-012 §6.7; `deleted`/`cancelled` are both 0 on
+    /// any abort).
     ///
     /// The protocol, as run here:
     ///
     /// 1. **Single-flight**: the in-process sweep lease (ADR-013 §2's
     ///    single-node analog); a contended second sweeper aborts with
-    ///    [`GcAbortCause::SweeperLock`], nothing deleted.
+    ///    [`GcAbortCause::SweeperLock`], nothing deleted, nothing
+    ///    staged or executed.
     /// 2. **Zero registered liveness sources** aborts with
-    ///    [`GcAbortCause::NoLivenessSources`] before anything is marked
-    ///    — the safe default (nothing deleted).
+    ///    [`GcAbortCause::NoLivenessSources`] before anything is marked —
+    ///    the safe default (nothing deleted).
     /// 3. **Protect**: each registered [`ProtectCallback`] is consulted
     ///    pre-sweep; its keys union into the live set; a callback that
     ///    fails to answer aborts with [`GcAbortCause::ProtectFailed`] —
     ///    a flaky protection source skips the sweep rather than risk
-    ///    deletion (iroh's `ProtectOutcome::Abort` conclusion).
-    /// 4. **Mark**: the whole pool enumerated (both tiers' complete
-    ///    `list()`, ADR-003 — a malformed pool list is fatal
-    ///    `Err(KeyInvalid)`); the live set computed from all three
-    ///    sources (registered sources, pins, protect callbacks); the
-    ///    candidate-dead set staged (sorted, in-process).
+    ///    deletion (iroh's `ProtectOutcome::Abort` conclusion). On the
+    ///    durable form this consult's coverage extends to the executor
+    ///    phase, so an abort here precedes any overdue execution too.
+    /// 4. **Durable form (SQL-backed kv engine)**, per sweep under the
+    ///    lease: **execute overdue** first (window_take re-arbitrates
+    ///    each overdue candidate at execution time — pin-before-publish
+    ///    during the window delay cancels the candidate — and deletes in
+    ///    batches), then **mark** (unchanged), then **stage** the fresh
+    ///    candidate-dead set at `visible_after = mark_end + window` (the
+    ///    constructor-tunable delay, ADR-008 §2). The executor of the
+    ///    staged rows is the next sweep (ADR-008 §2; ADR-012 §6.7).
+    ///    **Inline form**: the mark phase enumerates the whole pool
+    ///    (both tiers' complete `list()`, ADR-003 — a malformed pool
+    ///    list is fatal `Err(KeyInvalid)`); the live set computed from
+    ///    all three sources (registered sources, pins, protect
+    ///    callbacks); the candidate-dead set staged (sorted, in-process).
     /// 5. **Delete window**: per candidate, **arbitration at delete
     ///    time** (not once upfront) under the arbitration lock — pinned
     ///    or re-observed-live cancels the candidate (counted in
@@ -1409,7 +1429,9 @@ impl Store {
     /// is never deleted by a sweep started before it committed — the
     /// in-flight delete markers plus the blocked pin mint are the
     /// mechanism ([`crate::gc`]'s module doc), and the concurrency
-    /// tests below assert it.
+    /// tests below assert it (the durable form's race holds *across the
+    /// window delay* — a put landing during the delay pins in-process
+    /// and the execution-time re-arbitration cancels the candidate).
     pub async fn sweep(&self) -> Result<SweepReport, Error> {
         let started = SystemTime::now();
 
@@ -1438,10 +1460,16 @@ impl Store {
         result
     }
 
-    /// The mark → arbitrate-at-delete-time → batched commit body,
-    /// under the sweep lease. Operational failures (engine I/O, a
-    /// malformed pool list) are the `Err` half of the return shape
-    /// (ADR-012 §1); every deliberate no-op is report data instead.
+    /// The sweep body, under the sweep lease: the protect consult
+    /// (whose abort shape is [`GcAbortCause::ProtectFailed`] — and whose
+    /// coverage extends to the durable form's execution phase), then
+    /// the form branch on the delete window's host — the kv-engine
+    /// enum itself (ADR-012 §5's one-window-host rule), never a flag:
+    /// the **durable form** on a SQL-backed kv engine, the **inline
+    /// form** verbatim on `local`/`mem`. Operational failures (engine
+    /// I/O, a malformed pool or window list) are the `Err` half of the
+    /// return shape (ADR-012 §1); every deliberate no-op is report data
+    /// instead.
     async fn sweep_window(&self, started: SystemTime) -> Result<SweepReport, Error> {
         let mut live: BTreeSet<Key> = BTreeSet::new();
 
@@ -1462,6 +1490,39 @@ impl Store {
             }
         }
 
+        match &self.kv {
+            #[cfg(feature = "mem")]
+            KvEngine::Mem(_) => self.sweep_window_inline(live, started).await,
+            #[cfg(feature = "sqlite")]
+            KvEngine::Sqlite(engine) => {
+                // The durable form (ADR-008 §2; ADR-012 §6.7): the
+                // overdue staged rows execute first — each re-arbitrated
+                // at execution time — then this sweep's own mark and
+                // stage at `visible_after = mark_end + window`.
+                let (deleted, cancelled_by_arbitration) =
+                    self.execute_overdue_window(engine).await?;
+                let candidates_staged = self.mark_and_stage_window(engine, live).await?;
+                Ok(Self::sweep_report(
+                    candidates_staged,
+                    deleted,
+                    cancelled_by_arbitration,
+                    None,
+                    started,
+                ))
+            }
+        }
+    }
+
+    /// The inline form (ADR-012 §5): the mark → arbitrate-at-delete-
+    /// time → batched commit body, unchanged — the non-SQL engines
+    /// stage candidates in-process and execute the window protocol
+    /// inline in the same sweep. Preserved verbatim across the durable
+    /// form's arrival.
+    async fn sweep_window_inline(
+        &self,
+        mut live: BTreeSet<Key>,
+        started: SystemTime,
+    ) -> Result<SweepReport, Error> {
         let mut pool = typed_keys(self.kv().list().await?)?;
         if let Some(large) = self.large() {
             pool.extend(typed_keys(large.list().await?)?);
@@ -1503,6 +1564,78 @@ impl Store {
         ))
     }
 
+    /// **The durable form's executor step** (ADR-012 §6.7): take the
+    /// overdue staged candidates (`visible_after <= now`) off the
+    /// window and execute each through the same
+    /// [`Store::arbitrated_delete`] path — the re-arbitration happens
+    /// at execution time, so a put landing (pin-before-publish) during
+    /// the window delay cancels its candidate here — in ~100-key
+    /// batches. The consult posture on this path is the sweep's protect
+    /// consult (recorded named design point): it already ran for this
+    /// sweep, above, and its coverage extends to the executor phase —
+    /// so these arbitrations carry `consult_protect = false`, the same
+    /// posture as the inline form's delete-time arbitration. A vanished
+    /// candidate (deleted during the delay by a direct delete) settles
+    /// into the cancelled count.
+    async fn execute_overdue_window<W: EngineWindowState>(
+        &self,
+        host: &W,
+    ) -> Result<(u64, u64), Error> {
+        let overdue: Vec<Key> = typed_keys(host.window_take(SystemTime::now()).await?)?;
+        let mut deleted = 0u64;
+        let mut cancelled_by_arbitration = 0u64;
+        for chunk in overdue.chunks(SWEEP_DELETE_BATCH) {
+            for key in chunk {
+                match self.arbitrated_delete(key, false).await? {
+                    crate::gc::DeleteOutcome::Deleted(true) => deleted += 1,
+                    _ => cancelled_by_arbitration += 1,
+                }
+            }
+        }
+        Ok((deleted, cancelled_by_arbitration))
+    }
+
+    /// **The durable form's mark + stage step**: the mark phase exactly
+    /// the inline form's (pool enumerated, live set from pins/registered
+    /// sources plus the protect consult's keys, candidate-dead set
+    /// computed), then the candidate-dead set staged into the window at
+    /// `visible_after = mark_end + window` — the constructor-tunable
+    /// delay (ADR-008 §2; the store carries the value its constructor
+    /// took), so execution waits at earliest for the next sweep past
+    /// the deadline. Returns the staged-candidate count.
+    async fn mark_and_stage_window<W: EngineWindowState>(
+        &self,
+        host: &W,
+        mut live: BTreeSet<Key>,
+    ) -> Result<u64, Error> {
+        let mut pool = typed_keys(self.kv().list().await?)?;
+        if let Some(large) = self.large() {
+            pool.extend(typed_keys(large.list().await?)?);
+        }
+        live.extend(self.gc.pinned_keys());
+        for source in self.gc.registered_sources() {
+            live.extend(source.live_keys());
+        }
+
+        let enumerated: BTreeSet<Key> = pool.into_iter().collect();
+        let mut candidates: Vec<Key> = enumerated.difference(&live).cloned().collect();
+        candidates.sort();
+
+        let mark_end = SystemTime::now();
+        let visible_after = mark_end.checked_add(self.window_delay).ok_or_else(|| {
+            Error::Io("the delete-window delay overflows the system clock".to_owned())
+        })?;
+        let staged: Vec<WindowCandidate> = candidates
+            .iter()
+            .map(|key| WindowCandidate {
+                key: key.as_bytes().to_vec(),
+                visible_after,
+            })
+            .collect();
+        host.window_stage(&staged).await?;
+        Ok(candidates.len() as u64)
+    }
+
     /// A finished report constructor — every sweep outcome is a
     /// completed record (ADR-012 §1: the report is data, not an error
     /// carrier).
@@ -4011,3 +4144,501 @@ mod sqlite_construction_tests {
         drop(pin);
     }
 }
+
+/// The sweep's durable form (feature `sqlite`): the staged
+/// re-arbitrated delete (ADR-008 §2; ADR-012 §5/§6.7) — the first
+/// durable delete window, single-node — beside the inline form. The
+/// two-sweep exact-count shapes, the race across the delay, the
+/// vanished-candidate form, the abort shapes, and the delay knob's
+/// staging semantics.
+#[cfg(all(test, feature = "sqlite"))]
+mod sqlite_window_tests {
+    use super::*;
+    use crate::backend::Backend;
+    use std::fs;
+    use std::path::{Path, PathBuf};
+    use std::sync::atomic::{AtomicU64, Ordering};
+    use std::time::{SystemTime, UNIX_EPOCH};
+
+    /// A fresh temp site per test (the construction tests' helper
+    /// shape); Drop removes the file tree and the WAL sidecars.
+    struct TempSite {
+        database: PathBuf,
+        root: PathBuf,
+    }
+
+    impl TempSite {
+        fn path(&self) -> &Path {
+            &self.root
+        }
+
+        fn drop_sidecar(&self, suffix: &str) {
+            let base = self.database.to_string_lossy().into_owned();
+            let _ = fs::remove_file(format!("{base}{suffix}"));
+        }
+    }
+
+    impl Drop for TempSite {
+        fn drop(&mut self) {
+            self.drop_sidecar("-wal");
+            self.drop_sidecar("-shm");
+            let _ = fs::remove_file(&self.database);
+            let _ = fs::remove_dir_all(&self.root);
+        }
+    }
+
+    fn temp_site(tag: &str) -> TempSite {
+        static SEQ: AtomicU64 = AtomicU64::new(0);
+        let seq = SEQ.fetch_add(1, Ordering::Relaxed);
+        let nanos = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map(|d| d.as_nanos())
+            .unwrap_or(0);
+        let base = std::env::temp_dir().join(format!(
+            "alkblobs-store-window-{tag}-{}-{nanos:x}-{seq}",
+            std::process::id()
+        ));
+        fs::create_dir_all(&base).expect("temp site created");
+        TempSite {
+            database: base.join("pool.db"),
+            root: base.join("local-root"),
+        }
+    }
+
+    fn key_for(content: &[u8]) -> Key {
+        Key::derive(content, content.len() as u64)
+    }
+
+    fn sqlite_engine(store: &Store) -> &SqliteBackend {
+        match store.kv() {
+            #[cfg(feature = "mem")]
+            KvEngine::Mem(_) => panic!("these tests construct sqlite stores"),
+            #[cfg(feature = "sqlite")]
+            KvEngine::Sqlite(engine) => engine,
+        }
+    }
+
+    async fn plant_kv(store: &Store, key: &Key, value: &[u8]) {
+        sqlite_engine(store)
+            .put(&key.as_bytes(), value)
+            .await
+            .expect("kv plant");
+    }
+
+    async fn plant_large(store: &Store, key: &Key, value: &[u8]) {
+        match store.large() {
+            Some(LargeEngine::Local(engine)) => engine
+                .put(&key.as_bytes(), value)
+                .await
+                .expect("large plant"),
+            None => panic!("planting into a mode with no large tier"),
+        }
+    }
+
+    /// A set-backed liveness source — the live-shared seam, shared
+    /// with the wave-2 tests' fixture shape.
+    #[derive(Default)]
+    struct SetSource {
+        live: parking_lot::Mutex<BTreeSet<Key>>,
+    }
+
+    impl SetSource {
+        fn add(&self, key: Key) {
+            self.live.lock().insert(key);
+        }
+    }
+
+    impl LivenessSource for SetSource {
+        fn live_keys(&self) -> Vec<Key> {
+            self.live.lock().iter().cloned().collect()
+        }
+
+        fn contains(&self, key: &Key) -> bool {
+            self.live.lock().contains(key)
+        }
+    }
+
+    /// A set-backed protect callback; `None` models the
+    /// failed-to-answer case.
+    #[derive(Default)]
+    struct SetProtect {
+        set: parking_lot::Mutex<Option<Vec<Key>>>,
+    }
+
+    impl SetProtect {
+        fn set_keys(&self, keys: Vec<Key>) {
+            *self.set.lock() = Some(keys);
+        }
+
+        fn fail(&self) {
+            *self.set.lock() = None;
+        }
+    }
+
+    impl ProtectCallback for SetProtect {
+        fn protect(&self) -> Option<Vec<Key>> {
+            self.set.lock().clone()
+        }
+    }
+
+    /// A protect callback that blocks mid-consult — the contention
+    /// window for the single-flight assert on a sqlite store.
+    struct LatchProtect {
+        entered: std::sync::mpsc::Sender<()>,
+        release: parking_lot::Mutex<Option<std::sync::mpsc::Receiver<()>>>,
+    }
+
+    impl ProtectCallback for LatchProtect {
+        fn protect(&self) -> Option<Vec<Key>> {
+            self.entered.send(()).ok();
+            let receiver = self.release.lock().take();
+            if let Some(receiver) = receiver {
+                let _ = receiver.recv();
+            }
+            Some(Vec::new())
+        }
+    }
+
+    /// A non-destructive read of the window is not on the seam — but a
+    /// far-horizon `window_take` drains it entirely, which is exactly
+    /// the observation the assertions below need (and leaves the
+    /// window clean for the next step of the test).
+    async fn window_drain(store: &Store) -> Vec<Key> {
+        let far = SystemTime::now()
+            .checked_add(Duration::from_secs(365 * 24 * 3600))
+            .unwrap();
+        let raws = sqlite_engine(store).window_take(far).await.unwrap();
+        typed_keys(raws).expect("the window holds typed keys")
+    }
+
+    /// The two-sweep exact-count shape at the degenerate delay (the
+    /// test tuning, ADR-008 §2): sweep 1 stages N — nothing deleted,
+    /// the pool intact; sweep 2 executes exactly N —
+    /// `deleted + cancelled == N` against the staged set, staged-this-
+    /// sweep and executed-this-sweep different sets, and the window
+    /// left empty.
+    #[tokio::test]
+    async fn durable_two_sweep_exact_counts_at_delay_zero() {
+        let site = temp_site("two-sweep");
+        let store = Store::dual_tier_sqlite_local(&site.database, site.path(), 64, Duration::ZERO)
+            .await
+            .expect("dual-tier sqlite+local constructs");
+        let source = Arc::new(SetSource::default());
+        store.register_liveness_source(source.clone());
+
+        let live = key_for(b"kept by the registered source");
+        plant_kv(&store, &live, b"kept by the registered source").await;
+        source.add(live.clone());
+
+        let dead_kv = key_for(b"dead kv entry in the window");
+        plant_kv(&store, &dead_kv, b"dead kv entry in the window").await;
+        let dead_large = key_for(b"dead large entry in the window");
+        plant_large(&store, &dead_large, b"dead large entry in the window").await;
+
+        let first = store.sweep().await.expect("sweeps are never errors");
+        assert_eq!(first.aborted, None);
+        assert_eq!(
+            first.candidates_staged, 2,
+            "both tiers' dead entries staged"
+        );
+        assert_eq!(
+            first.deleted, 0,
+            "staging is not executing — the delay separates"
+        );
+        assert_eq!(first.cancelled_by_arbitration, 0);
+
+        assert!(
+            store.has(&dead_kv.as_bytes()).await.unwrap(),
+            "sweep 1 staged, did not execute: the kv entry is intact"
+        );
+        assert!(
+            store.has(&dead_large.as_bytes()).await.unwrap(),
+            "sweep 1 staged, did not execute: the large entry is intact"
+        );
+        assert!(store.has(&live.as_bytes()).await.unwrap());
+
+        let second = store.sweep().await.expect("sweeps are never errors");
+        assert_eq!(second.aborted, None);
+        assert_eq!(second.candidates_staged, 0, "nothing newly dead");
+        assert_eq!(
+            second.deleted + second.cancelled_by_arbitration,
+            2,
+            "sweep 2 executed exactly the staged set"
+        );
+        assert_eq!(second.deleted, 2);
+        assert_eq!(second.cancelled_by_arbitration, 0);
+
+        assert!(!store.has(&dead_kv.as_bytes()).await.unwrap());
+        assert!(!store.has(&dead_large.as_bytes()).await.unwrap());
+        assert!(store.has(&live.as_bytes()).await.unwrap());
+        assert!(
+            window_drain(&store).await.is_empty(),
+            "the executed rows left the window"
+        );
+    }
+
+    /// The delay knob is the staging's cadence (ADR-008 §2): at a
+    /// future-facing delay the staged rows are not overdue at
+    /// `window_take(now)` — no execution, no deletion — and a repeated
+    /// sweep re-stages the still-dead set without duplicating rows.
+    #[tokio::test]
+    async fn durable_stage_at_a_future_delay_waits_past_the_window() {
+        let site = temp_site("future-delay");
+        let store = Store::kv_only_sqlite(&site.database, 64, Duration::from_secs(600))
+            .await
+            .expect("kv-only sqlite constructs");
+        store.register_liveness_source(Arc::new(SetSource::default()));
+
+        let dead_a = key_for(b"dead under a future-facing delay a");
+        plant_kv(&store, &dead_a, b"dead under a future-facing delay a").await;
+        let dead_b = key_for(b"dead under a future-facing delay b");
+        plant_kv(&store, &dead_b, b"dead under a future-facing delay b").await;
+
+        let first = store.sweep().await.expect("sweeps are never errors");
+        assert_eq!(first.aborted, None);
+        assert_eq!(first.candidates_staged, 2);
+        assert_eq!(first.deleted, 0, "nothing is overdue at staging time");
+
+        let second = store.sweep().await.expect("sweeps are never errors");
+        assert_eq!(second.candidates_staged, 2, "still dead, re-staged");
+        assert_eq!(
+            second.deleted, 0,
+            "the window delay pushed execution out again"
+        );
+
+        let drained = window_drain(&store).await;
+        let mut expected = vec![dead_a.clone(), dead_b.clone()];
+        expected.sort();
+        assert_eq!(drained, expected);
+        assert!(store.has(&dead_a.as_bytes()).await.unwrap());
+        assert!(store.has(&dead_b.as_bytes()).await.unwrap());
+    }
+
+    /// The race across the delay (the window form of ADR-005's
+    /// test-asserted property): a put landing during the window (the
+    /// pin minted between the staging sweep and the executor sweep,
+    /// pin-before-publish) survives — the execution-time re-arbitration
+    /// cancels its candidate — while the unpinned twin deletes. The
+    /// cancelled candidate re-stages (and executes) only once its
+    /// protection drops.
+    #[tokio::test]
+    async fn durable_race_across_the_delay_put_during_window_survives() {
+        let site = temp_site("race-across-delay");
+        let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO)
+            .await
+            .expect("kv-only sqlite constructs");
+        store.register_liveness_source(Arc::new(SetSource::default()));
+
+        let pinned = key_for(b"the twin a put lands under during the window");
+        plant_kv(
+            &store,
+            &pinned,
+            b"the twin a put lands under during the window",
+        )
+        .await;
+        let unpinned = key_for(b"the unpinned twin");
+        plant_kv(&store, &unpinned, b"the unpinned twin").await;
+
+        let staged = store.sweep().await.expect("sweeps are never errors");
+        assert_eq!(staged.candidates_staged, 2);
+        assert_eq!(staged.deleted, 0);
+
+        {
+            let _pin = store.gc().acquire(&pinned).await;
+            assert_eq!(store.gc().pin_count(&pinned), 1, "pin-before-publish");
+
+            let executed = store.sweep().await.expect("sweeps are never errors");
+            assert_eq!(executed.aborted, None);
+            assert_eq!(
+                executed.deleted + executed.cancelled_by_arbitration,
+                2,
+                "execution-time re-arbitration ran over the taken set"
+            );
+            assert_eq!(executed.deleted, 1, "the unpinned twin deleted");
+            assert_eq!(
+                executed.cancelled_by_arbitration, 1,
+                "the pinned candidate cancelled"
+            );
+            assert!(
+                store.has(&pinned.as_bytes()).await.unwrap(),
+                "a put landing during the window survives the next sweep's execution"
+            );
+            assert!(!store.has(&unpinned.as_bytes()).await.unwrap());
+        }
+
+        // The pin dropped: the still-present candidate re-stages, and
+        // then (and only then) executes — its window row was consumed
+        // by the cancellation, not by a delete.
+        let restaged = store.sweep().await.expect("sweeps are never errors");
+        assert_eq!(
+            restaged.candidates_staged, 1,
+            "unpinned, the candidate re-stages"
+        );
+        assert_eq!(restaged.deleted, 0);
+
+        let executed = store.sweep().await.expect("sweeps are never errors");
+        assert_eq!(executed.deleted, 1, "the re-staged candidate executes");
+        assert!(!store.has(&pinned.as_bytes()).await.unwrap());
+    }
+
+    /// The vanished-candidate form: a staged candidate deleted during
+    /// the delay by a direct delete settles into
+    /// `cancelled_by_arbitration` at execution time — the vanished-
+    /// candidate accounting carries over (nothing to delete, counts
+    /// stay exact).
+    #[tokio::test]
+    async fn durable_vanished_candidate_settles_into_cancelled() {
+        let site = temp_site("vanished");
+        let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO)
+            .await
+            .expect("kv-only sqlite constructs");
+        store.register_liveness_source(Arc::new(SetSource::default()));
+
+        let victim = key_for(b"deleted during the delay");
+        plant_kv(&store, &victim, b"deleted during the delay").await;
+
+        let staged = store.sweep().await.expect("sweeps are never errors");
+        assert_eq!(staged.candidates_staged, 1);
+
+        store
+            .delete(&victim.as_bytes())
+            .await
+            .expect("the direct delete lands on the unprotected entry");
+
+        let executed = store.sweep().await.expect("sweeps are never errors");
+        assert_eq!(executed.aborted, None);
+        assert_eq!(executed.candidates_staged, 0);
+        assert_eq!(executed.deleted, 0);
+        assert_eq!(
+            executed.cancelled_by_arbitration, 1,
+            "the vanished candidate settles into the cancelled count"
+        );
+        assert!(window_drain(&store).await.is_empty());
+    }
+
+    /// The abort shapes on a sqlite store delete nothing, execute
+    /// nothing overdue, and stage nothing: zero sources
+    /// (`NoLivenessSources`), a failing protect callback
+    /// (`ProtectFailed`), and lease contention (`SweeperLock`).
+    #[tokio::test]
+    async fn durable_aborts_delete_nothing_and_execute_nothing() {
+        let site = temp_site("aborts-sources");
+        let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO)
+            .await
+            .expect("kv-only sqlite constructs");
+        let dead = key_for(b"dead while no sources are registered");
+        plant_kv(&store, &dead, b"dead while no sources are registered").await;
+
+        let aborted = store.sweep().await.expect("sweeps are never errors");
+        assert_eq!(aborted.aborted, Some(GcAbortCause::NoLivenessSources));
+        assert_eq!(aborted.candidates_staged, 0);
+        assert_eq!(aborted.deleted, 0);
+        assert!(store.has(&dead.as_bytes()).await.unwrap());
+        assert!(window_drain(&store).await.is_empty(), "nothing staged");
+    }
+
+    /// `ProtectFailed` aborts before the executor step on the durable
+    /// form: an overdue row staged by an earlier sweep is not taken
+    /// while the protection consult fails — abort precedes execution.
+    #[tokio::test]
+    async fn durable_protect_failed_aborts_before_executing_the_overdue_set() {
+        let site = temp_site("aborts-protect");
+        let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO)
+            .await
+            .expect("kv-only sqlite constructs");
+        store.register_liveness_source(Arc::new(SetSource::default()));
+        let protect = Arc::new(SetProtect::default());
+        store.register_protect_callback(protect.clone());
+
+        let staged_dead = key_for(b"staged dead while the consult answers");
+        plant_kv(
+            &store,
+            &staged_dead,
+            b"staged dead while the consult answers",
+        )
+        .await;
+        protect.set_keys(Vec::new());
+        let first = store.sweep().await.expect("sweeps are never errors");
+        assert_eq!(first.aborted, None);
+        assert_eq!(first.candidates_staged, 1);
+
+        protect.fail();
+        let later_dead = key_for(b"dead while the consult fails");
+        plant_kv(&store, &later_dead, b"dead while the consult fails").await;
+
+        let aborted = store.sweep().await.expect("sweeps are never errors");
+        assert_eq!(aborted.aborted, Some(GcAbortCause::ProtectFailed));
+        assert_eq!(aborted.candidates_staged, 0);
+        assert_eq!(aborted.deleted, 0);
+        assert_eq!(aborted.cancelled_by_arbitration, 0);
+        assert!(
+            store.has(&staged_dead.as_bytes()).await.unwrap(),
+            "the overdue candidate was not executed past the abort"
+        );
+        assert!(
+            store.has(&later_dead.as_bytes()).await.unwrap(),
+            "and nothing was deleted"
+        );
+        assert_eq!(
+            window_drain(&store).await,
+            vec![staged_dead],
+            "the overdue row still waits in the window"
+        );
+    }
+
+    /// Contention on the single-node lease aborts the second sweeper
+    /// as report data on the durable form too — nothing executed,
+    /// nothing staged, nothing deleted (`Ok(SweepReport)`, ADR-012 §1).
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn durable_lease_contention_aborts_the_second_sweeper_as_data() {
+        let site = temp_site("aborts-lease");
+        let store = Arc::new(
+            Store::kv_only_sqlite(&site.database, 64, Duration::ZERO)
+                .await
+                .expect("kv-only sqlite constructs"),
+        );
+        store.register_liveness_source(Arc::new(SetSource::default()));
+
+        let (entered_tx, entered_rx) = std::sync::mpsc::channel::<()>();
+        let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
+        store.register_protect_callback(Arc::new(LatchProtect {
+            entered: entered_tx,
+            release: parking_lot::Mutex::new(Some(release_rx)),
+        }));
+
+        let dead = key_for(b"dead under the contended sweeps");
+        plant_kv(&store, &dead, b"dead under the contended sweeps").await;
+
+        let first = {
+            let store = Arc::clone(&store);
+            tokio::spawn(async move { store.sweep().await })
+        };
+
+        while entered_rx
+            .recv_timeout(std::time::Duration::from_millis(500))
+            .is_err()
+        {
+            tokio::task::yield_now().await;
+        }
+
+        let second = store.sweep().await.expect("sweeps are never errors");
+        assert_eq!(second.aborted, Some(GcAbortCause::SweeperLock));
+        assert_eq!(second.deleted, 0);
+        assert_eq!(second.candidates_staged, 0, "nothing staged");
+        assert!(store.has(&dead.as_bytes()).await.unwrap());
+        assert!(window_drain(&store).await.is_empty(), "nothing executed");
+
+        release_tx.send(()).ok();
+        let first = first
+            .await
+            .expect("first sweep task joined")
+            .expect("first sweep returns a report");
+        assert_eq!(first.aborted, None);
+        assert_eq!(first.candidates_staged, 1, "the lease holder stages");
+        assert_eq!(first.deleted, 0, "and executes nothing at staging time");
+        assert!(
+            store.has(&dead.as_bytes()).await.unwrap(),
+            "the lease holder staged, did not execute"
+        );
+    }
+}

tasks/store-sweep-durable-window.md

@@ -1,7 +1,7 @@
 ---
 id: store-sweep-durable-window
 name: the sweep's durable-window form — staged candidates, next-sweep execution
-status: pending
+status: completed
 depends_on: [store-sqlite-construction]
 scope: moderate
 risk: medium
@@ -94,6 +94,71 @@ window, single-node:
 
 > To be filled by implementation agent
 
+The durable form's shape as it landed:
+
+- **The branch is the kv-engine enum match itself** (`sweep_window`'s
+  `match &self.kv`): `KvEngine::Sqlite(engine)` runs the durable form,
+  `KvEngine::Mem(_)` runs the inline form. The two durable helpers are
+  generic over `W: EngineWindowState` (the seam has `Backend`
+  associated types, so a `&dyn` handout is not object-safe — the
+  concrete `&SqliteBackend` passes directly). Wave 5's pg-lo slots one
+  more match arm in.
+- **The inline body was extracted verbatim** into
+  `sweep_window_inline` (the wave-2/all-features mem+local suites run
+  green unchanged); the protect consult stays in `sweep_window` above
+  the branch, shared by both forms.
+- **The protect-consult posture on overdue execution is recorded as a
+  doc point** on `execute_overdue_window`: the consult already ran for
+  the sweep (its abort, `ProtectFailed`, precedes execution) and its
+  coverage extends to the executor phase, so the overdue
+  re-arbitrations carry `consult_protect = false` — the inline form's
+  same posture.
+- **Malformed window rows are fatal** (`Error::KeyInvalid` via the
+  shared `typed_keys` boundary) — the pool-list discipline extended to
+  the window read.
+- **`visible_after` overflow is `Err(Error::Io)`** (checked clock add);
+  no panics on the library path.
+- **Report shape**: `candidates_staged` = staged this sweep;
+  `deleted`/`cancelled_by_arbitration` = the overdue set executed this
+  sweep (a vanished candidate settles into the cancelled count). The
+  `sweep()` doc states the exact-count invariants relative to each
+  form's schedule.
+
 ## Summary
 
-> To be filled on completion
+> To be filled on completion
+
+The sweep's durable SQL-backed form landed beside the inline form in
+`src/store.rs` (the first durable delete window, single-node; ADR-008
+§2 staged re-arbitrated delete, ADR-012 §5/§6.7 the-executor-is-the-
+next-sweep):
+
+- `sweep_window` branches on the delete window's host — the kv-engine
+  enum (`KvEngine::Sqlite` → durable; `KvEngine::Mem` → inline,
+  extracted verbatim). Under the lease: the protect consult (abort
+  `ProtectFailed` precedes any execution), then for the durable form
+  **execute overdue** — `window_take(now)` → execution-time
+  re-arbitration through `arbitrated_delete(key, false)` in
+  ~100-key batches (vanishes count as cancelled) — then **mark**
+  (unchanged), then **stage** at `visible_after = mark_end + window`
+  (the constructor-tunable delay; `Duration::ZERO` = execute-on-next-
+  sweep). Report counts exact relative to the schedule.
+- Abort shapes unchanged and unit-tested on sqlite:
+  `NoLivenessSources`/`ProtectFailed`/`SweeperLock` abort before
+  anything is executed or staged; the `ProtectFailed` test proves the
+  overdue set is left waiting in the window (abort precedes
+  `window_take`).
+- Unit tests (`sqlite_window_tests`): two-sweep exact counts at delay 0
+  (sweep 1 stages both tiers' dead entries without executing; sweep 2
+  executes exactly the staged set), future-delay staging (no execution
+  before the window closes; re-stage is idempotent), the race across
+  the delay (a put landing during the window survives via the
+  execution-time re-arbitration; the unpinned twin deletes; the
+  cancelled candidate re-stages and executes once unpinned), the
+  vanished-candidate form, and the three abort shapes.
+
+Verification: `cargo test` (69 lib + 5 integration, default), 124
+tests `--all-features`, 101 tests `--no-default-features --features
+mem`, `cargo clippy --all-targets -- -D warnings` (default and
+`--all-features`), and `cargo fmt --check` all green; the mem/local
+wave-2 suites pass unchanged (the inline form untouched).