alkgit-substrate 0.1.0

//! The GC maintenance chain's consumer side (wave 7 — substrate.md
//! §GC; ADR-019 §6; POC-R's validated cross-process run, POC-R's
//! `maintenance.rs`): the schedule registration + the runner, the
//! maintenance worker's claim loop, the leader TTL lock, and the sweep
//! dispatch — the chain that turns [`crate::reactor::MAINTENANCE_QUEUE`]'s
//! payload vocabulary ([`MaintenanceJob`]) into GC.
//!
//! ```text
//! MaintenanceChain::register_sweep_schedule (the `@every` schedule)
//!   → the alkstore leader-elected scheduler (run_schedules — one
//!     leader per fleet, the reserved `__alkstore_scheduler` lock)
//!   → firing lands in the maintenance queue (the `Sweep` payload)
//!   → worker claim (the queue's stamped visibility window is the
//!     discipline: the worker acks inside it — a Sweep dispatch longer
//!     than its stamped window simply lapses, and the at-least-once
//!     redelivery is held off by these same locks: the redelivered
//!     sweep is either a locked-out skip or an alkblobs `SweeperLock`
//!     abort, never a concurrent double reclaim)
//!   → named-TTL leader lock ([`MAINTENANCE_LOCK`], cross-process GC
//!        exclusion; a locked-out sweep is a skip + short retry, a
//!        value not an error)
//!   → alkblobs Store::sweep() -> Result<SweepReport, Error>
//!        (the report is the GC outcome carrier — stages, deletes,
//!        aborts; Err is substrate-level failure: a data verdict + a
//!        queue retry, the no-log-dep posture — the outcome is
//!        observable through the [`WorkerEvent`] seam and get_job)
//! ```
//!
//! Wave 6 landed the enqueue producer side only (the recorded
//! interpretation, gate-confirmed); this module is the consumer side —
//! and, the wave-6 gate's N-b rider, the **runner supervision** the
//! chain owns:
//!
//! - the **wake runner** ([`Reactor::run_wake_runner`]) returns
//!   `Ok(())` on both `recv_timeout` `Err` arms (the alkstore
//!   contract's `Closed` source-death and `Database` storage-failure
//!   arms collapsed by design), so until supervised a runner stop
//!   means wake-driven refresh stops silently — [`MaintenanceChain`]'s
//!   supervision restarts it on **both** stop shapes (recovery is a
//!   fresh `listen`, [`Reactor::run_wake_runner`]'s posture; the
//!   stale window between stop and restart is exactly what the
//!   reconcile tick exists to bound).
//! - the **reconcile tick's driving mechanism** (the wave-6 recorded
//!   interpretation's pinned answer): the chain itself ticks it —
//!   [`MaintenanceOpts::reconcile_tick`], one
//!   [`Reactor::reconcile`] pass per tick alongside the worker loop.
//!   (Of the three candidates — a schedule firing, a worker-loop
//!   tick, a dedicated job — the chain tick is the shape the
//!   supervision already owns: schedule firings land *GC* jobs into
//!   the queue, and a reconcile job variant would be a second
//!   vocabulary member for what the runner owns anyway.)
//! - the **scheduler runner** supervision: `run_schedules`
//!   respawns on `Err(LeadershipLost)` (the alkstore recipe) and
//!   propagates any other stop — a substrate-level failure of the
//!   scheduler leg is the chain's failure (the closed-store posture:
//!   respawning against a dead store is misdirected).
//!
//! # The pack-housekeeping consumer (pinned in-task)
//!
//! The `PackHousekeeping` job's consumer action is the through-store
//! pack cache's release path ([`PackCache::release_all`]) — the
//! released packs' pins drop, and the next sweep reclaims their pack
//! blobs (substrate.md §object storage). The whole cache releases:
//! the request-digest cache is process-local, per-repo scoping is a
//! non-goal, and the enqueue leg's repo id is observation data only. An
//! assembly without a through-store [`PackGen`](crate::packgen::PackGen)
//! carries no cache ([`MaintenanceChain::new`]'s `packs: None`) — such
//! a job acks with nothing to release, the composition-honest shape.
//!
//! # The sweep verdicts (data, never errors)
//!
//! - a completed report (`SweepReport.aborted` included — an aborted
//!   sweep is still a ran sweep, its result carried) — acked;
//! - a locked-out sweep ([`MAINTENANCE_LOCK`] held elsewhere —
//!   cross-process GC exclusion is the point) — a skip + short retry,
//!   a value: the job goes back to pending with
//!   [`LOCKED_OUT_RETRY_SECS`];
//! - a substrate-level sweep failure (alkblobs `Err`) — a data
//!   verdict + a queue retry (the at-least-once curve runs it again;
//!   max attempts dead-letters it, observable via get_job).
//!
//! A poison payload (a foreign producer on the queue, codec drift)
//! dead-letters immediately (`fail`) — the stream consumer's poison
//! posture, queue-shaped: never a wedge, never a retry loop.

use std::sync::Arc;
use std::time::Duration;

use alkstore::{JobHandle, Queue, QueueOpts, Schedule, ScheduleOpts, StopToken, Store};

use alkblobs::Store as BlobStore;
use serde_json::json;

use crate::packgen::PackCache;
use crate::reactor::{
    AppliedWatermark, MaintenanceJob, Reactor, MAINTENANCE_LOCK, MAINTENANCE_QUEUE,
};

/// The sweep schedule's name — registered (upsert by name) firing
/// into [`MAINTENANCE_QUEUE`] with the `MaintenanceJob::Sweep` payload
/// (the POC-R-validated schedule; the chain's runner supervision owns
/// the `run_schedules` side).
pub const SWEEP_SCHEDULE: &str = "alkgit-sweep-schedule";

/// Seconds a locked-out sweep goes back to pending for (the short
/// retry — a value, not an error; ADR-019 §6; the schedule's own
/// cadence is the backstop after it).
pub const LOCKED_OUT_RETRY_SECS: i64 = 2;

/// The stop loop's granularity for the chain's stop-responsive arms
/// (the reactor's posture: responsiveness, not a wake source).
const STOP_GRACE_MS: u64 = 50;

/// The chain's tuning datums (assembly-level knobs, not structure —
/// the schedule spec is the sweep cadence; the tick is the postgres
/// staleness bound; the rest is the loop granularity).
///
/// The defaults: a 30-second sweep cadence (`@every 30s`), a
/// 30-second reconcile tick, a 500 ms worker poll, a 1 s wake-restart
/// backoff, and a 30 s leader-lock TTL. Every value exists for the
/// tests' short forms too — pass a `MaintenanceOpts` with the fields
/// set.
#[derive(Clone, Debug)]
pub struct MaintenanceOpts {
    /// The `@every` spec the sweep schedule registers with
    /// (`schedule()`'s v1 grammar).
    pub sweep_spec: String,
    /// The reconcile tick's cadence — one [`Reactor::reconcile`] pass
    /// per tick (the pinned driving mechanism; on postgres this tick
    /// is the cache staleness's real bound after a lost hint).
    pub reconcile_tick: Duration,
    /// The worker claim loop's poll cadence.
    pub worker_poll: Duration,
    /// The supervised wake runner's restart backoff (both stop
    /// shapes; a fresh `listen` after the sleep).
    pub wake_retry_backoff: Duration,
    /// The leader lock's TTL seconds (`try_lock` — held across the
    /// sweep; a sweep that outgrows the TTL lapses it silently and the
    /// sweep's own `SweeperLock` lease still serializes the stores).
    pub lock_ttl_secs: i64,
}

impl Default for MaintenanceOpts {
    fn default() -> Self {
        MaintenanceOpts {
            sweep_spec: "@every 30s".to_owned(),
            reconcile_tick: Duration::from_secs(30),
            worker_poll: Duration::from_millis(500),
            wake_retry_backoff: Duration::from_secs(1),
            lock_ttl_secs: 30,
        }
    }
}

/// What the maintenance worker did — the observation seam (the
/// logged-as-data posture: the outcome carriers surface here, no
/// tracing dependency exists anywhere). A consumer of
/// [`MaintenanceChain::run`] sees every executed or skipped job.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum WorkerEvent {
    /// A `MaintenanceJob::Sweep` job executed — the verdict below.
    Sweep(SweepVerdict),
    /// A `MaintenanceJob::PackHousekeeping` job executed — the
    /// released count ([`PackCache::release_all`]).
    PackHousekeeping(usize),
}

/// A sweep job's outcome, as data (substrate.md §GC — the chain never
/// turns a sweep outcome into an error).
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum SweepVerdict {
    /// The store ran the sweep and reported — the [`alkblobs::SweepReport`]
    /// carries stages, deletes, aborts (an aborted sweep is a ran
    /// sweep whose cause is carried; the report is the outcome
    /// carrier).
    Report(alkblobs::SweepReport),
    /// The leader lock was held elsewhere: a skip + short retry — a
    /// value, not an error (the cross-process exclusion doing its
    /// job; the job goes back to pending for
    /// [`LOCKED_OUT_RETRY_SECS`]).
    LockedOut,
    /// A substrate-level sweep failure (the store's `Err`): the
    /// string is the error's display for the data carrier; the job
    /// retries on the queue's curve (at-least-once; the attempt
    /// budget dead-letters it — observable via get_job).
    Failed(String),
}

/// The maintenance chain's consumer side — the schedule registration +
/// the runner, the worker's claim loop, the leader TTL lock, the sweep
/// dispatch, and the supervised halves the wave-6 gate booked the
/// chain to own (the wake runner's restart on both stop shapes; the
/// reconcile tick's driving).
///
/// Contract stamp: substrate.md §GC; ADR-019 §6.
///
/// Constructed from the assembly's pieces: the engine handle (the
/// same one the reactor rides), the alkblobs store (the
/// `sweep()` leg's target — the liveness source itself is
/// registered at the composition construction, e.g.
/// [`crate::liveness::RefClosure::register`]; **a sweep dispatched
/// without one aborts with `NoLivenessSources`** — the report carries
/// it), the reactor + the applied watermark (the supervision +
/// reconcile legs), the shared through-store pack cache when the
/// assembly serves packs through-store, and the tuning datums.
///
/// The wave-8 single composition entry unifies the construction;
/// this is the per-piece constructor (the recorded interpretation —
/// `substrate/w6-scaffold-manifest`'s shape).
pub struct MaintenanceChain {
    store: Arc<dyn Store>,
    blobs: Arc<BlobStore>,
    reactor: Arc<Reactor>,
    applied: Arc<AppliedWatermark>,
    packs: Option<Arc<PackCache>>,
    opts: MaintenanceOpts,
}

impl MaintenanceChain {
    /// The chain over the assembly's pieces. `packs` is the
    /// through-store pack cache when the assembly serves packs
    /// through-store (`None` on the direct shape — the
    /// pack-housekeeping job then acks with nothing to release).
    pub fn new(
        store: Arc<dyn Store>,
        blobs: Arc<BlobStore>,
        reactor: Arc<Reactor>,
        applied: Arc<AppliedWatermark>,
        packs: Option<Arc<PackCache>>,
        opts: MaintenanceOpts,
    ) -> Self {
        MaintenanceChain {
            store,
            blobs,
            reactor,
            applied,
            packs,
            opts,
        }
    }

    /// The engine handle (the schedule, the queue and the lock ride
    /// it — the same handle the reactor rides).
    pub fn store(&self) -> &Arc<dyn Store> {
        &self.store
    }

    /// The alkblobs store (the sweep dispatch's target).
    pub fn blobs(&self) -> &Arc<BlobStore> {
        &self.blobs
    }

    /// The chain's tuning datums.
    pub fn opts(&self) -> &MaintenanceOpts {
        &self.opts
    }

    /// **The schedule registration** — upsert [`SWEEP_SCHEDULE`]
    /// firing into [`MAINTENANCE_QUEUE`] with the
    /// `MaintenanceJob::Sweep` payload (the worker's dispatch
    /// vocabulary). Idempotent by upsert; the spec is
    /// [`schedule()`](alkstore::Store::schedule)'s v1 grammar.
    pub async fn register_sweep_schedule(&self, spec: &str) -> alkstore::Result<Schedule> {
        let payload = json!(MaintenanceJob::Sweep);
        self.store
            .schedule(
                SWEEP_SCHEDULE,
                spec,
                MAINTENANCE_QUEUE,
                payload,
                ScheduleOpts::default(),
            )
            .await
    }

    /// **The chain's run** — registration, the supervised legs, the
    /// reconciliation tick, and the worker's claim loop, all under
    /// one stop token:
    ///
    /// - the **sweep schedule** is upserted first (loud — a chain
    ///   without its schedule never fires);
    /// - the **scheduler runner** runs under supervision — respawned
    ///   on `Err(LeadershipLost)`; any other stop (the alkstore
    ///   contract's closed-store/storage-failure posture) is
    ///   propagated: the chain returns it. The legs wind down via the
    ///   chain's own internal token — cancelled on every return (the
    ///   wave-7 gate's O-b disposition: the wake leg's restart loop
    ///   exits on it without waiting out the caller's token; the
    ///   caller's stop token is never cancelled by the chain).
    /// - the **wake runner** is supervised — restarted on both stop
    ///   shapes (the N-b rider: the runner's `Ok(())` on a
    ///   `recv_timeout`/`listen` failure is the collapsed stop; the
    ///   restart's backoff is
    ///   [`MaintenanceOpts::wake_retry_backoff`], the recovery a
    ///   fresh `listen`);
    /// - the **reconcile tick** drives one [`Reactor::reconcile`]
    ///   pass every [`MaintenanceOpts::reconcile_tick`] (fail-soft —
    ///   a failed pass leaves the previous snapshot serving);
    /// - the **worker claim loop** drains [`MAINTENANCE_QUEUE`] on
    ///   every poll tick (claim until no work, dispatch on the
    ///   [`MaintenanceJob`] vocabulary, observation through
    ///   `on_event`).
    ///
    /// Returns `Ok(())` on a cancelled stop token (legs joined) and
    /// `Err(alkstore::Error)` on a scheduler-leg substrate failure.
    pub async fn run(
        &self,
        worker_id: &str,
        stop: StopToken,
        mut on_event: impl FnMut(WorkerEvent) + Send,
    ) -> alkstore::Result<()> {
        self.register_sweep_schedule(&self.opts.sweep_spec).await?;
        let queue = self
            .store
            .queue(MAINTENANCE_QUEUE, QueueOpts::default())
            .await?;

        let chain_stop = StopToken::new();

        // The scheduler leg reports through a channel — the leg's
        // outcome (and a panic's silent close) reaches the run loop
        // without ever double-polling a completed task handle (the
        // O-b-era shape's latent bug the probe caught).
        let (sched_done, sched_rx) = tokio::sync::oneshot::channel();
        let sched_store = self.store.clone();
        let sched_stop = stop.clone();
        let handler = tokio::spawn(async move {
            let outcome = loop {
                match sched_store.run_schedules(sched_stop.clone()).await {
                    Err(alkstore::Error::LeadershipLost) => continue,
                    outcome => break outcome,
                }
            };
            let _ = sched_done.send(outcome);
        });

        let wake_reactor = self.reactor.clone();
        let wake_stop = stop.clone();
        let wake_chain_stop = chain_stop.clone();
        let backoff = self.opts.wake_retry_backoff;
        let wake = tokio::spawn(async move {
            loop {
                if wake_stop.is_cancelled() || wake_chain_stop.is_cancelled() {
                    return;
                }
                let runner = wake_reactor.run_wake_runner(wake_stop.clone());
                tokio::pin!(runner);
                tokio::select! {
                    _ = &mut runner => {}
                    _ = stop_grace(&wake_chain_stop) => return,
                }
                if wake_stop.is_cancelled() || wake_chain_stop.is_cancelled() {
                    return;
                }
                tokio::time::sleep(backoff).await;
            }
        });

        let mut poll = tokio::time::interval(self.opts.worker_poll);
        poll.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        let mut rec_tick = tokio::time::interval(self.opts.reconcile_tick);
        rec_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        tokio::pin!(sched_rx);

        let outcome = loop {
            tokio::select! {
                _ = stop_grace(&stop) => break Ok(()),
                sched_result = &mut sched_rx => break match sched_result {
                    Ok(result) => result,
                    Err(_) => Err(alkstore::Error::database(
                        "the scheduler runner task failed without reporting".to_owned(),
                    )),
                },
                _ = rec_tick.tick() => {
                    let _ = self.reactor.reconcile(&self.applied).await;
                }
                _ = poll.tick() => {
                    let _ = self.drain(&*queue, worker_id, &stop, &mut on_event).await;
                }
            }
        };

        // The O-b rider's wired disposition: the chain's internal token
        // is the legs' wind-down — cancelled on every chain return (the
        // caller's stop, the scheduler leg's substrate failure — the
        // wake leg's runner is cut and the restart loop exits without
        // waiting out the caller's own token, and the join's window
        // stays a safety net, not the mechanism). Nothing cancels it
        // while the loop runs, so the arm it used to own is dropped,
        // not kept dead.
        chain_stop.cancel();
        join_legs(handler, wake).await;
        outcome
    }

    /// One poll tick's claim loop: claim until no work (or a stop),
    /// dispatching each job on the [`MaintenanceJob`] vocabulary.
    /// Claim failures are transient (a busy or closing storage) — the
    /// next poll tick retries; the queue's own visibility discipline
    /// governs re-delivery. `Err` ends the tick's drain; the chain
    /// keeps running (its scheduler leg owns the substrate-death
    /// detection).
    async fn drain(
        &self,
        queue: &dyn Queue,
        worker_id: &str,
        stop: &StopToken,
        on_event: &mut (impl FnMut(WorkerEvent) + Send),
    ) -> alkstore::Result<()> {
        loop {
            if stop.is_cancelled() {
                return Ok(());
            }
            let Some(handle) = queue.claim_one(worker_id).await? else {
                return Ok(());
            };
            self.dispatch(handle, worker_id, stop, on_event).await;
        }
    }

    /// Dispatch one claimed job on the [`MaintenanceJob`] vocabulary
    /// (substrate.md §GC — the wave-7 pinned actions). A poison
    /// payload (undecodable) dead-letters; `Sweep` rides the leader
    /// lock + the store's sweep; `PackHousekeeping` rides the pack
    /// cache's release path.
    async fn dispatch(
        &self,
        handle: Box<dyn JobHandle>,
        worker_id: &str,
        stop: &StopToken,
        on_event: &mut (impl FnMut(WorkerEvent) + Send),
    ) {
        match handle.job().payload_as::<MaintenanceJob>() {
            Err(_) => {
                let _ = handle
                    .fail(Some("unparseable maintenance payload".to_owned()))
                    .await;
            }
            Ok(MaintenanceJob::PackHousekeeping { .. }) => {
                let released = self
                    .packs
                    .as_ref()
                    .map(|packs| packs.release_all())
                    .unwrap_or(0);
                let _ = handle.ack().await;
                on_event(WorkerEvent::PackHousekeeping(released));
            }
            Ok(MaintenanceJob::Sweep) => {
                match self
                    .store
                    .try_lock(MAINTENANCE_LOCK, worker_id, self.opts.lock_ttl_secs)
                    .await
                {
                    Ok(Some(lease)) => {
                        let verdict = self.sweep_under_lock(stop).await;
                        let _ = lease.release().await;
                        match &verdict {
                            None => {}
                            Some(SweepVerdict::Report(_)) => {
                                let _ = handle.ack().await;
                            }
                            Some(SweepVerdict::Failed(reason)) => {
                                let _ = handle.retry(Some(reason.clone()), None).await;
                            }
                            Some(SweepVerdict::LockedOut) => {
                                let _ = handle
                                    .retry(
                                        Some("the sweeper lock is held elsewhere".to_owned()),
                                        Some(LOCKED_OUT_RETRY_SECS),
                                    )
                                    .await;
                            }
                        }
                        if let Some(verdict) = verdict {
                            on_event(WorkerEvent::Sweep(verdict));
                        }
                    }
                    Ok(None) => {
                        let _ = handle
                            .retry(
                                Some("the sweeper lock is held elsewhere".to_owned()),
                                Some(LOCKED_OUT_RETRY_SECS),
                            )
                            .await;
                        on_event(WorkerEvent::Sweep(SweepVerdict::LockedOut));
                    }
                    Err(err) => {
                        let _ = handle.retry(Some(err.to_string()), None).await;
                        on_event(WorkerEvent::Sweep(SweepVerdict::Failed(err.to_string())));
                    }
                }
            }
        }
    }

    /// The sweep dispatch under the leader lease: the store's sweep.
    /// The stop token cuts a *stopped* chain out of a long sweep (the
    /// dropped sweep is interruption-safe by design — its candidate
    /// windows are durable and its claim's at-least-once redelivery
    /// governs the job). `None` = the chain is stopping — no verdict,
    /// no job op.
    async fn sweep_under_lock(&self, stop: &StopToken) -> Option<SweepVerdict> {
        let sweep = self.blobs.sweep();
        tokio::pin!(sweep);
        tokio::select! {
            report = &mut sweep => {
                Some(match report {
                    Ok(report) => SweepVerdict::Report(report),
                    Err(err) => SweepVerdict::Failed(err.to_string()),
                })
            }
            _ = stop_grace(stop) => None,
        }
    }
}

/// The stop-responsive wait arm ([`crate::reactor`]'s posture): a
/// canceled token returns promptly instead of at the interval's
/// cadence.
async fn stop_grace(stop: &StopToken) {
    let mut tick = tokio::time::interval(Duration::from_millis(STOP_GRACE_MS));
    loop {
        tick.tick().await;
        if stop.is_cancelled() {
            return;
        }
    }
}
/// Wind the supervised legs down: both legs exit on their cancelled
/// tokens within a bounded window (the wake leg's sleep, the
/// scheduler's tick); join within the bound, abort past it.
async fn join_legs(handler: tokio::task::JoinHandle<()>, wake: tokio::task::JoinHandle<()>) {
    let handler_abort = handler.abort_handle();
    let wake_abort = wake.abort_handle();
    let joins = async move {
        let _ = handler.await;
        let _ = wake.await;
    };
    if tokio::time::timeout(Duration::from_millis(STOP_GRACE_MS * 20), joins)
        .await
        .is_err()
    {
        handler_abort.abort();
        wake_abort.abort();
    }
}

#[cfg(test)]
mod tests {
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::sync::{Arc, Mutex};
    use std::time::Duration;

    use alkstore::{JobState, QueueOpts, StopToken, Store, WakeReceiver};
    use serde_json::Value;

    use alkblobs::Store as BlobStore;
    #[cfg(feature = "sqlite")]
    use alkgit::GitRegistryStore as _;
    use alkgit::{GitPackGen as _, GitRefs, Limits, RefUpdate};
    use gix_hash::ObjectId;
    use gix_object::Kind;

    use crate::odb::{key_to_oid, ObjectFront};
    use crate::packgen::{PackCache, PackGen};
    use crate::reactor::{AppliedWatermark, MaintenanceJob, Reactor, RegistryEvent};
    use crate::refs::InMemoryRefTable;
    use crate::registry::{
        CachedRegistryProvider, InMemoryRegistry, RegistryRead as _, RegistryWrite,
    };
    #[cfg(feature = "sqlite")]
    use crate::testutil::scratch_dir;
    use crate::testutil::{record, wait_until};

    use super::{
        MaintenanceChain, MaintenanceOpts, SweepVerdict, WorkerEvent, MAINTENANCE_LOCK,
        MAINTENANCE_QUEUE,
    };

    type Seen = Arc<Mutex<Vec<WorkerEvent>>>;

    fn saw_sweep_verdict(seen: &Seen, pred: impl Fn(&SweepVerdict) -> bool) -> bool {
        seen.lock()
            .unwrap()
            .iter()
            .any(|event| matches!(event, WorkerEvent::Sweep(v) if pred(v)))
    }

    /// The mem-engine fixture: the alkstore mem engine + the alkblobs
    /// mem store + the in-memory registry behind the cached provider —
    /// the chain over the completeness assembly.
    struct MemFixture {
        store: Arc<dyn Store>,
        blobs: Arc<BlobStore>,
        reactor: Arc<Reactor>,
        registry: Arc<InMemoryRegistry>,
        provider: Arc<CachedRegistryProvider>,
        applied: Arc<AppliedWatermark>,
        chain: Arc<MaintenanceChain>,
    }

    fn mem_fixture(opts: MaintenanceOpts, packs: Option<Arc<PackCache>>) -> MemFixture {
        let store: Arc<dyn Store> = Arc::new(alkstore_mem::MemStore::new());
        let blobs = Arc::new(BlobStore::mem_only());
        let registry = Arc::new(InMemoryRegistry::new());
        let provider = Arc::new(CachedRegistryProvider::new());
        let reactor = Arc::new(Reactor::new(
            store.clone(),
            registry.clone() as Arc<dyn RegistryWrite>,
            provider.clone(),
        ));
        let applied = AppliedWatermark::new();
        let chain = Arc::new(MaintenanceChain::new(
            store.clone(),
            blobs.clone(),
            reactor.clone(),
            applied.clone(),
            packs,
            opts,
        ));
        MemFixture {
            store,
            blobs,
            reactor,
            registry,
            provider,
            applied,
            chain,
        }
    }

    /// The liveness wiring over the fixture's blob store: the
    /// `RefClosure` registered (the sweep dispatch's requirement — a
    /// sweep without one aborts `NoLivenessSources`) with an in-memory
    /// ref table the tests write through.
    struct WiredLiveness {
        front: ObjectFront,
        refs: Arc<InMemoryRefTable>,
        _source: Arc<crate::liveness::RefClosure>,
    }

    fn wire_liveness(fixture: &MemFixture) -> WiredLiveness {
        let front = ObjectFront::new(fixture.blobs.clone());
        let refs = Arc::new(InMemoryRefTable::new());
        let writes = Arc::new(AtomicU64::new(0));
        let source = crate::liveness::RefClosure::new(
            front.clone(),
            refs.clone() as Arc<dyn crate::liveness::RefRows>,
            writes,
        );
        source.register(fixture.blobs.as_ref()).unwrap();
        WiredLiveness {
            front,
            refs,
            _source: source,
        }
    }

    async fn put_object(front: &ObjectFront, kind: Kind, content: &[u8]) -> ObjectId {
        let (key, _pin) = front.put(kind, content).await.unwrap();
        key_to_oid(&key).expect("the put's key is the family key")
    }

    async fn create_ref<T: GitRefs + ?Sized>(refs: &T, id: &str, name: &str, oid: ObjectId) {
        let outcomes = GitRefs::apply_updates(
            refs,
            &record(id),
            vec![RefUpdate {
                name: name.to_owned(),
                expected: None,
                new: Some(oid),
            }],
            false,
        )
        .await
        .unwrap();
        assert!(outcomes.iter().all(|o| o.result.is_ok()));
    }

    /// Fast the loops, slow the schedule — the deterministic-enqueue
    /// tests' opts (the firing cadence rides its own test).
    fn quiet_opts() -> MaintenanceOpts {
        MaintenanceOpts {
            sweep_spec: "@every 600s".to_owned(),
            reconcile_tick: Duration::from_secs(600),
            worker_poll: Duration::from_millis(25),
            wake_retry_backoff: Duration::from_millis(50),
            lock_ttl_secs: 30,
        }
    }

    fn spawn_run(
        chain: Arc<MaintenanceChain>,
        stop: StopToken,
        seen: Seen,
    ) -> tokio::task::JoinHandle<alkstore::Result<()>> {
        tokio::spawn(async move {
            chain
                .run("worker", stop, move |event| {
                    seen.lock().unwrap().push(event)
                })
                .await
        })
    }

    async fn stop_and_join(stop: StopToken, join: tokio::task::JoinHandle<alkstore::Result<()>>) {
        stop.cancel();
        let outcome = tokio::time::timeout(Duration::from_secs(5), join)
            .await
            .expect("the run ended inside the deadline")
            .expect("the run ended clean");
        if let Err(err) = outcome {
            panic!("the run ended with an error: {err}");
        }
    }

    /// Poll an object away (the async predicate's poll loop — a sync
    /// `wait_until` closure cannot await the front's reads).
    async fn wait_object_gone(front: &ObjectFront, oid: &ObjectId) {
        let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
        let mut gone = false;
        while tokio::time::Instant::now() < deadline {
            gone = front.stat(oid).await.unwrap().is_none();
            if gone {
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        assert!(gone, "the object did not reclaim inside the deadline");
    }

    /// The full scheduled sweep chain on the mem assembly: the
    /// registration + the scheduler runner fire the `Sweep` payload
    /// into the maintenance queue, the worker claims it, takes the
    /// leader lock, and dispatches the store's sweep — the orphan is
    /// reclaimed (the registered ref closure decides liveness), the
    /// covered closure survives, the job acked away.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn the_schedule_fires_and_the_worker_sweeps() {
        let fixture = mem_fixture(
            MaintenanceOpts {
                sweep_spec: "@every 1s".to_owned(),
                ..quiet_opts()
            },
            None,
        );
        let wired = wire_liveness(&fixture);

        let covered = put_object(&wired.front, Kind::Blob, b"covered\n").await;
        let orphan = put_object(&wired.front, Kind::Blob, b"orphan\n").await;
        create_ref(&*wired.refs, "repo", "refs/heads/main", covered).await;

        let seen: Seen = Arc::new(Mutex::new(Vec::new()));
        let stop = StopToken::new();
        let runner = spawn_run(fixture.chain.clone(), stop.clone(), seen.clone());

        wait_until(|| {
            saw_sweep_verdict(&seen, |verdict| {
                matches!(
                    verdict,
                    SweepVerdict::Report(report) if report.aborted.is_none()
                )
            })
        })
        .await;
        wait_object_gone(&wired.front, &orphan).await;
        assert!(
            wired.front.stat(&covered).await.unwrap().is_some(),
            "the covered closure survives the chain's sweep"
        );

        stop_and_join(stop, runner).await;
    }

    /// A locked-out sweep is a skip + short retry — a value, not an
    /// error: with theMaintenanceLock held by an outsider, the
    /// worker's sweep verdict is LockedOut, the job returns to
    /// pending (a live row with no error), and once the outsider
    /// releases, the retried job sweeps for real and acks away.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_locked_out_sweep_is_a_skip_and_short_retry() {
        let fixture = mem_fixture(quiet_opts(), None);
        let wired = wire_liveness(&fixture);

        let covered = put_object(&wired.front, Kind::Blob, b"covered\n").await;
        let orphan = put_object(&wired.front, Kind::Blob, b"orphan\n").await;
        create_ref(&*wired.refs, "repo", "refs/heads/main", covered).await;

        let lease = fixture
            .store
            .try_lock(MAINTENANCE_LOCK, "an-external-holder", 30)
            .await
            .unwrap()
            .expect("the outsider holds the lock");

        let queue = fixture
            .store
            .queue(MAINTENANCE_QUEUE, QueueOpts::default())
            .await
            .unwrap();
        let sweep_id = fixture
            .reactor
            .enqueue(&MaintenanceJob::Sweep)
            .await
            .unwrap();

        let seen: Seen = Arc::new(Mutex::new(Vec::new()));
        let stop = StopToken::new();
        let runner = spawn_run(fixture.chain.clone(), stop.clone(), seen.clone());

        wait_until(|| saw_sweep_verdict(&seen, |v| matches!(v, SweepVerdict::LockedOut))).await;
        let row = queue.get_job(sweep_id).await.unwrap().expect("a live row");
        assert_eq!(
            row.state,
            JobState::Pending,
            "the locked-out sweep is a retry — back in pending for the short delay"
        );
        assert_eq!(
            row.last_error, None,
            "a retried (live) row carries no error the locked-out skip surfaces"
        );

        lease.release().await.unwrap();
        wait_until(|| {
            saw_sweep_verdict(&seen, |v| {
                matches!(
                    v,
                    SweepVerdict::Report(report) if report.aborted.is_none()
                )
            })
        })
        .await;
        wait_object_gone(&wired.front, &orphan).await;
        assert!(
            queue.get_job(sweep_id).await.unwrap().is_none(),
            "the sweep job acked away"
        );

        stop_and_join(stop, runner).await;
    }

    /// The N-b supervision, the `Closed` shape: the wake runner's
    /// source dies (the sqlite watcher-death class, probe-pinned with
    /// an immediately-closing receiver), `run_wake_runner` returns
    /// quietly — and the chain restarts it (a fresh `listen`), so the
    /// wake-driven refresh lands and the silent-stop window is closed.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn the_wake_runner_is_supervised_through_a_closed_source() {
        wake_supervision_shape(FaultyListenStore::closed_once(Box::new(
            alkstore_mem::MemStore::new(),
        )))
        .await;
    }

    /// The N-b supervision, the `Database` shape: the wake runner's
    /// `listen` fails substrate-level (the storage-failure arm) — the
    /// chain restarts it and the refresh lands.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn the_wake_runner_is_supervised_through_a_listen_failure() {
        wake_supervision_shape(FaultyListenStore::database_once(Box::new(
            alkstore_mem::MemStore::new(),
        )))
        .await;
    }

    async fn wake_supervision_shape(store: Arc<FaultyListenStore>) {
        let blobs = Arc::new(BlobStore::mem_only());
        let registry = Arc::new(InMemoryRegistry::new());
        let provider = Arc::new(CachedRegistryProvider::new());
        let reactor = Arc::new(Reactor::new(
            store.clone() as Arc<dyn Store>,
            registry.clone() as Arc<dyn RegistryWrite>,
            provider.clone(),
        ));
        let chain = Arc::new(MaintenanceChain::new(
            store.clone() as Arc<dyn Store>,
            blobs,
            reactor.clone(),
            AppliedWatermark::new(),
            None,
            quiet_opts(),
        ));

        let seen: Seen = Arc::new(Mutex::new(Vec::new()));
        let stop = StopToken::new();
        let runner = spawn_run(chain, stop.clone(), seen.clone());
        tokio::time::sleep(Duration::from_millis(300)).await;

        let seq = registry.put(&record("owner/wake")).await.unwrap();
        reactor
            .announce(&RegistryEvent::Put {
                seq,
                record: record("owner/wake"),
            })
            .await;

        wait_until(|| provider.resolve("owner/wake").is_some()).await;
        assert_eq!(
            provider.resolve("owner/wake"),
            Some(record("owner/wake")),
            "the wake-driven refresh landed after the supervised restart"
        );
        let listens = store.listens.load(Ordering::Relaxed);
        assert!(
            listens >= 2,
            "the runner restarted — a fresh `listen` re-attached (listens: {listens})"
        );

        stop_and_join(stop, runner).await;
    }

    /// The reconcile tick's driving (the pinned mechanism): a durable
    /// registry write with no announce at all leaves the provider
    /// stale, and the chain's tick closes the window (the
    /// postgres-lost-hint posture's backstop) and advances the applied
    /// watermark.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn the_reconcile_tick_drives_the_stale_provider() {
        let fixture = mem_fixture(
            MaintenanceOpts {
                reconcile_tick: Duration::from_millis(100),
                ..quiet_opts()
            },
            None,
        );

        fixture.registry.put(&record("owner/tick")).await.unwrap();
        assert_eq!(
            fixture.provider.resolve("owner/tick"),
            None,
            "no announce: the cache is stale until the tick"
        );

        let seen: Seen = Arc::new(Mutex::new(Vec::new()));
        let stop = StopToken::new();
        let runner = spawn_run(fixture.chain.clone(), stop.clone(), seen);
        wait_until(|| fixture.provider.resolve("owner/tick").is_some()).await;
        assert!(
            fixture.applied.get() > 0,
            "the tick also advanced the applied watermark"
        );
        stop_and_join(stop, runner).await;
    }

    /// The `PackHousekeeping` consumer action (pinned in-task): the
    /// through-store pack cache's release path — the job drains the
    /// cache (the generated pack's pin released), acks away, and the
    /// next sweep can reclaim the pack blob.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn pack_housekeeping_runs_the_release_path() {
        let cache = Arc::new(PackCache::new(2));
        let fixture = mem_fixture(quiet_opts(), Some(cache.clone()));
        let front = ObjectFront::new(fixture.blobs.clone());
        let blob = put_object(&front, Kind::Blob, b"pack me\n").await;
        let tree = put_object(&front, Kind::Tree, &tree_bytes(&[("b", &blob)])).await;
        let commit = put_object(&front, Kind::Commit, &commit_bytes(&tree, &[])).await;

        let packgen = PackGen::through_store_with(front.clone(), cache.clone());
        let sink = VecSink::default();
        packgen
            .generate(
                &record("repo"),
                vec![commit],
                vec![],
                Limits::default(),
                Box::new(sink.clone()),
            )
            .await
            .unwrap();
        assert!(
            sink.bytes().starts_with(b"PACK"),
            "generation produced a pack (the cache holds it)"
        );

        fixture
            .reactor
            .enqueue(&MaintenanceJob::PackHousekeeping {
                repo_id: "repo".to_owned(),
            })
            .await
            .unwrap();

        let seen: Seen = Arc::new(Mutex::new(Vec::new()));
        let stop = StopToken::new();
        let runner = spawn_run(fixture.chain.clone(), stop.clone(), seen.clone());
        wait_until(|| {
            seen.lock()
                .unwrap()
                .iter()
                .any(|event| matches!(event, WorkerEvent::PackHousekeeping(1)))
        })
        .await;
        assert_eq!(
            cache.release_all(),
            0,
            "the job's release path drained the cache"
        );
        stop_and_join(stop, runner).await;
    }

    /// The repo-deletion-is-GC proof through the full chain (the
    /// acceptance line): removing a repo's registry + ref rows (the
    /// removal's composition semantics) uncovers its objects, and the
    /// chain's sweep dispatch reclaims them — the scheduled GC, not a
    /// bulk delete.
    #[cfg(feature = "sqlite")]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn the_chain_reclaims_a_deleted_repos_objects() {
        use crate::liveness::{RefClosure, RefRows};

        let dir = scratch_dir("maintenance", "deletion");
        let db = dir.join("deletion.sqlite3");
        let assembly = crate::sqlite::SqliteRegistryAssembly::open(&db)
            .await
            .unwrap();
        let blobs = Arc::new(
            BlobStore::kv_only_sqlite(&db, 128 * 1024, Duration::ZERO)
                .await
                .unwrap(),
        );
        let front = ObjectFront::new(blobs.clone());
        let source = RefClosure::new(
            front.clone(),
            assembly.refs().clone() as Arc<dyn RefRows>,
            assembly.refs_version().clone(),
        );
        source.register(blobs.as_ref()).unwrap();

        let victim = record("owner/victimized");
        let survivor = record("owner/surviving");
        assembly.backend().put_repo(&victim).await.unwrap();
        assembly.backend().put_repo(&survivor).await.unwrap();

        let survivor_covered = put_object(&front, Kind::Blob, b"survivor\n").await;
        create_ref(
            &**assembly.refs(),
            "owner/surviving",
            "refs/heads/main",
            survivor_covered,
        )
        .await;

        let covered = put_object(&front, Kind::Blob, b"victim's only object\n").await;
        let orphan = put_object(&front, Kind::Blob, b"victim's stray\n").await;
        create_ref(
            &**assembly.refs(),
            "owner/victimized",
            "refs/heads/main",
            covered,
        )
        .await;

        assembly
            .backend()
            .remove_repo("owner/victimized")
            .await
            .unwrap();

        let chain = Arc::new(MaintenanceChain::new(
            assembly.store().clone(),
            blobs.clone(),
            assembly.reactor().clone(),
            AppliedWatermark::new(),
            None,
            quiet_opts(),
        ));
        let queue = assembly
            .store()
            .queue(MAINTENANCE_QUEUE, QueueOpts::default())
            .await
            .unwrap();
        for _ in 0..3 {
            queue
                .enqueue(
                    serde_json::to_value(MaintenanceJob::Sweep).unwrap(),
                    Default::default(),
                )
                .await
                .unwrap();
        }

        let seen: Seen = Arc::new(Mutex::new(Vec::new()));
        let stop = StopToken::new();
        let runner = spawn_run(chain, stop.clone(), seen.clone());
        wait_object_gone(&front, &covered).await;
        wait_object_gone(&front, &orphan).await;
        assert!(
            front.stat(&survivor_covered).await.unwrap().is_some(),
            "the surviving repo's covered object is untouched"
        );
        // The object's reclamation is observable inside the sweep's
        // run — before the worker's own report event lands (the
        // poll-versus-callback ordering). Wait for the verdict the
        // reclaiming sweep carries, not for the objects alone.
        wait_until(|| {
            saw_sweep_verdict(&seen, |v| {
                matches!(
                    v,
                    SweepVerdict::Report(report) if report.aborted.is_none() && report.deleted >= 2
                )
            })
        })
        .await;

        stop_and_join(stop, runner).await;
        drop(queue);
        drop(assembly);
        drop(blobs);
        drop(front);
        let _ = std::fs::remove_dir_all(&dir);
    }

    fn tree_bytes(entries: &[(&str, &ObjectId)]) -> Vec<u8> {
        let mut tree = Vec::new();
        for (name, oid) in entries {
            tree.extend_from_slice(format!("100644 {name}\0").as_bytes());
            tree.extend_from_slice(oid.as_bytes());
        }
        tree
    }

    fn commit_bytes(tree: &ObjectId, parents: &[&ObjectId]) -> Vec<u8> {
        let mut body = format!("tree {tree}\n");
        for parent in parents {
            body.push_str(&format!("parent {parent}\n"));
        }
        body.push_str(
            "author A <a@example.invalid> 1750000000 +0000\n\
             committer A <a@example.invalid> 1750000000 +0000\n\nthe message\n",
        );
        body.into_bytes()
    }

    #[derive(Clone, Default)]
    struct VecSink(Arc<Mutex<Vec<u8>>>);

    impl VecSink {
        fn bytes(&self) -> Vec<u8> {
            self.0.lock().unwrap().clone()
        }
    }

    impl std::io::Write for VecSink {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            self.0.lock().unwrap().extend_from_slice(buf);
            Ok(buf.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    /// The wake-runner supervision's probe store: `listen` fails the
    /// configured number of times (either arm of the N-b rider's stop
    /// shapes) before delegating; everything else delegates.
    struct FaultyListenStore {
        inner: Box<dyn Store>,
        closed_left: AtomicU64,
        database_left: AtomicU64,
        listens: AtomicU64,
    }

    impl FaultyListenStore {
        fn closed_once(inner: Box<dyn Store>) -> Arc<Self> {
            Arc::new(FaultyListenStore {
                inner,
                closed_left: AtomicU64::new(1),
                database_left: AtomicU64::new(0),
                listens: AtomicU64::new(0),
            })
        }

        fn database_once(inner: Box<dyn Store>) -> Arc<Self> {
            Arc::new(FaultyListenStore {
                inner,
                closed_left: AtomicU64::new(0),
                database_left: AtomicU64::new(1),
                listens: AtomicU64::new(0),
            })
        }
    }

    /// The immediately-closing receiver: `recv_timeout` errs `Closed`
    /// — the runner's quiet-return arm (the source-death class).
    struct DeadReceiver;

    impl WakeReceiver for DeadReceiver {
        fn recv<'a>(&'a mut self) -> alkstore::BoxedFuture<'a, Option<alkstore::Wake>> {
            Box::pin(async { None })
        }

        fn try_recv(&mut self) -> alkstore::Result<Option<alkstore::Wake>> {
            Err(alkstore::Error::Closed)
        }

        fn recv_timeout<'a>(
            &'a mut self,
            _timeout: Duration,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Option<alkstore::Wake>>> {
            Box::pin(async { Err(alkstore::Error::Closed) })
        }
    }

    impl Store for FaultyListenStore {
        fn begin_tx(
            &self,
        ) -> alkstore::BoxedFuture<'_, alkstore::Result<Box<dyn alkstore::TxHandle + Send>>>
        {
            self.inner.begin_tx()
        }

        fn notify<'a>(
            &'a self,
            channel: &str,
            payload: Value,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<()>> {
            self.inner.notify(channel, payload)
        }

        fn listen<'a>(
            &'a self,
            channel: &str,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn WakeReceiver>>> {
            self.listens.fetch_add(1, Ordering::Relaxed);
            let inner = &self.inner;
            if self.closed_left.swap(0, Ordering::Relaxed) > 0 {
                return Box::pin(
                    async move { Ok(Box::new(DeadReceiver) as Box<dyn WakeReceiver>) },
                );
            }
            if self.database_left.swap(0, Ordering::Relaxed) > 0 {
                return Box::pin(async {
                    Err(alkstore::Error::database("listen storage failure (probe)"))
                });
            }
            inner.listen(channel)
        }

        fn stream<'a>(
            &'a self,
            name: &str,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn alkstore::StreamHandle>>> {
            self.inner.stream(name)
        }

        fn queue<'a>(
            &'a self,
            name: &str,
            opts: QueueOpts,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn alkstore::Queue>>> {
            self.inner.queue(name, opts)
        }

        fn outbox<'a>(
            &'a self,
            name: &str,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn alkstore::Outbox>>> {
            self.inner.outbox(name)
        }

        fn try_lock<'a>(
            &'a self,
            name: &str,
            owner: &str,
            ttl: i64,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Option<Box<dyn alkstore::Lock>>>> {
            self.inner.try_lock(name, owner, ttl)
        }

        fn schedule<'a>(
            &'a self,
            name: &str,
            spec: &str,
            queue: &str,
            payload: Value,
            opts: alkstore::ScheduleOpts,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<alkstore::Schedule>> {
            self.inner.schedule(name, spec, queue, payload, opts)
        }

        fn unschedule<'a>(
            &'a self,
            name: &str,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<bool>> {
            self.inner.unschedule(name)
        }

        fn run_schedules<'a>(
            &'a self,
            stop: StopToken,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<()>> {
            self.inner.run_schedules(stop)
        }
    }
}