alkgit-substrate 0.1.0

//! The reactive halves — the engine-generic reactor (ADR-019 §5,
//! substrate.md §the reactive halves, POC-R's validated composition).
//!
//! Per record/mutation write, **after the durable commit**, the writing
//! side announces its effects independently and fail-soft:
//!
//! 1. **Publish** → the durable change feed ([`EVENTS_STREAM`], keyed
//!    by repo id) — per-consumer durable cursors, monotone saves, replay
//!    from head; *per-seq idempotency is the consumer's rule* (keyed
//!    publishes are carried metadata, not partitioning/dedup — the
//!    alkstore streams contract).
//! 2. **Notify** → the wake hint (`notify`/`listen` on [`WAKE_CHANNEL`])
//!    — the invalidation trigger for every cached provider. Publish
//!    precedes notify (the durable carrier before the latency hint).
//! 3. **Enqueue** → the maintenance queue ([`MAINTENANCE_QUEUE`]) — the
//!    housekeeping job riding the scheduler chain. Wave 6 landed the
//!    enqueue producer side only (the recorded interpretation,
//!    gate-confirmed); the **consumer side** — the chain (schedule →
//!    queue → worker claim → leader TTL lock → sweep dispatch) and the
//!    liveness sources — lives in [`crate::maintenance`] and
//!    [`crate::liveness`].
//!
//! Every leg is fail-soft: its loss is exactly what the **reconcile
//! sweep** exists to close — [`Reactor::reconcile`] re-reads
//! `rows_since(applied_watermark)` → snapshot install → watermark
//! advance, idempotent by construction (pure reads, monotone
//! watermark), callable on demand and driven by the maintenance
//! chain's scheduler tick. No
//! cross-store transaction machinery exists anywhere: alkstore's
//! `TxHandle` carries no consumer SQL, so an announce is always
//! post-commit.
//!
//! # The sqlite/postgres asymmetry (designed-in, ADR-019 §5)
//!
//! - **sqlite**: the co-tenant file is itself the invalidation bus —
//!   alkstore's watcher `data_version` poll fans a wake to `listen`
//!   receivers on *any* commit to the file, including alkgit's own
//!   registry writes on their own connection, including cross-process
//!   (~3.2 ms measured, POC-R). A lost hint cannot exist while any
//!   write lands; the reconcile sweep is a correctness backstop, not a
//!   latency bound. The post-commit notify is redundant but kept — for
//!   semantics and for postgres.
//! - **postgres**: LISTEN/NOTIFY is **load-bearing** — no file-level
//!   bump exists. A lost/failed notify leaves caches stale until the
//!   reconcile tick: **the tick's cadence is the real staleness bound
//!   on postgres**. (The live-pg asymmetry probe rides
//!   `substrate/w6-registry-postgres`'s tests.)
//!
//! # The construction wiring
//!
//! [`Reactor`] is constructed from the assembly's engine handle
//! (`Arc<dyn alkstore::Store>`) + the registry write side + the cached
//! provider ([`CachedRegistryProvider`]) — the per-piece constructor
//! the engine-twin adapter constructors wire (and the wave-8 single
//! composition entry will unify; the recorded interpretation, flagged
//! for the gate). Constructing the struct is infallible — every
//! consumed surface is a fallible call at use time; the adapter-twin
//! constructors own the loud-failure assembly posture
//! ([`OpenError`](crate::OpenError)).
//!
//! # Names pinned here (the OQ-03 substrate-side inventory for this
//! wave's members)
//!
//! `WAKE_CHANNEL`, `EVENTS_STREAM`, `MAINTENANCE_QUEUE`,
//! `MAINTENANCE_LOCK`, `RegistryEvent` (the change-feed payload:
//! `Put`/`Mutation`/`Removed` — `Mutation` is the per-push marker's
//! kind), `MaintenanceJob` (the maintenance-queue payload vocabulary),
//! `AnnounceReport`, `AppliedWatermark`, `Reactor`, and the announced
//! writing surface (`AnnouncingRegistry` — the durable write side +
//! the reactor's post-commit announce, the wrapper the engine-twin
//! assembly constructors compose).

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

use alkstore::{EnqueueOpts, QueueOpts, StopToken, Store, StreamEvent};
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::registry::{CachedRegistryProvider, RegistryRow, RegistryWrite};

use alkgit::RegistryError;

/// The wake channel — the notify/listen channel the announce's notify
/// leg signals and the wake runner listens on. Wakes are opaque
/// best-effort hints (coalesced, replay-free, possibly repeated);
/// consumers re-read — the wake receiver's idempotency obligation is
/// the provider's whole-table refresh (a swap, delta-free).
/// Contract stamp: ADR-019 §5 (the notify wake leg).
pub const WAKE_CHANNEL: &str = "alkgit-registry";

/// The durable change feed — the record/mutation event log. Keyed
/// publishes carry the repo id as metadata (per-seq idempotency, not
/// the key, is the consumer's dedup). The feed is the notify leg's
/// durable carrier: a wake is latency, an event is correctness.
/// Contract stamp: ADR-019 §5 (the durable event leg).
pub const EVENTS_STREAM: &str = "alkgit-registry-events";

/// The maintenance queue — the enqueue leg's target and (from wave 7)
/// the chain the scheduler fires and the maintenance worker claims
/// from. Contract stamp: ADR-019 §5 (the maintenance enqueue leg's
/// target; §6 — the chain consumes it).
pub const MAINTENANCE_QUEUE: &str = "alkgit-maintenance";

/// The named TTL lock the sweep dispatch takes for cross-process GC
/// exclusion (a locked-out sweep is a skip + short retry, a value not
/// an error — the chain's consumer side pins the usage, wave 7;
/// ADR-019 §6).
pub const MAINTENANCE_LOCK: &str = "alkgit-sweeper";

/// The registry mutation event — the durable change feed's payload.
/// Every variant carries the mutation's sequence number: the
/// consumer's per-seq idempotency key and the reconcile watermark's
/// currency. The `Mutation` variant is the per-push mutation-marker
/// event's kind (ADR-019 §7 — pushes do not modify `RepoRecord`; the
/// marker write's announce carries the bare mutation).
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum RegistryEvent {
    /// A record was created or replaced (the `put`/`update` mutations).
    Put {
        /// The mutation's sequence number.
        seq: i64,
        /// The record as committed.
        record: alkgit::RepoRecord,
    },
    /// The repo mutated without a record change — the per-push
    /// mutation-marker write's event (the watermark leg that drives
    /// reconcile and diff-driven replication on every push).
    Mutation {
        /// The mutation's sequence number.
        seq: i64,
        /// The repo id the mutation anchored.
        repo_id: String,
    },
    /// A record was removed (the removal consumed a sequence; the row
    /// is gone — the event is the removal's only row-carrying trace).
    Removed {
        /// The mutation's sequence number.
        seq: i64,
        /// The removed repo id.
        repo_id: String,
    },
}

impl RegistryEvent {
    /// The event's repo id (the publish key).
    pub fn repo_id(&self) -> &str {
        match self {
            RegistryEvent::Put { record, .. } => &record.repo_id,
            RegistryEvent::Mutation { repo_id, .. } | RegistryEvent::Removed { repo_id, .. } => {
                repo_id
            }
        }
    }

    /// The event's sequence number.
    pub fn seq(&self) -> i64 {
        match self {
            RegistryEvent::Put { seq, .. }
            | RegistryEvent::Mutation { seq, .. }
            | RegistryEvent::Removed { seq, .. } => *seq,
        }
    }

    /// Decode a feed event. Undecodable payloads (a foreign producer on
    /// the stream, codec drift) yield `None` — the consumer skips and
    /// checkpoints past them rather than wedging its cursor on the
    /// poison event.
    pub fn decode_event(event: &StreamEvent) -> Option<RegistryEvent> {
        let value = event.payload_as::<Value>().ok()?;
        serde_json::from_value(value).ok()
    }
}

/// The maintenance-queue payload vocabulary — what the enqueue leg
/// lands and what the wave-7 worker dispatches on; additions extend
/// the set.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum MaintenanceJob {
    /// The per-mutation housekeeping job the announce's enqueue leg
    /// lands — the pack work the mutated repo now owes the maintenance
    /// worker (the wave-7 dispatch).
    PackHousekeeping {
        /// The repo whose mutation landed the job.
        repo_id: String,
    },
    /// The scheduled GC sweep firing — the maintenance chain's
    /// `@every` schedule payload (the chain's consumer side is wave
    /// 7's: schedule → queue → worker claim → leader TTL lock →
    /// sweep dispatch, which needs a registered liveness source).
    Sweep,
}

/// The announce's per-leg report — fail-soft outcomes as observation
/// data, never an error: every leg's loss is the reconcile sweep's
/// closure, so a suppressed announce carries no error path.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct AnnounceReport {
    /// The publish leg landed (the durable event is in the feed).
    pub published: bool,
    /// The notify leg landed (the wake hint was signaled).
    pub notified: bool,
    /// The enqueue leg landed (`Some` carries the job id).
    pub enqueued_job_id: Option<i64>,
}

/// The consumer-side high-water mark — the maximum sequence this
/// composition has applied (stream events or a reconcile sweep).
/// Monotone: per-repo keys reorder nothing (the feed is global FIFO by
/// offset), but a reconcile may apply a sequence a live event never
/// carried; a watermark only advances, so the composition converges to
/// the durable side's own truth.
///
/// Process-local by design: the durable per-consumer cursor is the
/// stream's own checkpoint (monotone saves; restart resumes *after*
/// the saved checkpoint), so a restart re-delivers only the window
/// between apply and checkpoint — and per-seq idempotency is the
/// consumer's rule, exactly that re-application posture.
#[derive(Default)]
pub struct AppliedWatermark {
    seq: std::sync::atomic::AtomicI64,
}

impl AppliedWatermark {
    /// A fresh watermark at 0 (the whole table is unapplied —
    /// `rows_since(0)`).
    pub fn new() -> Arc<Self> {
        Arc::new(AppliedWatermark::default())
    }

    /// Raise the watermark to `seq` if it is higher (monotone).
    pub fn observe(&self, seq: i64) {
        self.seq
            .fetch_max(seq, std::sync::atomic::Ordering::Relaxed);
    }

    /// The current high-water mark.
    pub fn get(&self) -> i64 {
        self.seq.load(std::sync::atomic::Ordering::Relaxed)
    }
}

/// The reactor — the composition's wired reactive halves, constructed
/// from the substrate engine handle + the registry write side + the
/// cached provider (the assembly-level pieces; the wave-8 composition
/// entry unifies the whole backend's construction).
///
/// The registry write side is held for the wake runner's and the
/// reconcile sweep's refresh reads (`rows_since`) — the pair the
/// engine-twin adapters implement, so the reactor wiring is
/// engine-generic (it ran unchanged over all three engines in POC-R).
pub struct Reactor {
    store: Arc<dyn Store>,
    registry: Arc<dyn RegistryWrite>,
    provider: Arc<CachedRegistryProvider>,
}

/// The stop-grace loop granularity of the stop-responsive consumer
/// arms (responsiveness only — no consumer loop here wakes on a
/// timer).
const STOP_GRACE_MS: u64 = 50;

/// The wake runner's wait ceiling — stop-token responsiveness on the
/// receiver's timeout form; wakes never wait for it.
const WAKE_WAIT_MS: u64 = 250;

impl Reactor {
    /// Construct the reactor from the assembly's pieces.
    pub fn new(
        store: Arc<dyn Store>,
        registry: Arc<dyn RegistryWrite>,
        provider: Arc<CachedRegistryProvider>,
    ) -> Self {
        Reactor {
            store,
            registry,
            provider,
        }
    }

    /// The substrate engine handle (the wave-7 maintenance chain rides
    /// the same handle).
    pub fn store(&self) -> &Arc<dyn Store> {
        &self.store
    }

    /// The registry write side (the refresh source).
    pub fn registry(&self) -> &Arc<dyn RegistryWrite> {
        &self.registry
    }

    /// The cached provider (the invalidation target).
    pub fn provider(&self) -> &Arc<CachedRegistryProvider> {
        &self.provider
    }

    /// **The announce composition** — the one call a writing side makes
    /// after its durable row-write transaction commits: publish
    /// ([`EVENTS_STREAM`], keyed by repo id), notify ([`WAKE_CHANNEL`]),
    /// and enqueue ([`MAINTENANCE_QUEUE`], a pack-housekeeping job for
    /// the mutated repo) — each leg independent, each fail-soft. The
    /// legs' outcomes are report data ([`AnnounceReport`]); this call
    /// never fails, and any leg that failed is closed by the reconcile
    /// sweep.
    ///
    /// A suppression (the process dying between the durable commit and
    /// this call, or mid-legs) is the same posture — the sweep closes
    /// the window in one pass. On postgres the notify leg is
    /// load-bearing (the module docs' asymmetry): its loss leaves
    /// caches stale until the reconcile tick.
    pub async fn announce(&self, event: &RegistryEvent) -> AnnounceReport {
        let payload = match serde_json::to_value(event) {
            Ok(payload) => payload,
            Err(_) => {
                return AnnounceReport::default();
            }
        };

        let published = match self.store.stream(EVENTS_STREAM).await {
            Ok(stream) => stream
                .publish_with_key(Some(event.repo_id().to_owned()), payload.clone())
                .await
                .is_ok(),
            Err(_) => false,
        };

        let notified = self.store.notify(WAKE_CHANNEL, payload).await.is_ok();

        let enqueued_job_id = self
            .enqueue(&MaintenanceJob::PackHousekeeping {
                repo_id: event.repo_id().to_owned(),
            })
            .await
            .ok();

        AnnounceReport {
            published,
            notified,
            enqueued_job_id,
        }
    }

    /// The enqueue leg as a direct producer API — one housekeeping job
    /// on [`MAINTENANCE_QUEUE`]. The consumer side (the claim → leader
    /// lock → dispatch chain) is wave 7's; the job's payload
    /// (`MaintenanceJob`) is that worker's dispatch vocabulary.
    pub async fn enqueue(&self, job: &MaintenanceJob) -> alkstore::Result<i64> {
        let payload =
            serde_json::to_value(job).map_err(|e| alkstore::Error::Codec(e.to_string()))?;
        let queue = self
            .store
            .queue(MAINTENANCE_QUEUE, QueueOpts::default())
            .await?;
        queue.enqueue(payload, EnqueueOpts::default()).await
    }

    /// **The reconcile sweep** — the closure for any lost post-commit
    /// effect: `rows_since(applied_watermark)` → snapshot install →
    /// watermark advance. Idempotent by construction (pure reads, a
    /// monotone watermark — the second pass is empty). Driven by the
    /// scheduler tick once the maintenance chain's consumer side lands
    /// (wave 7), callable on demand. Returns the rows it caught up on
    /// (empty = nothing missed). A failed sweep changes nothing — the
    /// watermark does not advance and the previous snapshot serves; the
    /// next wake, write, or tick retries.
    pub async fn reconcile(
        &self,
        applied: &AppliedWatermark,
    ) -> Result<Vec<RegistryRow>, alkgit::RegistryError> {
        let rows = self.registry.rows_since(applied.get()).await?;
        self.provider.refresh_from(&*self.registry).await?;
        applied.observe(self.registry.watermark().await?);
        Ok(rows)
    }

    /// **The wake runner** — the invalidation trigger's delivery leg:
    /// `listen([`WAKE_CHANNEL`])` → one whole-table refresh
    /// ([`CachedRegistryProvider::refresh_from`]) per wake. Coalesced
    /// bursts collapse engine-side (the sqlite watcher's burst
    /// coalescing; per-notify on postgres) and one refresh carries a
    /// whole burst — the count of refreshes is correct by the snapshot
    /// swap. There is no polling at this layer: no commits → no
    /// refreshes (the refresh count is stable across idle windows).
    /// Refresh failures leave the previous snapshot serving (the next
    /// wake, write, or tick retries).
    ///
    /// Returns when the source closes — a closed receiver never
    /// reopens; recovery is a fresh `listen` (the runner attached
    /// again; the reconcile tick is the staleness bound in between).
    pub async fn run_wake_runner(&self, stop: StopToken) -> alkstore::Result<()> {
        let mut rx = self.store.listen(WAKE_CHANNEL).await?;
        while !stop.is_cancelled() {
            match rx.recv_timeout(Duration::from_millis(WAKE_WAIT_MS)).await {
                Ok(Some(_)) => {
                    self.provider.refresh_from(&*self.registry).await.ok();
                }
                Ok(None) => {}
                Err(_) => return Ok(()),
            }
        }
        Ok(())
    }

    /// **The stream consumer** — the durable change feed's consumer
    /// side under the alkstore cursor model: `subscribe(consumer)`
    /// attaches at the consumer's stored checkpoint (absent = 0 = the
    /// log head — a fresh consumer replays from the head; a restarted
    /// one resumes *after* the saved checkpoint); each observation is
    /// checkpointed after the apply (monotone saves — saves cannot
    /// regress); **per-seq idempotency is applied as the consumer's
    /// rule** — an event whose seq is at or below the applied watermark
    /// is checkpointed past without re-invoking the callback.
    ///
    /// `on_event` receives the decoded event and its raw feed position.
    /// An undecodable payload is checkpointed past (a foreign producer
    /// must not wedge the cursor). A storage failure backs off and
    /// retries — the feed is durable, silence loses nothing. Returns on
    /// the receiver's terminal close (recovery is a fresh re-attach —
    /// the saved checkpoint resumes) or a stopped token; only a
    /// checkpoint save failure fails the run (the durable cursor is the
    /// restart boundary — losing it silently would widen the replay),
    /// and re-attaching retries it from the last landed checkpoint.
    pub async fn run_stream_consumer(
        &self,
        consumer: &str,
        applied: &AppliedWatermark,
        stop: StopToken,
        mut on_event: impl FnMut(RegistryEvent, StreamEvent) + Send,
    ) -> alkstore::Result<()> {
        let stream = self.store.stream(EVENTS_STREAM).await?;
        let mut rx = stream.subscribe(consumer).await?;
        loop {
            tokio::select! {
                _ = stop_grace(&stop) => return Ok(()),
                event = rx.recv() => {
                    match event {
                        None => return Ok(()),
                        Some(Err(_)) => {
                            tokio::time::sleep(Duration::from_millis(STOP_GRACE_MS)).await;
                        }
                        Some(Ok(event)) => {
                            if let Some(decoded) = RegistryEvent::decode_event(&event) {
                                let seq = decoded.seq();
                                if seq > applied.get() {
                                    on_event(decoded, event.clone());
                                    applied.observe(seq);
                                }
                            }
                            rx.save_offset()?;
                        }
                    }
                }
            }
        }
    }
}

/// **The announced write side** — the reactive composition's writing
/// surface: the durable registry write side folded with the reactor,
/// so every mutation's post-commit announce ([`Reactor::announce`],
/// fail-soft each) fires right after its durable commit. This is the
/// wrapper the engine-twin assembly constructors compose — the pair's
/// durable contract below is unchanged (announce is never the pair's;
/// the wrapper is the *caller*, and the announce's machinery is the
/// reactor's).
///
/// Consumers bind this as the write side of their
/// [`RegistryBackend`](crate::RegistryBackend) (the store methods then
/// announce post-commit); a suppressed announce is the crash-window
/// shape the reconcile sweep closes — the mutation still committed
/// (the seq is transactionally exact, never announce-dependent).
/// Contract stamp: ADR-019 §5 (the post-commit announce posture) and
/// §7 (the announced receive flow).
pub struct AnnouncingRegistry {
    durable: Arc<dyn RegistryWrite>,
    reactor: Arc<Reactor>,
}

impl AnnouncingRegistry {
    /// Wrap the durable write side + the reactor that announces for it.
    pub fn new(durable: Arc<dyn RegistryWrite>, reactor: Arc<Reactor>) -> Self {
        AnnouncingRegistry { durable, reactor }
    }

    /// The durable write side below the announce (the marker write's
    /// and the reconcile machinery's legs run at this layer).
    pub fn durable(&self) -> &Arc<dyn RegistryWrite> {
        &self.durable
    }

    /// The reactor firing the announce legs.
    pub fn reactor(&self) -> &Arc<Reactor> {
        &self.reactor
    }
}

#[async_trait]
impl RegistryWrite for AnnouncingRegistry {
    async fn put(&self, record: &alkgit::RepoRecord) -> Result<i64, RegistryError> {
        let seq = self.durable.put(record).await?;
        self.reactor
            .announce(&RegistryEvent::Put {
                seq,
                record: record.clone(),
            })
            .await;
        Ok(seq)
    }

    async fn update(&self, record: &alkgit::RepoRecord) -> Result<i64, RegistryError> {
        let seq = self.durable.update(record).await?;
        self.reactor
            .announce(&RegistryEvent::Put {
                seq,
                record: record.clone(),
            })
            .await;
        Ok(seq)
    }

    async fn remove(&self, repo_id: &str) -> Result<i64, RegistryError> {
        let seq = self.durable.remove(repo_id).await?;
        self.reactor
            .announce(&RegistryEvent::Removed {
                seq,
                repo_id: repo_id.to_owned(),
            })
            .await;
        Ok(seq)
    }

    async fn watermark(&self) -> Result<i64, RegistryError> {
        self.durable.watermark().await
    }

    async fn rows_since(&self, seq: i64) -> Result<Vec<RegistryRow>, RegistryError> {
        self.durable.rows_since(seq).await
    }
}

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;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::AtomicBool;
    use std::sync::atomic::Ordering::Relaxed;
    use std::sync::Mutex;

    use alkstore::StreamHandle;
    use alkstore::WakeReceiver;

    use crate::registry::{InMemoryRegistry, RegistryRead};
    #[cfg(feature = "sqlite")]
    use crate::testutil::unique_db_path;
    use crate::testutil::{record, spawn_wake_runner, stop_and_join, wait_until};

    /// A `Store` double that suppresses one announce leg's entry point
    /// (everything else delegates to the mem engine) — the
    /// fail-soft/independence pin's driver.
    struct LegFailureStore {
        inner: Box<dyn Store>,
        fail_publish: AtomicBool,
        fail_notify: AtomicBool,
        fail_enqueue: AtomicBool,
    }

    #[derive(Clone, Copy, PartialEq)]
    enum Leg {
        Publish,
        Notify,
        Enqueue,
    }

    impl LegFailureStore {
        fn suppressing(leg: Leg) -> Arc<Self> {
            let store = Arc::new(LegFailureStore {
                inner: Box::new(alkstore_mem::MemStore::new()),
                fail_publish: AtomicBool::new(false),
                fail_notify: AtomicBool::new(false),
                fail_enqueue: AtomicBool::new(false),
            });
            match leg {
                Leg::Publish => store.fail_publish.store(true, Relaxed),
                Leg::Notify => store.fail_notify.store(true, Relaxed),
                Leg::Enqueue => store.fail_enqueue.store(true, Relaxed),
            }
            store
        }
    }

    impl Store for LegFailureStore {
        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<()>> {
            if self.fail_notify.load(Relaxed) {
                return Box::pin(async {
                    Err(alkstore::Error::database("notify leg suppressed (probe)"))
                });
            }
            self.inner.notify(channel, payload)
        }

        fn listen<'a>(
            &'a self,
            channel: &str,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn WakeReceiver>>> {
            self.inner.listen(channel)
        }

        fn stream<'a>(
            &'a self,
            name: &str,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn StreamHandle>>> {
            if self.fail_publish.load(Relaxed) {
                return Box::pin(async {
                    Err(alkstore::Error::database("publish leg suppressed (probe)"))
                });
            }
            self.inner.stream(name)
        }

        fn queue<'a>(
            &'a self,
            name: &str,
            opts: QueueOpts,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn alkstore::Queue>>> {
            if self.fail_enqueue.load(Relaxed) {
                return Box::pin(async {
                    Err(alkstore::Error::database("enqueue leg suppressed (probe)"))
                });
            }
            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)
        }
    }

    fn event(seq: i64, id: &str) -> RegistryEvent {
        RegistryEvent::Put {
            seq,
            record: record(id),
        }
    }

    fn assembly() -> (
        Arc<Reactor>,
        Arc<dyn RegistryWrite>,
        Arc<CachedRegistryProvider>,
    ) {
        let registry: Arc<dyn RegistryWrite> = Arc::new(InMemoryRegistry::new());
        let provider = Arc::new(CachedRegistryProvider::new());
        let reactor = Arc::new(Reactor::new(
            mem_store(),
            registry.clone(),
            provider.clone(),
        ));
        (reactor, registry, provider)
    }

    fn mem_store() -> Arc<dyn Store> {
        Arc::new(alkstore_mem::MemStore::new())
    }

    /// Publish `RegistryEvent`s straight onto the feed — the test
    /// producer path that bypasses the announce (the cursor tests
    /// stage feeds precisely with it).
    async fn stage(reactor: &Reactor, events: &[RegistryEvent]) {
        let stream = reactor.store().stream(EVENTS_STREAM).await.unwrap();
        for event in events {
            stream
                .publish_with_key(
                    Some(event.repo_id().to_owned()),
                    serde_json::to_value(event).unwrap(),
                )
                .await
                .unwrap();
        }
    }

    type Seen = Arc<Mutex<Vec<(String, i64)>>>;

    fn spawn_consumer(
        reactor: Arc<Reactor>,
        consumer: &'static str,
        applied: Arc<AppliedWatermark>,
        seen: Seen,
    ) -> (StopToken, tokio::task::JoinHandle<()>) {
        let stop = StopToken::new();
        let join = tokio::spawn({
            let stop = stop.clone();
            async move {
                reactor
                    .run_stream_consumer(consumer, &applied, stop, move |event, raw| {
                        seen.lock()
                            .unwrap()
                            .push((event.repo_id().to_owned(), raw.offset));
                    })
                    .await
                    .expect("the consumer ran clean");
            }
        });
        (stop, join)
    }

    #[tokio::test]
    async fn announce_lands_all_three_legs() {
        let (reactor, registry, _) = assembly();
        let store = reactor.store().clone();

        let mut listener = store.listen(WAKE_CHANNEL).await.unwrap();
        let seq = registry.put(&record("demo")).await.unwrap();
        let report = reactor
            .announce(&RegistryEvent::Put {
                seq,
                record: record("demo"),
            })
            .await;
        assert_eq!(
            report,
            AnnounceReport {
                published: true,
                notified: true,
                enqueued_job_id: Some(1),
            }
        );

        let wake = listener
            .recv_timeout(Duration::from_secs(2))
            .await
            .unwrap()
            .expect("wake source open");
        assert_eq!(wake.channel, WAKE_CHANNEL);

        let stream = store.stream(EVENTS_STREAM).await.unwrap();
        let events = stream.read_since(0, 10).await.unwrap();
        assert_eq!(
            events.len(),
            1,
            "one announced mutation = one durable feed event"
        );
        assert_eq!(events[0].key.as_deref(), Some("demo"), "keyed by repo id");
        assert_eq!(
            RegistryEvent::decode_event(&events[0]),
            Some(RegistryEvent::Put {
                seq,
                record: record("demo"),
            })
        );

        let queue = store
            .queue(MAINTENANCE_QUEUE, QueueOpts::default())
            .await
            .unwrap();
        let job = queue
            .get_job(report.enqueued_job_id.unwrap())
            .await
            .unwrap()
            .expect("the enqueue leg's job row");
        assert_eq!(
            job.payload_as::<MaintenanceJob>().unwrap(),
            MaintenanceJob::PackHousekeeping {
                repo_id: "demo".to_string(),
            }
        );
    }

    /// The announced write side composes the pair + the reactor: the
    /// durable mutation commits and the announce fires post-commit.
    #[tokio::test]
    async fn announcing_registry_writes_and_announces_post_commit() {
        let store = mem_store();
        let registry: Arc<dyn RegistryWrite> = Arc::new(InMemoryRegistry::new());
        let provider = Arc::new(CachedRegistryProvider::new());
        let reactor = Arc::new(Reactor::new(
            store.clone(),
            registry.clone(),
            provider.clone(),
        ));
        let announced = AnnouncingRegistry::new(registry.clone(), reactor.clone());

        let mut listener = store.listen(WAKE_CHANNEL).await.unwrap();
        let seq = announced.put(&record("demo")).await.unwrap();
        assert_eq!(seq, 1, "the durable mutation's seq is the wrapper's return");
        assert_eq!(
            registry.watermark().await.unwrap(),
            1,
            "the durable side is the watermark's truth"
        );

        let wake = listener
            .recv_timeout(Duration::from_secs(2))
            .await
            .unwrap()
            .expect("wake source open");
        assert_eq!(wake.channel, WAKE_CHANNEL);
        let stream = store.stream(EVENTS_STREAM).await.unwrap();
        let events = stream.read_since(0, 10).await.unwrap();
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].key.as_deref(), Some("demo"));
        assert_eq!(
            RegistryEvent::decode_event(&events[0]),
            Some(event(seq, "demo"))
        );
    }

    /// A suppressed announce never fails the mutation — the seq is
    /// transactionally exact, and the crash-window shape closes by one
    /// reconcile pass.
    #[tokio::test]
    async fn announcing_registry_fail_soft_legs_do_not_fail_the_write() {
        let registry: Arc<dyn RegistryWrite> = Arc::new(InMemoryRegistry::new());
        let provider = Arc::new(CachedRegistryProvider::new());
        let reactor = Arc::new(Reactor::new(
            LegFailureStore::suppressing(Leg::Publish) as Arc<dyn Store>,
            registry.clone(),
            provider.clone(),
        ));
        let announced = AnnouncingRegistry::new(registry.clone(), reactor.clone());

        let seq = announced.put(&record("x")).await.unwrap();
        assert_eq!(
            seq, 1,
            "the durable seq is exact though the announce's leg failed"
        );
        assert_eq!(registry.watermark().await.unwrap(), 1);

        let applied = AppliedWatermark::new();
        let caught = reactor.reconcile(&applied).await.unwrap();
        assert_eq!(caught.len(), 1, "the suppressed window closes in one pass");
        assert_eq!(provider.resolve("x"), Some(record("x")));
    }

    #[tokio::test]
    async fn announce_legs_are_independent_and_fail_soft() {
        let registry = Arc::new(InMemoryRegistry::new());
        let provider = Arc::new(CachedRegistryProvider::new());

        for (leg, expect) in [
            (
                Leg::Publish,
                AnnounceReport {
                    published: false,
                    notified: true,
                    enqueued_job_id: Some(1),
                },
            ),
            (
                Leg::Notify,
                AnnounceReport {
                    published: true,
                    notified: false,
                    enqueued_job_id: Some(1),
                },
            ),
            (
                Leg::Enqueue,
                AnnounceReport {
                    published: true,
                    notified: true,
                    enqueued_job_id: None,
                },
            ),
        ] {
            let store = LegFailureStore::suppressing(leg);
            let reactor = Reactor::new(
                store.clone() as Arc<dyn Store>,
                registry.clone(),
                provider.clone(),
            );

            let seq = registry.put(&record("leg")).await.unwrap();
            let event = RegistryEvent::Put {
                seq,
                record: record("leg"),
            };
            let report = reactor.announce(&event).await;
            assert_eq!(
                report, expect,
                "the other legs are untouched by the suppressed leg's loss"
            );

            registry.remove("leg").await.unwrap();
        }
    }

    #[tokio::test]
    async fn suppressed_announce_closes_by_one_reconcile_pass() {
        let (reactor, registry, provider) = assembly();
        let applied = AppliedWatermark::new();

        let seq = registry.put(&record("crash")).await.unwrap();
        let _ = registry.put(&record("also-durable")).await.unwrap();

        assert_eq!(
            provider.resolve("crash"),
            None,
            "no announce: the cache is stale until reconcile (the crash window)"
        );

        let caught = reactor.reconcile(&applied).await.unwrap();
        assert_eq!(
            caught
                .iter()
                .map(|row| row.record.repo_id.as_str())
                .collect::<Vec<_>>(),
            vec!["crash", "also-durable"],
            "rows_since(watermark) → snapshot install → watermark advance"
        );
        assert_eq!(provider.resolve("crash"), Some(record("crash")));
        assert_eq!(
            provider.resolve("also-durable"),
            Some(record("also-durable"))
        );
        assert!(applied.get() >= seq);

        let second = reactor.reconcile(&applied).await.unwrap();
        assert!(
            second.is_empty(),
            "the second pass is empty — idempotent by construction"
        );
    }

    #[tokio::test]
    async fn reconcile_covers_removals_and_the_marker_watermark() {
        let (reactor, registry, provider) = assembly();
        let applied = AppliedWatermark::new();

        let seq = registry.put(&record("a")).await.unwrap();
        let removals = registry.remove("a").await.unwrap();

        let caught = reactor.reconcile(&applied).await.unwrap();
        assert!(
            caught.is_empty(),
            "the removal leaves no row — the catch-up read carries none (the snapshot install surfaces it)"
        );
        assert_eq!(
            provider.resolve("a"),
            None,
            "the removal is in the snapshot"
        );
        assert!(
            applied.get() >= removals,
            "the removal consumed a sequence and the watermark advanced past it (past `seq` {seq})"
        );

        registry.put(&record("a")).await.unwrap();
        let caught = reactor.reconcile(&applied).await.unwrap();
        assert_eq!(
            caught
                .iter()
                .map(|row| row.record.repo_id.as_str())
                .collect::<Vec<_>>(),
            vec!["a"],
            "the marker leg's re-created row is caught up (per-seq mutation markers only move the watermark)"
        );
        assert_eq!(provider.resolve("a"), Some(record("a")));

        let second = reactor.reconcile(&applied).await.unwrap();
        assert!(second.is_empty());
    }

    #[tokio::test]
    async fn wake_runner_refreshes_from_wakes_and_never_polls() {
        let (reactor, registry, provider) = assembly();
        let (stop, runner) = spawn_wake_runner(reactor.clone());
        tokio::time::sleep(Duration::from_millis(200)).await;

        let seq = registry.put(&record("wake-me")).await.unwrap();
        reactor.announce(&event(seq, "wake-me")).await;

        wait_until(|| provider.resolve("wake-me").is_some()).await;
        assert_eq!(provider.resolve("wake-me"), Some(record("wake-me")));

        tokio::time::sleep(Duration::from_millis(400)).await;
        let before = provider.refresh_count();
        tokio::time::sleep(Duration::from_millis(600)).await;
        assert_eq!(
            before,
            provider.refresh_count(),
            "no commits → no refreshes — the runner does not poll"
        );

        stop_and_join(stop, runner).await;
    }

    #[tokio::test]
    async fn stream_consumer_replays_resumes_and_checkpoints() {
        let (reactor, registry, _) = assembly();
        let stream = reactor.store().stream(EVENTS_STREAM).await.unwrap();

        let s1 = registry.put(&record("a")).await.unwrap();
        stage(&reactor, &[event(s1, "a")]).await;

        let applied = AppliedWatermark::new();
        let seen: Seen = Arc::new(Mutex::new(Vec::new()));
        let (stop, join) =
            spawn_consumer(reactor.clone(), "consumer-1", applied.clone(), seen.clone());
        wait_until(|| seen.lock().unwrap().len() == 1).await;
        stop_and_join(stop, join).await;
        assert_eq!(
            &*seen.lock().unwrap(),
            &[("a".to_string(), 1)],
            "a fresh consumer replays from the log head"
        );
        assert_eq!(
            stream.get_offset("consumer-1").await.unwrap(),
            1,
            "the durable cursor landed at the applied event's offset"
        );

        let s2 = registry.put(&record("b")).await.unwrap();
        stage(&reactor, &[event(s2, "b")]).await;
        seen.lock().unwrap().clear();
        let (stop, join) =
            spawn_consumer(reactor.clone(), "consumer-1", applied.clone(), seen.clone());
        wait_until(|| seen.lock().unwrap().len() == 1).await;
        stop_and_join(stop, join).await;
        assert_eq!(
            &*seen.lock().unwrap(),
            &[("b".to_string(), 2)],
            "restart resumes AFTER the saved checkpoint"
        );
        assert_eq!(stream.get_offset("consumer-1").await.unwrap(), 2);

        let s3 = registry.put(&record("c")).await.unwrap();
        stage(&reactor, &[event(s3, "c")]).await;
        let fresh_seen: Seen = Arc::new(Mutex::new(Vec::new()));
        let fresh_watermark = AppliedWatermark::new();
        let (stop, join) = spawn_consumer(
            reactor.clone(),
            "consumer-2",
            fresh_watermark,
            fresh_seen.clone(),
        );
        wait_until(|| fresh_seen.lock().unwrap().len() == 3).await;
        stop_and_join(stop, join).await;
        assert_eq!(
            &*fresh_seen.lock().unwrap(),
            &[
                ("a".to_string(), 1),
                ("b".to_string(), 2),
                ("c".to_string(), 3)
            ],
            "a fresh consumer name has no cursor — the whole feed replays"
        );
    }

    #[tokio::test]
    async fn stream_consumer_is_per_seq_idempotent_and_skips_poison_events() {
        let (reactor, registry, _) = assembly();

        assert_eq!(registry.put(&record("a")).await.unwrap(), 1);
        registry.remove("a").await.unwrap();
        assert_eq!(registry.put(&record("b")).await.unwrap(), 3);

        let stream = reactor.store().stream(EVENTS_STREAM).await.unwrap();
        stream
            .publish(serde_json::json!("not-a-registry-event"))
            .await
            .unwrap();
        stage(
            &reactor,
            &[
                RegistryEvent::Put {
                    seq: 1,
                    record: record("a"),
                },
                RegistryEvent::Removed {
                    seq: 2,
                    repo_id: "a".to_string(),
                },
                RegistryEvent::Mutation {
                    seq: 3,
                    repo_id: "b".to_string(),
                },
            ],
        )
        .await;

        let applied = AppliedWatermark::new();
        applied.observe(1);

        let seen: Seen = Arc::new(Mutex::new(Vec::new()));
        let (stop, join) = spawn_consumer(reactor.clone(), "idem", applied.clone(), seen.clone());
        wait_until(|| seen.lock().unwrap().len() == 2).await;
        stop_and_join(stop, join).await;

        assert_eq!(
            &*seen.lock().unwrap(),
            &[("a".to_string(), 3), ("b".to_string(), 4)],
            "seq 1 is at/below the watermark (checkpointed, not re-invoked); \
             the poison event is checkpointed past without wedging the cursor"
        );
        assert_eq!(
            stream.get_offset("idem").await.unwrap(),
            4,
            "every observation was checkpointed (monotone saves)"
        );
        assert_eq!(
            applied.get(),
            3,
            "only applied events' sequences advance the watermark"
        );
    }

    #[tokio::test]
    async fn second_consumer_converges_via_the_feed_and_reconcile_is_empty() {
        let (reactor, registry, _) = assembly();

        let s1 = registry.put(&record("a")).await.unwrap();
        let s2 = registry.put(&record("b")).await.unwrap();
        stage(&reactor, &[event(s1, "a"), event(s2, "b")]).await;

        let applied = AppliedWatermark::new();
        let seen: Seen = Arc::new(Mutex::new(Vec::new()));
        let (stop, join) =
            spawn_consumer(reactor.clone(), "replica", applied.clone(), seen.clone());
        wait_until(|| seen.lock().unwrap().len() == 2).await;
        stop_and_join(stop, join).await;

        assert_eq!(
            &*seen.lock().unwrap(),
            &[("a".to_string(), 1), ("b".to_string(), 2)]
        );
        assert_eq!(applied.get(), 2);

        let caught = reactor.reconcile(&applied).await.unwrap();
        assert!(
            caught.is_empty(),
            "the watermark caught up from the feed — the reconcile pass is empty"
        );
    }

    #[tokio::test]
    async fn enqueue_is_the_producer_side_round_trip() {
        let (reactor, _, _) = assembly();
        let store = reactor.store().clone();

        let job_id = reactor
            .enqueue(&MaintenanceJob::PackHousekeeping {
                repo_id: "packed".to_string(),
            })
            .await
            .unwrap();
        let sweep_id = reactor.enqueue(&MaintenanceJob::Sweep).await.unwrap();

        let queue = store
            .queue(MAINTENANCE_QUEUE, QueueOpts::default())
            .await
            .unwrap();
        let job = queue.get_job(job_id).await.unwrap().expect("job row");
        assert_eq!(
            job.payload_as::<MaintenanceJob>().unwrap(),
            MaintenanceJob::PackHousekeeping {
                repo_id: "packed".to_string(),
            }
        );
        let sweep = queue.get_job(sweep_id).await.unwrap().expect("sweep row");
        assert_eq!(
            sweep.payload_as::<MaintenanceJob>().unwrap(),
            MaintenanceJob::Sweep
        );
    }

    #[tokio::test]
    async fn registry_event_payloads_round_trip_and_poison_decodes_none() {
        let stream = mem_store().stream(EVENTS_STREAM).await.unwrap();

        for original in [
            RegistryEvent::Put {
                seq: 7,
                record: record("owner/r"),
            },
            RegistryEvent::Mutation {
                seq: 8,
                repo_id: "owner/r".to_string(),
            },
            RegistryEvent::Removed {
                seq: 9,
                repo_id: "owner/r".to_string(),
            },
        ] {
            let value = serde_json::to_value(&original).unwrap();
            assert_eq!(value["kind"], expected_kind(&original));
            assert_eq!(value["seq"], original.seq());
            let decoded: RegistryEvent = serde_json::from_value(value).unwrap();
            assert_eq!(decoded, original);
        }

        let offset = stream
            .publish(serde_json::json!("a foreign producer's event"))
            .await
            .unwrap();
        let events = stream.read_since(offset - 1, 1).await.unwrap();
        assert_eq!(events.len(), 1);
        assert_eq!(RegistryEvent::decode_event(&events[0]), None);
    }

    fn expected_kind(event: &RegistryEvent) -> &str {
        match event {
            RegistryEvent::Put { .. } => "put",
            RegistryEvent::Mutation { .. } => "mutation",
            RegistryEvent::Removed { .. } => "removed",
        }
    }

    #[cfg(feature = "sqlite")]
    async fn sqlite_store(path: &std::path::Path) -> Arc<dyn Store> {
        let boxed = alkstore_sqlite::open(path.to_str().unwrap(), Default::default()).unwrap();
        Arc::from(boxed)
    }

    /// The reactive halves ran unchanged over the sqlite engine in
    /// POC-R; this leg re-pins the announce + cursor composition over
    /// it (the wake backstop of the sqlite co-tenant file is the
    /// registry twin's probe — its file shape is `w6-registry-sqlite`'s).
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn announce_and_stream_consumer_round_trip_over_sqlite() {
        let path = unique_db_path("reactor", "announce-cursor");
        let store = sqlite_store(&path).await;
        let registry: Arc<dyn RegistryWrite> = Arc::new(InMemoryRegistry::new());
        let provider = Arc::new(CachedRegistryProvider::new());
        let reactor = Arc::new(Reactor::new(
            store.clone(),
            registry.clone(),
            provider.clone(),
        ));

        let mut listener = store.listen(WAKE_CHANNEL).await.unwrap();
        let seq = registry.put(&record("s")).await.unwrap();
        let report = reactor.announce(&event(seq, "s")).await;
        assert_eq!(
            report,
            AnnounceReport {
                published: true,
                notified: true,
                enqueued_job_id: Some(1),
            }
        );

        let wake = listener
            .recv_timeout(Duration::from_secs(2))
            .await
            .unwrap()
            .expect("wake source open");
        assert_eq!(wake.channel, WAKE_CHANNEL);

        let stream = store.stream(EVENTS_STREAM).await.unwrap();
        let events = stream.read_since(0, 10).await.unwrap();
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].key.as_deref(), Some("s"));
        assert_eq!(
            RegistryEvent::decode_event(&events[0]),
            Some(event(seq, "s"))
        );

        let applied = AppliedWatermark::new();
        let seen: Seen = Arc::new(Mutex::new(Vec::new()));
        let (stop, join) =
            spawn_consumer(reactor.clone(), "sqlite-c", applied.clone(), seen.clone());
        wait_until(|| seen.lock().unwrap().len() == 1).await;
        stop_and_join(stop, join).await;
        assert_eq!(&*seen.lock().unwrap(), &[("s".to_string(), 1)]);
        assert_eq!(stream.get_offset("sqlite-c").await.unwrap(), 1);

        let caught = reactor.reconcile(&applied).await.unwrap();
        assert!(caught.is_empty(), "announced — reconcile is empty");
        drop(reactor);
        drop(store);
        let _ = std::fs::remove_file(&path);
    }

    /// The sqlite backstop (POC-R's DQ-8 probe, re-landed at the
    /// reactor layer): an announced co-tenant write pops the
    /// listener's wake via the file-wide `data_version` bump — no
    /// explicit notify involved — and the runner's refresh surfaces
    /// the durable record; then no commits → no refreshes (no
    /// polling).
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn sqlite_backstop_unannounced_write_wakes_the_runner() {
        let path = unique_db_path("reactor", "backstop");
        let store = sqlite_store(&path).await;
        let registry: Arc<dyn RegistryWrite> = Arc::new(InMemoryRegistry::new());
        let provider = Arc::new(CachedRegistryProvider::new());

        let seq = registry.put(&record("silent")).await.unwrap();
        assert_eq!(seq, 1);
        assert_eq!(provider.resolve("silent"), None, "cold cache");

        let reactor = Arc::new(Reactor::new(
            store.clone(),
            registry.clone(),
            provider.clone(),
        ));
        let (stop, runner) = spawn_wake_runner(reactor.clone());
        tokio::time::sleep(Duration::from_millis(200)).await;

        let queue = store
            .queue(MAINTENANCE_QUEUE, QueueOpts::default())
            .await
            .unwrap();
        queue
            .enqueue(serde_json::json!("probe"), EnqueueOpts::default())
            .await
            .unwrap();

        wait_until(|| provider.resolve("silent").is_some()).await;
        assert_eq!(provider.resolve("silent"), Some(record("silent")));

        tokio::time::sleep(Duration::from_millis(400)).await;
        let before = provider.refresh_count();
        tokio::time::sleep(Duration::from_millis(600)).await;
        assert_eq!(
            before,
            provider.refresh_count(),
            "no commits → no refreshes — the runner does not poll"
        );

        stop_and_join(stop, runner).await;
        drop(reactor);
        drop(store);
        let _ = std::fs::remove_file(&path);
    }
}