aion-rs 0.25.1

//! Expired timer polling at engine boot and shard adoption.
//!
//! THE BOOT/ADOPTION SWEEP IS THE ONLY SWEEP. [`TimerRecovery::recover_on_startup`]
//! runs exactly twice per engine lifetime shape: once at engine startup and
//! once per shard adoption (`Engine::adopt_shards`). There is deliberately NO
//! periodic driver in this landing — a row this sweep leaves behind (a failed
//! retirement, a wake lost after recording) heals at the NEXT BOOT OR
//! ADOPTION SWEEP, not on any timer-driven cadence. The periodic driver is
//! commissioned separately (interval from server config/builder, absence =
//! boot-only, loudly documented) — see the periodic-driver brief in the
//! collapse lane's follow-ups; this module must not pretend it exists.
//!
//! The sweep is GROUPED: due rows are bucketed by workflow and one
//! history read per workflow answers every question the sweep has — is the
//! workflow terminal (all its due rows are moot), which future timers still
//! need re-arming, and each due row's disposition. Only rows
//! that still owe fire-path work (a live arming whose row matches the
//! recorded arming, or a recorded fire whose mailbox wake may still be owed —
//! aion#145) enter
//! [`TimerService::fire_timer`]; consumed rows retire in bulk. Before
//! retirement existed, every consumed row was re-walked by every sweep
//! forever — the boot-walk collapse this module's grouping completes.
//!
//! The honest cost model: the SWEEP'S OWN store reads scale with distinct
//! workflows, never with rows. Rows that enter the fire or redelivery paths
//! additionally pay those paths' own costs — a live fire's gate read and
//! recorder append, and a redelivery's history read UNDER THE RECORDER LOCK
//! inside [`EngineHandle::record_redelivered_timer_fire`] — so the first
//! boot after a backlog accumulated still pays roughly one recorder-seam
//! round trip per surviving row. What the collapse guarantees is that each
//! such row pays it ONCE, EVER: retirement (or supersession) ends the row's
//! life, and the steady-state sweep reads nothing but the expired index.
//!
//! [`EngineHandle::record_redelivered_timer_fire`]: crate::engine_seam::EngineHandle::record_redelivered_timer_fire

use aion_core::{Event, TimerId, WorkflowId, status_from_events};
use std::collections::HashMap;
use std::sync::Arc;

use aion_store::{ReadableEventStore, StoreError, TimerEntry};
use chrono::{DateTime, Utc};

use crate::engine_seam::EngineSeamError;
use crate::time::timer_service::{
    RetireAttempt, TimerDisposition, armed_fire_at_in_active_segment,
    timer_disposition_in_active_segment,
};
use crate::time::{TimerService, TimerServiceError, is_deadline_timer};

/// Recovery service for durable timers that elapsed outside the live wheel path.
pub struct TimerRecovery {
    store: Arc<dyn ReadableEventStore>,
    timer_service: Arc<TimerService>,
    /// Workflows already warned about as orphans (rows with no usable
    /// history, or gone from the engine), so a permanent orphan produces ONE
    /// warning, not one per row per sweep. The rows themselves are still
    /// counted by every sweep — `skipped_orphans` in the summary line is the
    /// standing gauge of the population this sweep cannot explain.
    warned_orphans: std::sync::Mutex<std::collections::HashSet<WorkflowId>>,
}

/// Errors returned by [`TimerRecovery`].
#[derive(thiserror::Error, Debug, Clone, PartialEq, Eq)]
pub enum TimerRecoveryError {
    /// Durable timer polling failed.
    #[error("timer recovery store operation failed: {0}")]
    Store(#[from] StoreError),

    /// Recovered timer firing failed.
    #[error("timer recovery fire operation failed: {0}")]
    Timer(#[from] TimerServiceError),
}

impl TimerRecovery {
    /// Creates the boot/adoption timer-recovery sweep service.
    ///
    /// There is deliberately no interval or clock parameter: the sweep runs
    /// only when the engine boots or adopts a shard, and the sweep instant is
    /// the caller's `now` argument to [`Self::recover_on_startup`].
    #[must_use]
    pub fn new(store: Arc<dyn ReadableEventStore>, timer_service: Arc<TimerService>) -> Self {
        Self {
            store,
            timer_service,
            warned_orphans: std::sync::Mutex::new(std::collections::HashSet::new()),
        }
    }

    /// Runs the boot/adoption recovery sweep for timers due as of `now` — the
    /// ONLY sweep this landing has (engine startup and `adopt_shards` both
    /// drive it; nothing runs it periodically).
    ///
    /// One history read per workflow serves the WHOLE startup sweep: the same
    /// read re-arms the workflow's outstanding future timers on the wheel and
    /// disposes its due rows ([`Self::dispose_due_rows`]). Workflows outside
    /// the active set that still hold due rows (terminal, paused, or unknown
    /// to the store) are read once each after the active pass.
    ///
    /// Returns the sweep's [`SweepCounts`] — every due row lands in exactly
    /// one of its counters, so their sum equals the due-row population.
    ///
    /// # Errors
    ///
    /// Returns [`TimerRecoveryError`] when polling expired timers, reading a
    /// history, re-arming a future timer, or firing a due timer fails.
    pub async fn recover_on_startup(
        &self,
        now: DateTime<Utc>,
    ) -> Result<SweepCounts, TimerRecoveryError> {
        let sweep_started = std::time::Instant::now();
        let due_by_workflow = group_by_workflow(self.store.expired_timers(now).await?);
        let expired_rows: usize = due_by_workflow.values().map(Vec::len).sum();
        tracing::info!(
            expired_rows,
            due_workflows = due_by_workflow.len(),
            "timer recovery startup sweep started"
        );
        let mut counts = SweepCounts::default();
        let mut rearmed = 0usize;
        let result = self
            .startup_passes(now, due_by_workflow, &mut counts, &mut rearmed)
            .await;
        // The summary is emitted on BOTH exits: on the boot where the sweep
        // dies, the operator needs how far it got — a missing line cannot
        // distinguish a half-completed sweep from one that never started.
        let elapsed_ms = u64::try_from(sweep_started.elapsed().as_millis()).unwrap_or(u64::MAX);
        match &result {
            Ok(()) => tracing::info!(
                fired = counts.fired,
                redelivered = counts.redelivered,
                retired = counts.retired,
                superseded = counts.superseded,
                retire_failures = counts.retire_failures,
                skipped_orphans = counts.skipped_orphans,
                rearmed,
                elapsed_ms,
                "timer recovery startup sweep finished"
            ),
            Err(error) => tracing::warn!(
                fired = counts.fired,
                redelivered = counts.redelivered,
                retired = counts.retired,
                superseded = counts.superseded,
                retire_failures = counts.retire_failures,
                skipped_orphans = counts.skipped_orphans,
                rearmed,
                elapsed_ms,
                %error,
                "timer recovery startup sweep ABORTED; counters cover the work \
                 completed before the failure"
            ),
        }
        result.map(|()| counts)
    }

    /// The startup sweep's two passes (active re-arm + disposition, then
    /// leftover disposition), split out so the caller can emit the summary
    /// line on the error exit too.
    async fn startup_passes(
        &self,
        now: DateTime<Utc>,
        mut due_by_workflow: HashMap<WorkflowId, Vec<TimerEntry>>,
        counts: &mut SweepCounts,
        rearmed: &mut usize,
    ) -> Result<(), TimerRecoveryError> {
        for workflow_id in self.store.list_active().await? {
            let history = self.store.read_history(&workflow_id).await?;
            for (timer_id, fire_at, armed_seq) in outstanding_future_timers(&history, now) {
                self.timer_service
                    .schedule(workflow_id.clone(), timer_id, fire_at, armed_seq)
                    .await?;
                *rearmed += 1;
            }
            if let Some(entries) = due_by_workflow.remove(&workflow_id) {
                self.dispose_due_rows(&workflow_id, &history, entries, counts)
                    .await?;
            }
        }
        for (workflow_id, entries) in due_by_workflow {
            let history = self.store.read_history(&workflow_id).await?;
            self.dispose_due_rows(&workflow_id, &history, entries, counts)
                .await?;
        }
        Ok(())
    }

    /// Warn about an orphaned workflow exactly once per process lifetime.
    ///
    /// Returns `true` when this call was the first sighting. A poisoned set
    /// fails OPEN (warn again) — deduplication is a courtesy, never a reason
    /// to suppress an operator signal.
    fn first_orphan_sighting(&self, workflow_id: &WorkflowId) -> bool {
        self.warned_orphans
            .lock()
            .map_or(true, |mut warned| warned.insert(workflow_id.clone()))
    }

    /// Dispose one workflow's due rows from a single already-read history.
    ///
    /// - No history at all: the store holds rows for a workflow it has no
    ///   events for. Conservative orphan shape — skip and count, never retire
    ///   on a history that answers nothing.
    /// - Projected TERMINAL: every due row is moot (no fire can ever record),
    ///   so the rows retire in bulk without entering the fire path. This is
    ///   what empties the boot walk for completed workflows.
    /// - Otherwise, per row: a live arming WHOSE ROW MATCHES the recorded
    ///   arming's `fire_at` — or a recorded non-deadline fire whose mailbox
    ///   wake may still be owed (aion#145) — goes through
    ///   [`TimerService::fire_timer`], which owns terminal filtering, the
    ///   fire guard, recording, delivery, and the consumed row's retirement.
    ///   A live arming whose row DISAGREES with history is a stale row
    ///   (round-2 F1) and retires without firing. Consumed rows (fired
    ///   deadline, cancelled, absent from the active segment) retire
    ///   directly.
    async fn dispose_due_rows(
        &self,
        workflow_id: &WorkflowId,
        history: &[Event],
        entries: Vec<TimerEntry>,
        counts: &mut SweepCounts,
    ) -> Result<(), TimerRecoveryError> {
        if history.is_empty() {
            counts.skipped_orphans += entries.len();
            if self.first_orphan_sighting(workflow_id) {
                tracing::warn!(
                    %workflow_id,
                    rows = entries.len(),
                    "skipping due timer rows for a workflow with no recorded history \
                     (orphaned rows); nothing is fired and nothing is retired — \
                     counted in every sweep's skipped_orphans, warned once"
                );
            }
            return Ok(());
        }
        if status_from_events(history).is_terminal() {
            for entry in entries {
                tracing::debug!(
                    %workflow_id,
                    timer_id = %entry.timer_id,
                    fire_at = %entry.fire_at,
                    "retiring due timer row of a terminal workflow"
                );
                let attempt = self
                    .timer_service
                    .retire_consumed_row(
                        workflow_id,
                        &entry.timer_id,
                        entry.fire_at,
                        entry.armed_seq,
                    )
                    .await;
                counts.record_row_retirement(attempt);
            }
            return Ok(());
        }
        for entry in entries {
            match timer_disposition_in_active_segment(history, &entry.timer_id) {
                // A live arming fires ONLY when the row IS that arming
                // (round-2 F1): the row's `fire_at` must equal the recorded
                // arming's. A mismatched row predates the current arming —
                // the workflow recorded a new `TimerStarted` and died before
                // the replacement row write — and firing it would mint a
                // durable `TimerFired` for an instant the workflow never
                // armed: a three-month sleep returning immediately,
                // permanent in history. The stale row retires instead
                // (fire_at-conditionally, so if the startup re-arm pass has
                // already rewritten the row with the armed value, the
                // replacement survives as Superseded) and the recorded
                // arming's own row is restored by the re-arm pass when its
                // fire_at is still in the future.
                TimerDisposition::Live
                    if armed_fire_at_in_active_segment(history, &entry.timer_id)
                        == Some(entry.fire_at) =>
                {
                    self.fire_due(workflow_id, &entry, counts).await?;
                }
                TimerDisposition::Live => {
                    tracing::warn!(
                        %workflow_id,
                        timer_id = %entry.timer_id,
                        row_fire_at = %entry.fire_at,
                        armed_fire_at = ?armed_fire_at_in_active_segment(history, &entry.timer_id),
                        "due timer row disagrees with the recorded arming; retiring the \
                         stale row instead of firing it"
                    );
                    let attempt = self
                        .timer_service
                        .retire_consumed_row(
                            workflow_id,
                            &entry.timer_id,
                            entry.fire_at,
                            entry.armed_seq,
                        )
                        .await;
                    counts.record_row_retirement(attempt);
                }
                // A surviving row whose fire is already recorded IS the missing
                // acknowledgement (aion#145): retirement follows delivery on
                // the fire path, so a row that outlived its recorded fire says
                // the wake may never have landed. Deadlines never record
                // `TimerFired` through this path, so a fired deadline owes
                // nothing.
                TimerDisposition::Fired if !is_deadline_timer(&entry.timer_id) => {
                    self.redeliver_surviving_fired_row(workflow_id, &entry, counts)
                        .await?;
                }
                TimerDisposition::Fired
                | TimerDisposition::Cancelled
                | TimerDisposition::Absent => {
                    tracing::debug!(
                        %workflow_id,
                        timer_id = %entry.timer_id,
                        fire_at = %entry.fire_at,
                        "retiring consumed timer row without entering the fire path"
                    );
                    let attempt = self
                        .timer_service
                        .retire_consumed_row(
                            workflow_id,
                            &entry.timer_id,
                            entry.fire_at,
                            entry.armed_seq,
                        )
                        .await;
                    counts.record_row_retirement(attempt);
                }
            }
        }
        Ok(())
    }

    /// One surviving already-fired row's owed redelivery (aion#145), then its
    /// retirement — served from the sweep's already-read history, never a
    /// per-row read, so each such row costs this once, ever.
    async fn redeliver_surviving_fired_row(
        &self,
        workflow_id: &WorkflowId,
        entry: &TimerEntry,
        counts: &mut SweepCounts,
    ) -> Result<(), TimerRecoveryError> {
        match self
            .timer_service
            .redeliver_owed_wake(
                workflow_id.clone(),
                entry.timer_id.clone(),
                entry.fire_at,
                entry.armed_seq,
            )
            .await
        {
            // A wake was delivered: the row counts as redelivered whatever
            // its retirement answered (a failed retire is already warned at
            // the retire site and heals at the next boot or adoption sweep —
            // wake-idempotent by the seam's design).
            Ok((true, _)) => counts.redelivered += 1,
            Ok((false, attempt)) => counts.record_row_retirement(attempt),
            // `UnknownWorkflow` from the redelivery seam after residency
            // answered `Resident` means the run left the registry between the
            // two lookups — it terminated or was torn down mid-redelivery.
            // That is the NON-RESIDENT shape arriving late: no live wake is
            // deliverable, replay on any future residency restore consumes
            // the recorded fire, and the arming's row retires
            // (fire_at-conditionally). Leaving the row instead would
            // re-attempt it at every boot or adoption sweep forever.
            Err(TimerServiceError::Engine(EngineSeamError::UnknownWorkflow {
                workflow_id: gone,
            })) => {
                tracing::info!(
                    workflow_id = %gone,
                    timer_id = %entry.timer_id,
                    "workflow left residency mid-redelivery; retiring the \
                     consumed row without a wake"
                );
                let attempt = self
                    .timer_service
                    .retire_consumed_row(
                        workflow_id,
                        &entry.timer_id,
                        entry.fire_at,
                        entry.armed_seq,
                    )
                    .await;
                counts.record_row_retirement(attempt);
            }
            Err(other) => return Err(other.into()),
        }
        Ok(())
    }

    /// One due row through the fire path, with the orphan shape preserved.
    async fn fire_due(
        &self,
        workflow_id: &WorkflowId,
        entry: &TimerEntry,
        counts: &mut SweepCounts,
    ) -> Result<(), TimerRecoveryError> {
        match self
            .timer_service
            .fire_timer(workflow_id.clone(), entry.timer_id.clone(), entry.fire_at)
            .await
        {
            Ok(()) => counts.fired += 1,
            // An orphaned timer whose workflow no longer exists — e.g. the
            // workflow was cancelled and purged from the engine's known set —
            // must never abort recovery or block engine startup. The workflow
            // is gone, so the timer is moot: log it and skip. (A terminal but
            // still-known workflow's timer is already filtered inside
            // `fire_timer`'s liveness check, which returns `Ok` without firing.)
            Err(TimerServiceError::Engine(EngineSeamError::UnknownWorkflow { workflow_id })) => {
                if self.first_orphan_sighting(&workflow_id) {
                    tracing::warn!(
                        %workflow_id,
                        timer_id = %entry.timer_id,
                        "skipping recovered timer for unknown workflow (orphaned timer); \
                         the workflow no longer exists — counted in every sweep's \
                         skipped_orphans, warned once"
                    );
                }
                counts.skipped_orphans += 1;
            }
            Err(other) => return Err(other.into()),
        }
        Ok(())
    }
}

/// What one sweep did, for the summary line and the specimens.
///
/// Every due row lands in EXACTLY ONE counter, so their sum equals the
/// sweep's `expired_rows`: `fired` and `redelivered` claim their rows
/// whatever the internal housekeeping answered, and every other row is
/// counted by what actually happened to it — a deletion (`retired`), a
/// surviving replacement arming (`superseded`), a store refusal
/// (`retire_failures`), or an unexplainable owner (`skipped_orphans`).
/// `retired` therefore measures deletions, never attempts.
#[derive(Clone, Copy, Debug, Default)]
pub struct SweepCounts {
    /// Due rows that completed the fire path as live fires.
    pub fired: usize,
    /// Surviving already-recorded rows whose owed wake was redelivered
    /// (aion#145) before the row retired.
    pub redelivered: usize,
    /// Rows whose retirement DELETED the row (or found it already gone).
    pub retired: usize,
    /// Rows whose key a replacement arming owns; the new row was left
    /// standing and this sweep is done with the old arming forever.
    pub superseded: usize,
    /// Rows whose retirement the store refused (each already warned with its
    /// cause at the retire site); the rows survive until the next boot or
    /// adoption sweep.
    pub retire_failures: usize,
    /// Rows skipped because their workflow is unknown (to the store or to
    /// the engine) — the conservative orphan shape.
    pub skipped_orphans: usize,
}

impl SweepCounts {
    /// Count one wake-less row by its retirement outcome.
    fn record_row_retirement(&mut self, attempt: RetireAttempt) {
        match attempt {
            RetireAttempt::Retired => self.retired += 1,
            RetireAttempt::Superseded => self.superseded += 1,
            RetireAttempt::Failed => self.retire_failures += 1,
        }
    }
}

/// Bucket due rows by their owning workflow, so one history read serves all
/// of a workflow's rows.
fn group_by_workflow(entries: Vec<TimerEntry>) -> HashMap<WorkflowId, Vec<TimerEntry>> {
    let mut by_workflow: HashMap<WorkflowId, Vec<TimerEntry>> = HashMap::new();
    for entry in entries {
        by_workflow
            .entry(entry.workflow_id.clone())
            .or_default()
            .push(entry);
    }
    by_workflow
}

fn outstanding_future_timers(
    history: &[Event],
    now: DateTime<Utc>,
) -> Vec<(TimerId, DateTime<Utc>, u64)> {
    let mut outstanding: HashMap<TimerId, (DateTime<Utc>, u64)> = HashMap::new();
    for event in history {
        match event {
            Event::TimerStarted {
                envelope,
                timer_id,
                fire_at,
            } => {
                outstanding.insert(timer_id.clone(), (*fire_at, envelope.seq));
            }
            Event::TimerFired { timer_id, .. } | Event::TimerCancelled { timer_id, .. } => {
                outstanding.remove(timer_id);
            }
            _ => {}
        }
    }
    outstanding
        .into_iter()
        .filter(|(_, (fire_at, _))| *fire_at > now)
        .map(|(timer_id, (fire_at, armed_seq))| (timer_id, fire_at, armed_seq))
        .collect()
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use aion_core::{Event, EventEnvelope, RunId, TimerCancelCause, TimerId, WorkflowId};
    use aion_store::{
        InMemoryStore, ReadableEventStore, StoreError, WritableEventStore, WriteToken,
    };
    use chrono::{DateTime, Utc};

    use super::{TimerRecovery, TimerRecoveryError, outstanding_future_timers};
    use crate::engine_seam::test_support::{DeliveredWorkflowMessage, FakeEngineHandle};
    use crate::engine_seam::{
        EngineHandle, EngineSeamError, WorkflowProcessHandle, WorkflowResidency,
    };
    use crate::time::TimerService;
    use crate::time::deadline_timer_id;

    #[derive(Debug, thiserror::Error)]
    enum TestError {
        #[error(transparent)]
        Recovery(#[from] TimerRecoveryError),

        #[error(transparent)]
        Store(#[from] StoreError),

        #[error(transparent)]
        Engine(#[from] EngineSeamError),
    }

    fn instant(offset_seconds: i64) -> DateTime<Utc> {
        DateTime::from_timestamp(1_700_000_000 + offset_seconds, 0).unwrap_or_default()
    }

    fn recorded_at() -> DateTime<Utc> {
        instant(1)
    }

    fn workflow_id() -> WorkflowId {
        WorkflowId::new_v4()
    }

    fn timer_id(sequence: u64) -> TimerId {
        TimerId::anonymous(sequence)
    }

    fn recovery() -> (Arc<InMemoryStore>, Arc<FakeEngineHandle>, TimerRecovery) {
        let concrete_store = Arc::new(InMemoryStore::default());
        let writable: Arc<dyn WritableEventStore> = concrete_store.clone();
        let readable: Arc<dyn ReadableEventStore> = concrete_store.clone();
        let engine = Arc::new(FakeEngineHandle::recording_to(writable));
        let timer_service = Arc::new(TimerService::with_recorded_at(
            engine.clone(),
            readable.clone(),
            recorded_at,
        ));
        let recovery = TimerRecovery::new(readable, timer_service);
        (concrete_store, engine, recovery)
    }

    async fn history(
        store: &InMemoryStore,
        workflow_id: &WorkflowId,
    ) -> Result<Vec<Event>, StoreError> {
        store.read_history(workflow_id).await
    }

    /// A `TimerStarted` arming event. `fire_at` is explicit so every fixture
    /// keeps the recorded arming and the durable row in agreement — the
    /// boot sweep fires a due row only when the two match (round-2 F1).
    fn timer_started_event(
        workflow_id: &WorkflowId,
        timer_id: &TimerId,
        seq: u64,
        fire_at: DateTime<Utc>,
    ) -> Event {
        Event::TimerStarted {
            envelope: EventEnvelope {
                seq,
                recorded_at: instant(0),
                workflow_id: workflow_id.clone(),
            },
            timer_id: timer_id.clone(),
            fire_at,
        }
    }

    fn workflow_started_event(workflow_id: &WorkflowId, seq: u64) -> Event {
        Event::WorkflowStarted {
            envelope: EventEnvelope {
                seq,
                recorded_at: instant(0),
                workflow_id: workflow_id.clone(),
            },
            workflow_type: "fixture".to_owned(),
            input: aion_core::Payload::new(aion_core::ContentType::Json, b"null".to_vec()),
            run_id: RunId::new_v4(),
            parent_run_id: None,
            parent_workflow_id: None,
            package_version: aion_core::PackageVersion::new("a".repeat(64)),
        }
    }

    fn count_timer_fired(events: &[Event], timer_id: &TimerId) -> usize {
        events
            .iter()
            .filter(|event| {
                matches!(event, Event::TimerFired { timer_id: recorded, .. } if recorded == timer_id)
            })
            .count()
    }

    #[tokio::test]
    async fn startup_sweep_fires_past_timer_and_delivers() -> Result<(), TestError> {
        let process = WorkflowProcessHandle::new(42);
        let (store, engine, recovery) = recovery();
        let workflow_id = workflow_id();
        let timer_id = timer_id(1);
        let fire_at = instant(10);
        engine.set_residency(workflow_id.clone(), WorkflowResidency::Resident(process))?;
        engine.record_workflow_event(
            &workflow_id,
            timer_started_event(&workflow_id, &timer_id, 1, fire_at),
        )?;
        store
            .schedule_timer(&workflow_id, &timer_id, fire_at, 1)
            .await?;

        let recovered = recovery.recover_on_startup(instant(20)).await?;

        assert_eq!(recovered.fired, 1);
        assert_eq!(
            count_timer_fired(&history(&store, &workflow_id).await?, &timer_id),
            1
        );
        assert_eq!(
            engine.delivered_messages()?,
            vec![(
                process,
                DeliveredWorkflowMessage::TimerFired {
                    timer_id: timer_id.clone(),
                    fire_at
                }
            )]
        );
        Ok(())
    }

    #[tokio::test]
    async fn startup_sweep_does_not_fire_future_timer() -> Result<(), TestError> {
        let process = WorkflowProcessHandle::new(42);
        let (store, engine, recovery) = recovery();
        let workflow_id = workflow_id();
        let timer_id = timer_id(2);
        engine.set_residency(workflow_id.clone(), WorkflowResidency::Resident(process))?;
        store
            .schedule_timer(&workflow_id, &timer_id, instant(30), 1)
            .await?;

        let recovered = recovery.recover_on_startup(instant(20)).await?;

        assert_eq!(recovered.fired, 0);
        assert_eq!(
            count_timer_fired(&history(&store, &workflow_id).await?, &timer_id),
            0
        );
        assert!(engine.delivered_messages()?.is_empty());
        Ok(())
    }

    /// Round-2 F1, the boot specimen from the ruling: a STALE past-due row
    /// must never fire against a re-armed FUTURE arming.
    ///
    /// The crash shape: the workflow recorded `TimerStarted(T, Y three months
    /// out)` and died before the replacement row write, so the store still
    /// holds the PRIOR arming's row `(T, X past)`. Pre-fix, boot fired the
    /// stale row — a durable `TimerFired` with the recorded arming still
    /// three months out: the sleep returned immediately and the fabricated
    /// event was permanent in history. Post-fix the sweep compares the row
    /// against `armed_fire_at_in_active_segment` and retires the mismatch
    /// (counted), while the startup re-arm pass restores the row to the
    /// armed value — the keyspace converges to what history says.
    ///
    /// Mutation-witnessed: with the comparison broken (Live always fires),
    /// the premature `TimerFired` appears and this test goes red.
    #[tokio::test]
    async fn a_stale_past_due_row_is_retired_not_fired_against_a_rearmed_future_arming()
    -> Result<(), TestError> {
        let process = WorkflowProcessHandle::new(42);
        let (store, engine, recovery) = recovery();
        let workflow_id = workflow_id();
        let timer_id = timer_id(11);
        let stale_row_fire_at = instant(5); // past at sweep time
        let armed_fire_at = instant(300); // future at sweep time
        engine.set_residency(workflow_id.clone(), WorkflowResidency::Resident(process))?;
        // Active workflow (the startup re-arm pass walks `list_active`).
        engine.record_workflow_event(&workflow_id, workflow_started_event(&workflow_id, 1))?;
        // The recorded arming is the FUTURE one...
        engine.record_workflow_event(
            &workflow_id,
            timer_started_event(&workflow_id, &timer_id, 2, armed_fire_at),
        )?;
        // ...but the durable row still carries the PRIOR arming (the crash
        // landed between the `TimerStarted` append and the row write).
        store
            .schedule_timer(&workflow_id, &timer_id, stale_row_fire_at, 1)
            .await?;

        let recovered = recovery.recover_on_startup(instant(20)).await?;

        assert_eq!(recovered.fired, 0, "a stale row must never fire");
        assert_eq!(
            count_timer_fired(&history(&store, &workflow_id).await?, &timer_id),
            0,
            "no premature TimerFired may be fabricated for the future arming"
        );
        assert!(
            engine.delivered_messages()?.is_empty(),
            "no wake may be delivered for a stale row"
        );
        // The stale row was disposed as a retirement (the startup re-arm pass
        // rewrote the row to the armed value FIRST, so the conditional retire
        // of the stale arming answers Superseded — the replacement survives).
        assert_eq!(
            recovered.superseded, 1,
            "the stale row lands in the superseded bin: the re-arm pass's \
             replacement row won the key"
        );
        assert_eq!(
            recovered.fired
                + recovered.redelivered
                + recovered.retired
                + recovered.superseded
                + recovered.retire_failures
                + recovered.skipped_orphans,
            1,
            "the one due row lands in exactly one counter"
        );
        // The keyspace converged to the RECORDED arming: nothing is due any
        // more, and the row now carries the armed future fire_at.
        assert!(
            store.expired_timers(instant(20)).await?.is_empty(),
            "the stale past-due row is gone"
        );
        let rows = store.expired_timers(armed_fire_at).await?;
        assert_eq!(rows.len(), 1, "the armed row survives: {rows:?}");
        assert_eq!(
            rows[0].fire_at, armed_fire_at,
            "the row converged to the recorded arming's fire_at"
        );
        Ok(())
    }

    /// Round-2 F5: the `retire_failures` bin, exercised POSITIVELY. A store
    /// that refuses a consumed row's retirement must land that row in
    /// `retire_failures` (never silently in `retired`, never an aborted
    /// sweep), the partition must still sum to the population, and the row
    /// must SURVIVE the refusal so the next boot or adoption sweep heals it.
    /// Mutation-sensitive: counting the refusal as `retired` breaks the bin
    /// asserts; aborting the sweep on the refusal breaks the Ok return; and
    /// deleting the row anyway breaks the healing sweep's `retired == 1`.
    #[tokio::test]
    async fn a_refused_retirement_lands_in_retire_failures_and_heals_at_the_next_sweep()
    -> Result<(), TestError> {
        use crate::store_faults::FlakyStore;

        let flaky = Arc::new(FlakyStore::new());
        let writable: Arc<dyn WritableEventStore> = flaky.clone();
        let readable: Arc<dyn ReadableEventStore> = flaky.clone();
        let engine = Arc::new(FakeEngineHandle::recording_to(writable));
        let timer_service = Arc::new(TimerService::with_recorded_at(
            engine.clone(),
            readable.clone(),
            recorded_at,
        ));
        let recovery = TimerRecovery::new(readable, timer_service);

        // A consumed row: the timer was cancelled (its retirement is what the
        // sweep owes), and the row survived the original cancel.
        let workflow_id = workflow_id();
        let timer_id = timer_id(12);
        engine.record_workflow_event(
            &workflow_id,
            timer_started_event(&workflow_id, &timer_id, 1, instant(5)),
        )?;
        engine.record_workflow_event(
            &workflow_id,
            Event::TimerCancelled {
                cause: TimerCancelCause::WorkflowIntent,
                envelope: EventEnvelope {
                    seq: 2,
                    recorded_at: instant(6),
                    workflow_id: workflow_id.clone(),
                },
                timer_id: timer_id.clone(),
            },
        )?;
        flaky
            .schedule_timer(&workflow_id, &timer_id, instant(5), 1)
            .await?;

        flaky.fail_next_retirements(1);
        let refused = recovery.recover_on_startup(instant(20)).await?;

        assert_eq!(
            refused.retire_failures, 1,
            "the refused retirement must be counted as a failure, not a deletion"
        );
        assert_eq!(refused.retired, 0);
        assert_eq!(
            refused.fired
                + refused.redelivered
                + refused.retired
                + refused.superseded
                + refused.retire_failures
                + refused.skipped_orphans,
            1,
            "the refused row still lands in exactly one counter"
        );
        assert_eq!(
            flaky.expired_timers(instant(20)).await?.len(),
            1,
            "the row survives the refusal — nothing is silently dropped"
        );

        // The healer: the next boot or adoption sweep retires the survivor.
        let healed = recovery.recover_on_startup(instant(20)).await?;
        assert_eq!(healed.retired, 1, "the next sweep retires the survivor");
        assert_eq!(healed.retire_failures, 0);
        assert!(
            flaky.expired_timers(instant(20)).await?.is_empty(),
            "the row is gone once the store accepts the retirement"
        );
        Ok(())
    }

    /// Re-pointed from the removed `tick()` surface (round-2 F3): the sweep's
    /// per-sweep counters, driven through the ONLY entrypoint this landing
    /// has. The first boot sweep fires the due arming; the second finds
    /// nothing due, because retirement is the wake's acknowledgement — the
    /// fire path retires the row AFTER delivering, so no re-walk and no
    /// duplicate wake. (Before retirement existed, the row survived and every
    /// sweep redelivered forever.) A wake genuinely lost after recording
    /// leaves its row alive, and the surviving-row specimens below prove that
    /// row is redelivered then retired.
    #[tokio::test]
    async fn a_second_boot_sweep_after_a_fired_row_retires_finds_nothing_due()
    -> Result<(), TestError> {
        let process = WorkflowProcessHandle::new(42);
        let (store, engine, recovery) = recovery();
        let workflow_id = workflow_id();
        let timer_id = timer_id(3);
        let fire_at = instant(25);
        engine.set_residency(workflow_id.clone(), WorkflowResidency::Resident(process))?;
        engine.record_workflow_event(
            &workflow_id,
            timer_started_event(&workflow_id, &timer_id, 1, fire_at),
        )?;
        store
            .schedule_timer(&workflow_id, &timer_id, fire_at, 1)
            .await?;

        assert_eq!(recovery.recover_on_startup(instant(30)).await?.fired, 1);
        assert_eq!(recovery.recover_on_startup(instant(30)).await?.fired, 0);

        assert_eq!(
            count_timer_fired(&history(&store, &workflow_id).await?, &timer_id),
            1,
            "the durable TimerFired is recorded exactly once across repeated sweeps"
        );
        assert_eq!(engine.delivered_messages()?.len(), 1);
        Ok(())
    }

    #[tokio::test]
    async fn running_startup_sweep_twice_records_due_timer_once_total() -> Result<(), TestError> {
        let process = WorkflowProcessHandle::new(42);
        let (store, engine, recovery) = recovery();
        let workflow_id = workflow_id();
        let timer_id = timer_id(4);
        let fire_at = instant(10);
        engine.set_residency(workflow_id.clone(), WorkflowResidency::Resident(process))?;
        engine.record_workflow_event(
            &workflow_id,
            timer_started_event(&workflow_id, &timer_id, 1, fire_at),
        )?;
        store
            .schedule_timer(&workflow_id, &timer_id, fire_at, 1)
            .await?;

        recovery.recover_on_startup(instant(20)).await?;
        recovery.recover_on_startup(instant(20)).await?;

        assert_eq!(
            count_timer_fired(&history(&store, &workflow_id).await?, &timer_id),
            1,
            "the durable TimerFired is recorded exactly once across repeated sweeps"
        );
        // The first sweep's fire retired the row (retirement follows the
        // delivered wake), so the second sweep finds nothing due and delivers
        // nothing — see the surviving-row specimens for the lost-wake arm.
        assert_eq!(engine.delivered_messages()?.len(), 1);
        Ok(())
    }

    #[tokio::test]
    async fn cancelled_timer_is_never_fired_by_recovery() -> Result<(), TestError> {
        let process = WorkflowProcessHandle::new(42);
        let (store, engine, recovery) = recovery();
        let workflow_id = workflow_id();
        let timer_id = timer_id(5);
        let fire_at = instant(10);
        engine.set_residency(workflow_id.clone(), WorkflowResidency::Resident(process))?;
        store
            .schedule_timer(&workflow_id, &timer_id, fire_at, 1)
            .await?;
        engine.record_workflow_event(
            &workflow_id,
            Event::TimerCancelled {
                cause: aion_core::TimerCancelCause::WorkflowIntent,
                envelope: EventEnvelope {
                    seq: 1,
                    recorded_at: instant(9),
                    workflow_id: workflow_id.clone(),
                },
                timer_id: timer_id.clone(),
            },
        )?;

        recovery.recover_on_startup(instant(20)).await?;

        assert_eq!(
            count_timer_fired(&history(&store, &workflow_id).await?, &timer_id),
            0
        );
        assert!(engine.delivered_messages()?.is_empty());
        Ok(())
    }

    /// D5 resurrection hazard: `outstanding_future_timers` is whole-history
    /// scoped, so a continue-as-new predecessor's still-outstanding
    /// `deadline:{run}` WOULD be re-armed after failover — firing a timeout
    /// against a run that already continued. The `WorkflowIntent` cancel recorded
    /// at the CAN terminal closes exactly that hole. This proves both halves at
    /// the precise mechanism the scout flagged.
    #[test]
    fn cancelled_predecessor_deadline_is_not_rearmed_after_continue_as_new() {
        let workflow_id = workflow_id();
        let predecessor_run = RunId::new_v4();
        let deadline = deadline_timer_id(&predecessor_run).unwrap_or_else(|_| timer_id(0));
        let now = instant(0);
        let fire_at = instant(120); // future: eligible for re-arm

        let started = |seq: u64, run: &RunId| Event::WorkflowStarted {
            envelope: EventEnvelope {
                seq,
                recorded_at: instant(0),
                workflow_id: workflow_id.clone(),
            },
            workflow_type: "sleeper".to_owned(),
            input: aion_core::Payload::new(aion_core::ContentType::Json, b"null".to_vec()),
            run_id: run.clone(),
            parent_run_id: None,
            parent_workflow_id: None,
            package_version: aion_core::PackageVersion::new("a".repeat(64)),
        };
        let deadline_started = Event::TimerStarted {
            envelope: EventEnvelope {
                seq: 2,
                recorded_at: instant(0),
                workflow_id: workflow_id.clone(),
            },
            timer_id: deadline.clone(),
            fire_at,
        };
        let continued = Event::WorkflowContinuedAsNew {
            envelope: EventEnvelope {
                seq: 3,
                recorded_at: instant(1),
                workflow_id: workflow_id.clone(),
            },
            input: aion_core::Payload::new(aion_core::ContentType::Json, b"null".to_vec()),
            workflow_type: None,
            parent_run_id: predecessor_run.clone(),
        };

        // Before the cancel: the hazard is real — the predecessor's future
        // deadline is outstanding across the whole history even after CAN.
        let uncancelled = vec![
            started(1, &predecessor_run),
            deadline_started.clone(),
            continued.clone(),
        ];
        assert!(
            outstanding_future_timers(&uncancelled, now)
                .into_iter()
                .any(|(timer_id, _, _)| timer_id == deadline),
            "an uncancelled predecessor deadline WOULD be re-armed after failover"
        );

        // With the WorkflowIntent cancel recorded at the CAN terminal: closed.
        let cancelled = vec![
            started(1, &predecessor_run),
            deadline_started,
            continued,
            Event::TimerCancelled {
                envelope: EventEnvelope {
                    seq: 4,
                    recorded_at: instant(1),
                    workflow_id: workflow_id.clone(),
                },
                timer_id: deadline.clone(),
                cause: TimerCancelCause::WorkflowIntent,
            },
        ];
        assert!(
            !outstanding_future_timers(&cancelled, now)
                .into_iter()
                .any(|(timer_id, _, _)| timer_id == deadline),
            "the WorkflowIntent cancel closes the whole-history re-arm hole"
        );
    }

    #[tokio::test]
    async fn orphaned_timer_for_unknown_workflow_is_skipped_not_fatal() -> Result<(), TestError> {
        // Regression: a durable timer whose workflow was cancelled and purged
        // from the engine's known set must NOT abort startup recovery. Before the
        // fix, `fire_timer`'s `UnknownWorkflow` error propagated and bricked engine
        // startup — observed in production after a restart:
        //   "timer recovery fire operation failed: ... workflow <id> is unknown".
        let (store, engine, recovery) = recovery();
        let workflow_id = workflow_id();
        let timer_id = timer_id(6);
        let fire_at = instant(10);

        // The timer is live in history (started, never fired/cancelled) and
        // its row matches the recorded arming, so the sweep routes it into
        // the fire path (the F1 stale-row gate must not divert it) ...
        store
            .schedule_timer(&workflow_id, &timer_id, fire_at, 1)
            .await?;
        engine.record_workflow_event(
            &workflow_id,
            timer_started_event(&workflow_id, &timer_id, 1, fire_at),
        )?;
        // ... but the workflow itself is gone: the engine rejects the recovered
        // fire with `UnknownWorkflow` (exactly what the real engine does when a
        // cancelled workflow's record has been purged).
        engine.push_record_response(Err(EngineSeamError::UnknownWorkflow {
            workflow_id: workflow_id.clone(),
        }))?;

        // Recovery must SUCCEED by skipping the orphan, not error out.
        let recovered = recovery.recover_on_startup(instant(20)).await?;

        assert_eq!(
            recovered.fired, 0,
            "the orphaned timer is skipped, not fired"
        );
        assert_eq!(
            recovered.skipped_orphans, 1,
            "the engine-side orphan is counted in skipped_orphans (round-3 N5)"
        );
        assert_eq!(
            count_timer_fired(&history(&store, &workflow_id).await?, &timer_id),
            0,
            "no TimerFired is recorded for an unknown workflow"
        );
        assert!(
            engine.delivered_messages()?.is_empty(),
            "nothing is delivered for an unknown workflow"
        );
        Ok(())
    }

    /// A read-counting decorator: the R2 acceptance instrument. The grouped
    /// sweep's read complexity is pinned against the store's own surface —
    /// `read_history` calls counted at the seam — never inferred from timing.
    struct CountingStore {
        inner: Arc<InMemoryStore>,
        history_reads: std::sync::atomic::AtomicUsize,
    }

    impl CountingStore {
        fn new(inner: Arc<InMemoryStore>) -> Self {
            Self {
                inner,
                history_reads: std::sync::atomic::AtomicUsize::new(0),
            }
        }

        fn history_reads(&self) -> usize {
            self.history_reads.load(std::sync::atomic::Ordering::SeqCst)
        }
    }

    #[async_trait::async_trait]
    impl ReadableEventStore for CountingStore {
        async fn read_history(&self, workflow_id: &WorkflowId) -> Result<Vec<Event>, StoreError> {
            self.history_reads
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            self.inner.read_history(workflow_id).await
        }

        async fn read_history_from(
            &self,
            workflow_id: &WorkflowId,
            from_seq: u64,
        ) -> Result<Vec<Event>, StoreError> {
            self.inner.read_history_from(workflow_id, from_seq).await
        }

        async fn read_run_chain(
            &self,
            workflow_id: &WorkflowId,
        ) -> Result<Vec<aion_store::RunSummary>, StoreError> {
            self.inner.read_run_chain(workflow_id).await
        }

        async fn list_workflow_ids(&self) -> Result<Vec<WorkflowId>, StoreError> {
            self.inner.list_workflow_ids().await
        }

        async fn list_active(&self) -> Result<Vec<WorkflowId>, StoreError> {
            self.inner.list_active().await
        }

        async fn list_paused(&self) -> Result<Vec<WorkflowId>, StoreError> {
            self.inner.list_paused().await
        }

        async fn query(
            &self,
            filter: &aion_core::WorkflowFilter,
        ) -> Result<Vec<aion_core::WorkflowSummary>, StoreError> {
            self.inner.query(filter).await
        }

        async fn schedule_timer(
            &self,
            workflow_id: &WorkflowId,
            timer_id: &TimerId,
            fire_at: DateTime<Utc>,
            armed_seq: u64,
        ) -> Result<(), StoreError> {
            self.inner
                .schedule_timer(workflow_id, timer_id, fire_at, armed_seq)
                .await
        }

        async fn expired_timers(
            &self,
            as_of: DateTime<Utc>,
        ) -> Result<Vec<aion_store::TimerEntry>, StoreError> {
            self.inner.expired_timers(as_of).await
        }

        async fn retire_timer(
            &self,
            workflow_id: &WorkflowId,
            timer_id: &TimerId,
            fire_at: DateTime<Utc>,
            armed_seq: u64,
        ) -> Result<aion_store::TimerRetirement, StoreError> {
            self.inner
                .retire_timer(workflow_id, timer_id, fire_at, armed_seq)
                .await
        }
    }

    /// The counting fixture: the wrapper is BOTH the recovery's store and the
    /// timer service's, so every history read either sweep makes is counted.
    fn counting_recovery() -> (
        Arc<InMemoryStore>,
        Arc<CountingStore>,
        Arc<FakeEngineHandle>,
        TimerRecovery,
    ) {
        let inner = Arc::new(InMemoryStore::default());
        let counting = Arc::new(CountingStore::new(inner.clone()));
        let writable: Arc<dyn WritableEventStore> = inner.clone();
        let readable: Arc<dyn ReadableEventStore> = counting.clone();
        let engine = Arc::new(FakeEngineHandle::recording_to(writable));
        let timer_service = Arc::new(TimerService::with_recorded_at(
            engine.clone(),
            readable.clone(),
            recorded_at,
        ));
        let recovery = TimerRecovery::new(readable, timer_service);
        (inner, counting, engine, recovery)
    }

    fn timer_fired_event(workflow_id: &WorkflowId, timer_id: &TimerId, seq: u64) -> Event {
        Event::TimerFired {
            envelope: EventEnvelope {
                seq,
                recorded_at: instant(0),
                workflow_id: workflow_id.clone(),
            },
            timer_id: timer_id.clone(),
        }
    }

    /// Seed `count` CONSUMED armings (started at `instant(5)`, then fired,
    /// with the surviving durable row the incident left behind) for the
    /// estate-shaped specimen, returning the last history seq used.
    async fn seed_consumed_armings(
        engine: &FakeEngineHandle,
        store: &InMemoryStore,
        workflow_id: &WorkflowId,
        count: u64,
    ) -> Result<u64, TestError> {
        let mut seq = 0;
        for i in 0..count {
            let consumed = timer_id(i);
            seq += 1;
            let armed_seq = seq;
            engine.record_workflow_event(
                workflow_id,
                timer_started_event(workflow_id, &consumed, armed_seq, instant(5)),
            )?;
            seq += 1;
            engine.record_workflow_event(
                workflow_id,
                timer_fired_event(workflow_id, &consumed, seq),
            )?;
            store
                .schedule_timer(workflow_id, &consumed, instant(5), armed_seq)
                .await?;
        }
        Ok(seq)
    }

    /// The estate-shaped R2 acceptance specimen: hundreds of CONSUMED rows and
    /// one live arming across two workflows — one running, one terminal.
    ///
    /// Pinned facts, in the order they killed the estate boot:
    /// - history reads scale with DISTINCT WORKFLOWS (plus the fire path's own
    ///   two for the single live fire), never with rows. The pre-collapse
    ///   sweep put every consumed row through the fire path — two reads per
    ///   row, thousands of reads per boot, the 90-minute startup;
    /// - exactly one fire records and delivers;
    /// - the sweep leaves the expired-timers index EMPTY: consumed rows are
    ///   retired in bulk, the terminal workflow's rows without ever computing
    ///   a per-row disposition, and the fired arming by the fire path itself.
    #[tokio::test]
    async fn a_grouped_sweep_reads_once_per_workflow_and_retires_the_backlog()
    -> Result<(), TestError> {
        const CONSUMED_PER_WORKFLOW: u64 = 300;
        let process = WorkflowProcessHandle::new(42);
        let (inner, counting, engine, recovery) = counting_recovery();

        // Workflow A: running, resident, 300 consumed armings and ONE live
        // due arming.
        let running = workflow_id();
        engine.set_residency(running.clone(), WorkflowResidency::Resident(process))?;
        let mut seq =
            seed_consumed_armings(&engine, &inner, &running, CONSUMED_PER_WORKFLOW).await?;
        let live = timer_id(9_999);
        seq += 1;
        engine.record_workflow_event(
            &running,
            timer_started_event(&running, &live, seq, instant(5)),
        )?;
        inner
            .schedule_timer(&running, &live, instant(5), seq)
            .await?;

        // Workflow B: 300 consumed armings, then a terminal event — its rows
        // are moot whatever their per-timer dispositions say.
        let finished = workflow_id();
        let seq = seed_consumed_armings(&engine, &inner, &finished, CONSUMED_PER_WORKFLOW).await?;
        engine.record_workflow_event(
            &finished,
            Event::WorkflowCompleted {
                envelope: EventEnvelope {
                    seq: seq + 1,
                    recorded_at: instant(6),
                    workflow_id: finished.clone(),
                },
                result: aion_core::Payload::new(aion_core::ContentType::Json, b"null".to_vec()),
            },
        )?;

        let recovered = recovery.recover_on_startup(instant(20)).await?;

        assert_eq!(recovered.fired, 1, "exactly the one live arming fires");
        // Every one of the 601 due rows lands in exactly one counter: the one
        // live fire, the resident workflow's 300 redelivered rows, and the
        // terminal workflow's 300 bulk retirements. Nothing superseded,
        // nothing failed, nothing orphaned — and the sum IS the population,
        // so a row silently double-counted or dropped breaks this line.
        assert_eq!(
            recovered.redelivered, 300,
            "one redelivery per surviving fired row"
        );
        assert_eq!(
            recovered.retired, 300,
            "the terminal workflow's rows bulk-retire"
        );
        assert_eq!(recovered.superseded, 0);
        assert_eq!(recovered.retire_failures, 0);
        assert_eq!(recovered.skipped_orphans, 0);
        assert_eq!(
            recovered.fired
                + recovered.redelivered
                + recovered.retired
                + recovered.superseded
                + recovered.retire_failures
                + recovered.skipped_orphans,
            601,
            "the counters partition the due-row population exactly"
        );
        assert_eq!(
            count_timer_fired(&history(&inner, &running).await?, &live),
            1,
            "the live arming records its fire exactly once"
        );
        // The RESIDENT workflow's 300 surviving already-fired rows each owe
        // one redelivered wake (aion#145: a row that outlived its recorded
        // fire says the wake may never have landed — retirement is the ack).
        // Each is duplicate-safe and paid ONCE ever: the rows retire behind
        // it. The terminal workflow's 300 rows owe nothing.
        assert_eq!(
            engine.delivered_messages()?.len(),
            301,
            "one live-fire wake plus one owed-wake redelivery per surviving \
             fired row of the resident workflow"
        );
        assert_eq!(
            count_timer_fired(&history(&inner, &running).await?, &timer_id(0)),
            1,
            "a redelivered wake appends nothing: the recorder seam answers \
             AlreadyRecorded and the original fire stays the only record"
        );
        // The counting store sees the SWEEP'S OWN reads: one grouped read per
        // distinct workflow, plus the fire path's own two (its service-layer
        // disposition gate and its envelope read) for the single fire. 601
        // rows, four reads — the pre-collapse shape was two reads per ROW.
        // Out of this instrument's frame, DELIBERATELY: the redelivery seam's
        // history read under the recorder lock. The fake engine stands in for
        // that seam, so production's first boot after a backlog still pays
        // roughly one recorder-seam round trip per surviving row — once,
        // ever, per row (the module doc's honest cost model). This assertion
        // discriminates the sweep's own read scaling, nothing more.
        assert_eq!(
            counting.history_reads(),
            4,
            "the sweep's own history reads must scale with workflows (2) plus \
             the fire path's own reads (2 for 1 fire), never with the 601 rows"
        );
        assert!(
            inner.expired_timers(instant(20)).await?.is_empty(),
            "the sweep must leave the expired index EMPTY: every consumed row \
             retired in bulk and the fired arming retired by the fire path"
        );
        Ok(())
    }

    /// The conservative orphan shape at the STORE: rows whose workflow has no
    /// recorded history at all are skipped and SURVIVE — never fired, never
    /// retired on a history that answers nothing.
    #[tokio::test]
    async fn rows_with_no_history_are_skipped_and_survive() -> Result<(), TestError> {
        let (inner, _counting, engine, recovery) = counting_recovery();
        let orphaned = workflow_id();
        for i in 0..3 {
            inner
                .schedule_timer(&orphaned, &timer_id(i), instant(5), 1)
                .await?;
        }

        let recovered = recovery.recover_on_startup(instant(20)).await?;

        assert_eq!(
            recovered.fired, 0,
            "nothing fires for a workflow with no history"
        );
        assert_eq!(
            recovered.skipped_orphans, 3,
            "every orphaned row is COUNTED — the summary line's standing gauge \
             of the population the sweep cannot explain (round-3 N5)"
        );
        assert_eq!(
            inner.expired_timers(instant(20)).await?.len(),
            3,
            "orphaned rows survive the sweep — skip and count, never retire on \
             a history that answers nothing"
        );
        assert!(engine.delivered_messages()?.is_empty());
        Ok(())
    }

    /// aion#145, post-retirement carrier: a row that OUTLIVED its recorded
    /// fire is the missing acknowledgement (the fire path retires only after
    /// delivering), so the sweep redelivers the owed wake once — from its
    /// grouped read, appending nothing — and the retirement behind it keeps
    /// every later sweep quiet.
    #[tokio::test]
    async fn a_surviving_recorded_fire_row_is_redelivered_once_then_retires()
    -> Result<(), TestError> {
        let process = WorkflowProcessHandle::new(42);
        let (store, engine, recovery) = recovery();
        let workflow_id = workflow_id();
        let timer_id = timer_id(7);
        engine.set_residency(workflow_id.clone(), WorkflowResidency::Resident(process))?;
        // The ack-loss shape: the fire recorded durably, but the wake (and the
        // retirement that follows it) never happened — the row survives.
        engine.record_workflow_event(
            &workflow_id,
            timer_started_event(&workflow_id, &timer_id, 1, instant(5)),
        )?;
        engine
            .record_workflow_event(&workflow_id, timer_fired_event(&workflow_id, &timer_id, 2))?;
        store
            .schedule_timer(&workflow_id, &timer_id, instant(5), 1)
            .await?;

        assert_eq!(
            recovery.recover_on_startup(instant(20)).await?.fired,
            0,
            "a redelivery is not a fire"
        );
        assert_eq!(
            engine.delivered_messages()?.len(),
            1,
            "the owed wake is delivered exactly once"
        );
        assert_eq!(
            count_timer_fired(&history(&store, &workflow_id).await?, &timer_id),
            1,
            "redelivery appends nothing — the recorder seam answers AlreadyRecorded"
        );
        assert!(
            store.expired_timers(instant(20)).await?.is_empty(),
            "the redelivered row retires — the wake's acknowledgement is durable now"
        );

        // Paid once, ever: the next sweep is quiet.
        assert_eq!(recovery.recover_on_startup(instant(20)).await?.fired, 0);
        assert_eq!(engine.delivered_messages()?.len(), 1);
        Ok(())
    }

    /// The non-resident arm of the same shape: no live wake is owed (replay on
    /// residency restore consumes the recorded fire), so the row just retires.
    /// This is the arm the 2026-08-24 boot walked 1,434 times without ever
    /// emptying.
    #[tokio::test]
    async fn a_surviving_recorded_fire_row_of_a_nonresident_workflow_retires_silently()
    -> Result<(), TestError> {
        let (store, engine, recovery) = recovery();
        let workflow_id = workflow_id();
        let timer_id = timer_id(8);
        engine.record_workflow_event(
            &workflow_id,
            timer_started_event(&workflow_id, &timer_id, 1, instant(5)),
        )?;
        engine
            .record_workflow_event(&workflow_id, timer_fired_event(&workflow_id, &timer_id, 2))?;
        store
            .schedule_timer(&workflow_id, &timer_id, instant(5), 1)
            .await?;

        assert_eq!(recovery.recover_on_startup(instant(20)).await?.fired, 0);

        assert!(
            engine.delivered_messages()?.is_empty(),
            "no wake is owed to a non-resident workflow"
        );
        assert_eq!(
            count_timer_fired(&history(&store, &workflow_id).await?, &timer_id),
            1,
            "nothing is appended for a non-resident redelivery"
        );
        assert!(
            store.expired_timers(instant(20)).await?.is_empty(),
            "the consumed row retires instead of surviving every boot"
        );
        Ok(())
    }

    /// The stale-oracle interleave: the sweep's history snapshot says `Fired`,
    /// but by redelivery time the workflow has RE-ARMED the same timer name —
    /// the engine's recorded events are ahead of the swept snapshot. The
    /// redelivery seam, answering under the recorder's authority, must say
    /// `NotOwed`: NO second `TimerFired` is minted for the new arming (the
    /// premature-fire defect this pin exists for), no wake is delivered, and
    /// the OLD arming's row retires while the new arming's claim is untouched.
    #[tokio::test]
    async fn a_redelivery_for_a_rearmed_timer_appends_nothing_and_wakes_nobody()
    -> Result<(), TestError> {
        let concrete_store = Arc::new(InMemoryStore::default());
        let readable: Arc<dyn ReadableEventStore> = concrete_store.clone();
        // The fake is deliberately NOT wired to the store: its recorded
        // events model the recorder's (ahead) view, the store models the
        // sweep's (stale) snapshot.
        let engine = Arc::new(FakeEngineHandle::new());
        let timer_service = Arc::new(TimerService::with_recorded_at(
            engine.clone(),
            readable.clone(),
            recorded_at,
        ));
        let recovery = TimerRecovery::new(readable, timer_service);

        let workflow_id = workflow_id();
        let timer_id = timer_id(3);
        engine.set_residency(
            workflow_id.clone(),
            WorkflowResidency::Resident(WorkflowProcessHandle::new(7)),
        )?;
        // The recorder's view: fired, then RE-ARMED (`TimerStarted` again).
        engine.record_workflow_event(
            &workflow_id,
            timer_started_event(&workflow_id, &timer_id, 1, instant(5)),
        )?;
        engine
            .record_workflow_event(&workflow_id, timer_fired_event(&workflow_id, &timer_id, 2))?;
        engine.record_workflow_event(
            &workflow_id,
            timer_started_event(&workflow_id, &timer_id, 3, instant(5)),
        )?;
        // The sweep's view: history stops at the fire, and the old arming's
        // row still stands in the expired index.
        concrete_store
            .append(
                WriteToken::recorder(),
                &workflow_id,
                &[
                    timer_started_event(&workflow_id, &timer_id, 1, instant(5)),
                    timer_fired_event(&workflow_id, &timer_id, 2),
                ],
                0,
            )
            .await?;
        concrete_store
            .schedule_timer(&workflow_id, &timer_id, instant(5), 1)
            .await?;

        let recovered = recovery.recover_on_startup(instant(20)).await?;

        assert_eq!(recovered.fired, 0, "a stale row is never a live fire");
        assert!(
            engine.delivered_messages()?.is_empty(),
            "no wake may reach a workflow that already ran past the recorded fire"
        );
        let recorded: Vec<Event> = engine
            .recorded_events()?
            .into_iter()
            .map(|(_, event)| event)
            .collect();
        assert_eq!(
            count_timer_fired(&recorded, &timer_id),
            1,
            "the redelivery must NOT mint a premature `TimerFired` for the \
             re-armed timer — the original fire stays the only record"
        );
        assert!(
            concrete_store.expired_timers(instant(20)).await?.is_empty(),
            "the OLD arming's row retires; nothing re-walks it"
        );
        Ok(())
    }
}