use crate::executor::DlqPolicy;
use crate::runtime::InFlightMap;
use crate::storage::{QueueStorageRuntime, RuntimeStorage};
use awa_model::audited_sql;
use awa_model::cron::{
    atomic_enqueue, list_cron_jobs, upsert_cron_job, CronJobRow, CronMissedFirePolicy,
};
#[cfg(test)]
use awa_model::SkipReason;
use awa_model::{
    JobRow, JobState, PeriodicJob, PruneOutcome, RotateOutcome, TerminalDeltaRollupOutcome,
};
use chrono::Utc;
use croner::Cron;
use sqlx::pool::PoolConnection;
use sqlx::{PgPool, Postgres};
use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};
use uuid::Uuid;

/// The canonical unique-claim conflict raised by `awa.sync_job_unique_claims`
/// (SQLSTATE 23505, constraint reported as `idx_awa_jobs_unique`). See #388.
fn is_unique_claim_conflict(err: &awa_model::AwaError) -> bool {
    match err {
        awa_model::AwaError::Database(sqlx::Error::Database(db_err)) => {
            db_err.code().as_deref() == Some("23505")
        }
        _ => false,
    }
}

/// Canonical rescue statements write the physical tables directly — the
/// same `DELETE` from `awa.jobs_hot` + `INSERT` into `awa.scheduled_jobs`
/// move the `awa.jobs` view's canonical UPDATE branch performs. The view
/// itself is unusable here: once routing has flipped it exposes (and
/// raises on writes to) only the queue-storage plane, while stuck
/// canonical rows must keep being rescued until finalize (#456). The
/// table triggers keep unique claims and admin dirty keys in sync, so the
/// #388 unique-claim conflict semantics are unchanged.
///
/// Single-row variants of the batched canonical rescue statements, used by
/// the #388 fallback. Bodies mirror the batch statements exactly; only the
/// WHERE clause narrows to one id (re-checking the staleness predicate so a
/// row that recovered between selection and update is left alone).
const HEARTBEAT_RESCUE_PER_ROW_SQL: &str = r#"
    WITH candidates AS (
        SELECT id, attempt, max_attempts FROM awa.jobs_hot
        WHERE id = $1
          AND state = 'running'
          AND heartbeat_at < now() - ($2 * interval '1 millisecond')
        FOR UPDATE
    ),
    failed AS (
        UPDATE awa.jobs_hot
        SET state = 'failed',
            finalized_at = now(),
            heartbeat_at = NULL,
            deadline_at = NULL,
            errors = errors || jsonb_build_object(
                'error', 'heartbeat stale: worker presumed dead',
                'attempt', attempt,
                'at', now()
            )::jsonb
        WHERE id IN (SELECT id FROM candidates WHERE attempt >= max_attempts)
        RETURNING *
    ),
    deleted AS (
        DELETE FROM awa.jobs_hot
        WHERE id IN (SELECT id FROM candidates WHERE attempt < max_attempts)
        RETURNING *
    ),
    moved AS (
        INSERT INTO awa.scheduled_jobs (
            id, kind, queue, args, state, priority, attempt, max_attempts,
            run_at, heartbeat_at, deadline_at, attempted_at, finalized_at,
            created_at, errors, metadata, tags, unique_key, unique_states,
            callback_id, callback_timeout_at, callback_filter,
            callback_on_complete, callback_on_fail, callback_transform,
            run_lease, progress
        )
        SELECT
            id, kind, queue, args, 'retryable', priority, attempt, max_attempts,
            run_at, NULL, NULL, attempted_at, now(),
            created_at,
            errors || jsonb_build_object(
                'error', 'heartbeat stale: worker presumed dead',
                'attempt', attempt,
                'at', now()
            )::jsonb,
            metadata, tags, unique_key, unique_states,
            NULL, NULL, NULL, NULL, NULL, NULL,
            run_lease, progress
        FROM deleted
        RETURNING *
    )
    SELECT * FROM failed
    UNION ALL
    SELECT * FROM moved
"#;

const DEADLINE_RESCUE_PER_ROW_SQL: &str = r#"
    WITH candidates AS (
        SELECT id, attempt, max_attempts FROM awa.jobs_hot
        WHERE id = $1
          AND state = 'running'
          AND deadline_at IS NOT NULL
          AND deadline_at < now()
        FOR UPDATE
    ),
    failed AS (
        UPDATE awa.jobs_hot
        SET state = 'failed',
            finalized_at = now(),
            heartbeat_at = NULL,
            deadline_at = NULL,
            errors = errors || jsonb_build_object(
                'error', 'hard deadline exceeded',
                'attempt', attempt,
                'at', now()
            )::jsonb
        WHERE id IN (SELECT id FROM candidates WHERE attempt >= max_attempts)
        RETURNING *
    ),
    deleted AS (
        DELETE FROM awa.jobs_hot
        WHERE id IN (SELECT id FROM candidates WHERE attempt < max_attempts)
        RETURNING *
    ),
    moved AS (
        INSERT INTO awa.scheduled_jobs (
            id, kind, queue, args, state, priority, attempt, max_attempts,
            run_at, heartbeat_at, deadline_at, attempted_at, finalized_at,
            created_at, errors, metadata, tags, unique_key, unique_states,
            callback_id, callback_timeout_at, callback_filter,
            callback_on_complete, callback_on_fail, callback_transform,
            run_lease, progress
        )
        SELECT
            id, kind, queue, args, 'retryable', priority, attempt, max_attempts,
            run_at, NULL, NULL, attempted_at, now(),
            created_at,
            errors || jsonb_build_object(
                'error', 'hard deadline exceeded',
                'attempt', attempt,
                'at', now()
            )::jsonb,
            metadata, tags, unique_key, unique_states,
            NULL, NULL, NULL, NULL, NULL, NULL,
            run_lease, progress
        FROM deleted
        RETURNING *
    )
    SELECT * FROM failed
    UNION ALL
    SELECT * FROM moved
"#;

const CALLBACK_RESCUE_PER_ROW_SQL: &str = r#"
    WITH candidate AS (
        SELECT id, attempt, max_attempts FROM awa.jobs_hot
        WHERE id = $1
          AND state = 'waiting_external'
          AND callback_timeout_at IS NOT NULL
          AND callback_timeout_at < now()
        FOR UPDATE
    ),
    failed AS (
        UPDATE awa.jobs_hot
        SET state = 'failed',
            finalized_at = now(),
            callback_id = NULL,
            callback_timeout_at = NULL,
            callback_filter = NULL,
            callback_on_complete = NULL,
            callback_on_fail = NULL,
            callback_transform = NULL,
            errors = errors || jsonb_build_object(
                'error', 'callback timed out',
                'attempt', attempt,
                'at', now()
            )::jsonb
        WHERE id IN (SELECT id FROM candidate WHERE attempt >= max_attempts)
        RETURNING *
    ),
    deleted AS (
        DELETE FROM awa.jobs_hot
        WHERE id IN (SELECT id FROM candidate WHERE attempt < max_attempts)
        RETURNING *
    ),
    moved AS (
        INSERT INTO awa.scheduled_jobs (
            id, kind, queue, args, state, priority, attempt, max_attempts,
            run_at, heartbeat_at, deadline_at, attempted_at, finalized_at,
            created_at, errors, metadata, tags, unique_key, unique_states,
            callback_id, callback_timeout_at, callback_filter,
            callback_on_complete, callback_on_fail, callback_transform,
            run_lease, progress
        )
        SELECT
            id, kind, queue, args, 'retryable', priority, attempt, max_attempts,
            now() + awa.backoff_duration(attempt, max_attempts),
            heartbeat_at, deadline_at, attempted_at, now(),
            created_at,
            errors || jsonb_build_object(
                'error', 'callback timed out',
                'attempt', attempt,
                'at', now()
            )::jsonb,
            metadata, tags, unique_key, unique_states,
            NULL, NULL, NULL, NULL, NULL, NULL,
            run_lease, progress
        FROM deleted
        RETURNING *
    )
    SELECT * FROM failed
    UNION ALL
    SELECT * FROM moved
"#;

/// How often the leader scans for attempts past their deadline by default.
pub(crate) const DEFAULT_DEADLINE_RESCUE_INTERVAL: Duration = Duration::from_secs(30);

/// Per-queue or global retention policy for completed and failed/cancelled jobs.
#[derive(Debug, Clone)]
pub struct RetentionPolicy {
    /// How long to keep completed jobs before cleanup.
    pub completed: Duration,
    /// How long to keep failed/cancelled jobs before cleanup.
    pub failed: Duration,
    /// How long to keep DLQ rows before cleanup.
    pub dlq: Option<Duration>,
}

impl Default for RetentionPolicy {
    fn default() -> Self {
        Self {
            completed: Duration::from_secs(86400), // 24h
            failed: Duration::from_secs(259200),   // 72h
            dlq: None,
        }
    }
}

/// Per-branch observability state for the maintenance leader's main
/// `tokio::select!` loop. Tracks the most recent body duration so the
/// next fire of the same branch can decide whether it was already
/// overdue, and an `is_delayed` flag so we only log/count one event per
/// "overrun episode" rather than once per tick.
///
/// Wired through `AwaMetrics` — see [`crate::metrics::AwaMetrics`] for the
/// instrument definitions. Issue #242.
#[derive(Debug, Default)]
struct MaintenanceBranchState {
    /// Body duration of the previous run of this branch. `None` until the
    /// branch has run at least once.
    last_duration: Option<Duration>,
    /// True when we've emitted the on-time -> delayed warning/counter
    /// for the current overrun episode. Cleared on the recovery
    /// transition.
    is_delayed: bool,
    /// Number of consecutive overrun observations (last_duration above
    /// the upper threshold). Samples inside the deadband neither
    /// advance this counter nor reset it.
    consecutive_overrun: u32,
    /// Mirror of `consecutive_overrun` for clearly-on-time samples
    /// (last_duration below the lower threshold). Used to gate
    /// delayed -> on-time recovery with the same K-consecutive
    /// requirement.
    consecutive_ontime: u32,
    /// Wall-clock deadline that suppresses the branch body while the
    /// branch is unhealthy: ticks arriving before it are skipped. Armed on
    /// flip-to-delayed AND on every overrun observation while already
    /// delayed, so a branch that keeps failing keeps quietly waiting
    /// instead of hammering the database.
    cooldown_until: Option<Instant>,
}

/// How many consecutive observations the branch tracker requires before
/// flipping `is_delayed` either direction. Combined with the
/// duration-margin thresholds below: a sample inside the deadband
/// (between `LOWER_FACTOR_*` and `UPPER_FACTOR_*` of `tick_interval`)
/// doesn't advance either counter, so the boundary flap from #316's
/// post-mortem stops at the source. A sample outside the deadband
/// still needs `OVERRUN_HYSTERESIS_K` peers on the same side before
/// the state flips.
const OVERRUN_HYSTERESIS_K: u32 = 3;

/// Duration-margin thresholds. Crossing
/// `last_duration > tick_interval * UPPER` counts as an overrun
/// sample; `last_duration < tick_interval * LOWER` counts as an
/// on-time sample. Everything in between is a deadband that leaves
/// both counters untouched.
///
/// `UPPER = 3/2` and `LOWER = 7/10` give a generous deadband — the
/// branch has to be at 70% of its tick or below to count as recovered,
/// and 50% above its tick to count as overrunning. This prevents
/// 51ms-vs-49ms jitter at the 50ms boundary from advancing either
/// counter, which is what bench evidence on #316 showed was still
/// happening with K-only hysteresis. Integer ratios avoid f64 in the
/// hot path.
const OVERRUN_UPPER_NUM: u32 = 3;
const OVERRUN_UPPER_DEN: u32 = 2;
const OVERRUN_LOWER_NUM: u32 = 7;
const OVERRUN_LOWER_DEN: u32 = 10;

/// Cooldown armed after an overrun, as a multiple of the offending body's
/// duration. A branch that merely runs slower than its interval settles at
/// roughly a one-third duty cycle (`D / (D + 2D)`) instead of going
/// near-silent.
const BRANCH_COOLDOWN_BODY_MULTIPLE: u32 = 2;

/// Upper bound on a single cooldown. The cooldown is a wall-clock duration
/// because the leader loop's ticks are not evenly serviced: under CPU
/// starvation a branch's tick is consumed only once many loop iterations
/// have passed, so a tick-count cooldown stretches to many times its nominal
/// length. Five seconds bounds how long a pathologically slow body (minutes
/// long) can silence a branch, and is shorter than the default interval of
/// every rescue and cleanup branch, so at default settings the cooldown has
/// expired by the next tick.
const BRANCH_COOLDOWN_MAX: Duration = Duration::from_secs(5);

/// Interval for one branch of the leader loop. A tick that is serviced late
/// (the loop was busy with another branch, or the runtime was starved)
/// restarts the period from that moment, so a branch never gets back-to-back
/// catch-up ticks that would crowd out the others. `Skip` would realign to
/// the original grid and could fire again almost immediately after a late
/// tick; `Delay` guarantees a full period between consecutive bodies.
fn leader_loop_interval(period: Duration) -> tokio::time::Interval {
    let mut interval = tokio::time::interval(period);
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    interval
}

/// Records the start of one branch body and emits observability when the
/// body returns. Constructed by [`MaintenanceBranchTracker::try_begin`], which
/// also applies the previous-run overrun check before the body starts so
/// the warning/counter line up with the fire moment described in #242.
struct BranchTimer<'a> {
    tracker: &'a MaintenanceBranchTracker,
    branch: &'static str,
    metrics: &'a crate::metrics::AwaMetrics,
    started_at: Instant,
}

impl<'a> BranchTimer<'a> {
    /// Stop the timer, record the duration into both the per-branch
    /// histogram and the tracker's state. Must be called once after the
    /// body returns — there is no `Drop` impl so a panic-mid-body would
    /// leave the histogram un-recorded, which is fine: a panicking
    /// maintenance branch is a bigger story than the missing sample.
    fn finish(self) {
        let duration = self.started_at.elapsed();
        self.metrics
            .record_maintenance_branch_duration(self.branch, duration);
        self.tracker.record_finish(self.branch, duration);
    }
}

/// Owns the per-branch overrun state for the maintenance leader loop.
/// All access is logically single-threaded — the maintenance loop is the
/// only caller — but the loop runs inside a `tokio::spawn`-ed task that
/// requires `Send`, so we use `std::sync::Mutex` rather than `RefCell`.
/// The mutex is uncontended in practice (one acquirer); cost is one
/// uncontended lock per branch tick.
#[derive(Default)]
struct MaintenanceBranchTracker {
    branches: std::sync::Mutex<HashMap<&'static str, MaintenanceBranchState>>,
}

impl MaintenanceBranchTracker {
    fn new() -> Self {
        Self {
            branches: std::sync::Mutex::new(HashMap::new()),
        }
    }

    /// Call at the moment a branch arm fires, before running its body.
    /// Applies cooldown + duration-margin hysteresis, then either:
    ///   * returns `Some(BranchTimer)` so the caller runs the body
    ///     and calls `.finish()`, or
    ///   * returns `None` to signal "skip this tick" — the body
    ///     must not run, and no duration sample is recorded for this
    ///     tick (so the K-on-time counter doesn't advance for skipped
    ///     work).
    ///
    /// Two gating phases:
    ///
    /// 1. **Cooldown.** If `cooldown_until` is in the future, return
    ///    `None`. The deadline is wall-clock, set on flip-to-delayed and
    ///    re-armed on every subsequent overrun observation, so an
    ///    unhealthy branch backs off instead of beating on the database
    ///    every tick.
    /// 2. **Duration-margin hysteresis.** Compare `last_duration` to
    ///    `tick_interval * UPPER/LOWER`. Outside the deadband, advance
    ///    the matching K-counter; inside, leave both alone. Crossing
    ///    K-consecutive on either side flips `is_delayed` and may
    ///    arm cooldown.
    fn try_begin<'a>(
        &'a self,
        branch: &'static str,
        tick_interval: Duration,
        metrics: &'a crate::metrics::AwaMetrics,
    ) -> Option<BranchTimer<'a>> {
        self.try_begin_at(branch, tick_interval, metrics, true, Instant::now())
    }

    /// Like [`Self::try_begin`] for latency-critical, cheap, idempotent
    /// branches that must run on every tick: overruns are still tracked,
    /// warned about and counted, but never suppress the body.
    fn try_begin_without_cooldown<'a>(
        &'a self,
        branch: &'static str,
        tick_interval: Duration,
        metrics: &'a crate::metrics::AwaMetrics,
    ) -> Option<BranchTimer<'a>> {
        self.try_begin_at(branch, tick_interval, metrics, false, Instant::now())
    }

    fn try_begin_at<'a>(
        &'a self,
        branch: &'static str,
        tick_interval: Duration,
        metrics: &'a crate::metrics::AwaMetrics,
        arm_cooldown: bool,
        now: Instant,
    ) -> Option<BranchTimer<'a>> {
        let mut branches = self
            .branches
            .lock()
            .expect("maintenance branch tracker mutex");
        let state = branches.entry(branch).or_default();

        // Phase 1: cooldown gate. Purely wall-clock; the branch stays
        // quiet until the deadline passes, however many ticks that takes.
        match state.cooldown_until {
            Some(until) if now < until => return None,
            Some(_) => state.cooldown_until = None,
            None => {}
        }

        // Phase 2: duration-margin + K-consecutive hysteresis.
        // `take()` consumes the sample so we evaluate each body's
        // duration exactly once. Without this, when cooldown returns
        // None (no body runs → no new sample), the next post-cooldown
        // `try_begin` would re-read the same pre-flip overrun sample,
        // see `consecutive_overrun >= K` already, and re-arm cooldown
        // forever. The branch would never run its body again until
        // the worker restarts. Codex + CodeRabbit both flagged this
        // on PR #318.
        //
        // Cooldowns are paced by the offending body's own duration so a
        // consistently slow branch keeps a bounded duty cycle (~1/3).
        let cooldown_for = |last_duration: Duration| -> Duration {
            (last_duration * BRANCH_COOLDOWN_BODY_MULTIPLE).min(BRANCH_COOLDOWN_MAX)
        };
        if let Some(last_duration) = state.last_duration.take() {
            // Integer-ratio thresholds — avoid f64 in the hot path.
            let upper_threshold = tick_interval * OVERRUN_UPPER_NUM / OVERRUN_UPPER_DEN;
            let lower_threshold = tick_interval * OVERRUN_LOWER_NUM / OVERRUN_LOWER_DEN;

            if last_duration > upper_threshold {
                state.consecutive_overrun = state.consecutive_overrun.saturating_add(1);
                state.consecutive_ontime = 0;
                let cross_threshold = state.consecutive_overrun >= OVERRUN_HYSTERESIS_K;
                if cross_threshold && !state.is_delayed {
                    state.is_delayed = true;
                    let cooldown = arm_cooldown.then(|| cooldown_for(last_duration));
                    state.cooldown_until = cooldown.map(|cooldown| now + cooldown);
                    warn!(
                        branch,
                        last_duration_ms = last_duration.as_millis() as u64,
                        tick_interval_ms = tick_interval.as_millis() as u64,
                        upper_threshold_ms = upper_threshold.as_millis() as u64,
                        consecutive_overrun = state.consecutive_overrun,
                        cooldown_ms = cooldown.map_or(0, |cooldown| cooldown.as_millis() as u64),
                        "maintenance branch overran tick interval",
                    );
                    metrics.record_maintenance_branch_overrun(branch);
                    if cooldown.is_some() {
                        return None;
                    }
                } else if cross_threshold && state.is_delayed && arm_cooldown {
                    // Already delayed and another overrun arrived —
                    // re-arm a duration-paced cooldown but stay quiet
                    // about it.
                    state.cooldown_until = Some(now + cooldown_for(last_duration));
                    return None;
                }
            } else if last_duration < lower_threshold {
                state.consecutive_ontime = state.consecutive_ontime.saturating_add(1);
                state.consecutive_overrun = 0;
                if state.consecutive_ontime >= OVERRUN_HYSTERESIS_K && state.is_delayed {
                    state.is_delayed = false;
                    warn!(
                        branch,
                        last_duration_ms = last_duration.as_millis() as u64,
                        tick_interval_ms = tick_interval.as_millis() as u64,
                        lower_threshold_ms = lower_threshold.as_millis() as u64,
                        consecutive_ontime = state.consecutive_ontime,
                        "maintenance branch recovered to on-time",
                    );
                }
            } else {
                // Deadband — neither counter advances. This is the
                // explicit no-op that kills the 49ms-vs-51ms flap.
            }
        }
        drop(branches);
        Some(BranchTimer {
            tracker: self,
            branch,
            metrics,
            started_at: Instant::now(),
        })
    }

    /// Internal — called by [`BranchTimer::finish`] to stash the body's
    /// wall-clock duration so the next fire of the same branch can apply
    /// the overrun check.
    fn record_finish(&self, branch: &'static str, duration: Duration) {
        let mut branches = self
            .branches
            .lock()
            .expect("maintenance branch tracker mutex");
        let state = branches.entry(branch).or_default();
        state.last_duration = Some(duration);
    }

    /// Test helper — read the current state for one branch. Not used in
    /// production code paths.
    #[cfg(test)]
    fn snapshot(&self, branch: &'static str) -> Option<(Option<Duration>, bool)> {
        let branches = self
            .branches
            .lock()
            .expect("maintenance branch tracker mutex");
        branches
            .get(branch)
            .map(|state| (state.last_duration, state.is_delayed))
    }

    /// Test-only — read the per-branch cooldown deadline and consecutive
    /// counters so tests can assert on the full state machine.
    #[cfg(test)]
    fn cooldown_snapshot(&self, branch: &'static str) -> Option<(Option<Instant>, u32, u32)> {
        let branches = self
            .branches
            .lock()
            .expect("maintenance branch tracker mutex");
        branches.get(branch).map(|state| {
            (
                state.cooldown_until,
                state.consecutive_overrun,
                state.consecutive_ontime,
            )
        })
    }
}

/// Per-segment exponential backoff for the prune step of the queue /
/// lease / claim ring-rotation branches. The rotate step is always
/// cheap (cursor advance under an advisory lock); the prune step is
/// what hurts under pinned MVCC: every tick attempts a best-effort child
/// exclusive child lock and liveness checks.
/// Under a pinned snapshot the child can't be reclaimed, so prune
/// returns `SkippedActive` repeatedly. This tracker skips the next
/// 2^level ticks (capped at 32) after `SkippedActive`, doubling on each
/// repeat. `Blocked` is transient lock contention and retries next tick;
/// it and a successful `Pruned` reset to no backoff. `Noop` (nothing to
/// consider) leaves state unchanged — backoff is for "I tried and couldn't",
/// not "there's nothing to do."
#[derive(Default)]
struct PruneBackoffTracker {
    branches: std::sync::Mutex<HashMap<&'static str, PruneBackoffState>>,
}

#[derive(Debug, Default)]
struct PruneBackoffState {
    /// Number of upcoming ticks to skip the prune call entirely.
    /// Decremented on every `should_skip` poll; the tick where it
    /// reaches 0 actually runs prune again.
    skip_remaining: u32,
    /// Last backoff exponent applied. Used to set `skip_remaining` to
    /// `1 << level` on the next failure. Reset to 0 on `Pruned`.
    backoff_level: u8,
    /// Last destructive prune committed by this maintenance leader. An idle
    /// cursor derives its oldest sealed target and skips only on an exact
    /// slot/generation match.
    last_pruned: Option<(i32, i64)>,
}

/// Cap on the backoff exponent. `2^5 = 32` ticks; at the 1000ms default
/// `lease_rotate_interval` that is ~32s between prune attempts under
/// sustained pin pressure. Long enough to cut the per-tick scan cost
/// dramatically; short enough that prune resumes promptly once the
/// snapshot is released.
const MAX_PRUNE_BACKOFF_LEVEL: u8 = 5;

impl PruneBackoffTracker {
    fn new() -> Self {
        Self::default()
    }

    /// Returns true when the caller should skip this tick's prune.
    /// Side effect: decrements `skip_remaining` when non-zero.
    fn should_skip(&self, branch: &'static str) -> bool {
        let mut branches = self.branches.lock().expect("prune backoff tracker mutex");
        let state = branches.entry(branch).or_default();
        if state.skip_remaining > 0 {
            state.skip_remaining -= 1;
            true
        } else {
            false
        }
    }

    /// Update backoff state: active data backs off, lock contention retries
    /// next tick, successful pruning resets, and no-work outcomes are neutral.
    fn record_outcome(&self, branch: &'static str, outcome: &PruneOutcome) {
        let mut branches = self.branches.lock().expect("prune backoff tracker mutex");
        let state = branches.entry(branch).or_default();
        match outcome {
            PruneOutcome::Pruned {
                slot, generation, ..
            } => {
                state.skip_remaining = 0;
                state.backoff_level = 0;
                state.last_pruned = Some((*slot, *generation));
            }
            PruneOutcome::Blocked { .. } => {
                state.skip_remaining = 0;
                state.backoff_level = 0;
            }
            PruneOutcome::SkippedActive { .. } => {
                state.backoff_level = state
                    .backoff_level
                    .saturating_add(1)
                    .min(MAX_PRUNE_BACKOFF_LEVEL);
                state.skip_remaining = 1u32 << state.backoff_level;
            }
            PruneOutcome::Noop => {}
        }
    }

    fn already_pruned_for_idle_cursor(
        &self,
        branch: &'static str,
        current_slot: i32,
        cursor_generation: i64,
        slot_count: usize,
    ) -> Option<(i32, i64)> {
        let initialized_slots = (cursor_generation + 1).min(slot_count as i64);
        if initialized_slots <= 1 {
            return None;
        }
        let offset = initialized_slots - 1;
        let target_generation = cursor_generation - offset;
        let target_slot = (i64::from(current_slot) - offset).rem_euclid(slot_count as i64) as i32;
        let branches = self.branches.lock().expect("prune backoff tracker mutex");
        let (slot, generation) = branches.get(branch)?.last_pruned?;
        (slot == target_slot && generation == target_generation).then_some((slot, generation))
    }

    #[cfg(test)]
    fn snapshot(&self, branch: &'static str) -> Option<(u32, u8)> {
        let branches = self.branches.lock().expect("prune backoff tracker mutex");
        branches
            .get(branch)
            .map(|state| (state.skip_remaining, state.backoff_level))
    }
}

/// Branch keys used by the prune backoff tracker. Kept as `&'static
/// str` so the tracker's HashMap doesn't have to allocate per call.
const PRUNE_BRANCH_LEASE: &str = "lease";
const PRUNE_BRANCH_CLAIM: &str = "claim";
const PRUNE_BRANCH_QUEUE: &str = "queue";

/// Maintenance service: runs leader-elected background tasks.
///
/// Tasks: heartbeat rescue, deadline rescue, scheduled promotion, cleanup,
/// periodic job sync and evaluation.
pub struct MaintenanceService {
    pool: PgPool,
    metrics: crate::metrics::AwaMetrics,
    cancel: CancellationToken,
    leader: Arc<AtomicBool>,
    alive: Arc<AtomicBool>,
    periodic_jobs: Arc<Vec<PeriodicJob>>,
    /// ADR-029 follow-up specs registry — used to dispatch `Rescued`
    /// follow-ups (best-effort, separate tx) when this service rescues
    /// jobs from stale heartbeat / expired callback / exceeded deadline.
    enqueue_specs: Arc<
        HashMap<
            crate::enqueue_specs::Outcome,
            HashMap<String, Vec<crate::enqueue_specs::BoxedEnqueueSpec>>,
        >,
    >,
    /// In-process `Rescued` lifecycle hooks (ADR-015), shared with the
    /// executor. Best-effort fire-and-forget per rescued job.
    lifecycle_handlers: Arc<HashMap<String, Vec<crate::events::BoxedUntypedEventHandler>>>,
    /// In-flight job cancellation flags — used to signal deadline/heartbeat rescue
    /// to running handlers on this worker instance.
    in_flight: InFlightMap,
    storage: RuntimeStorage,
    /// #456: the configured queue-storage runtime of a canonical-resolved
    /// worker. A pre-flip maintenance leader uses it to promote the
    /// queue-storage deferred backlog once routing has flipped — post-flip
    /// runtimes execute that work but may not hold leadership.
    standby_queue_storage: Option<QueueStorageRuntime>,
    heartbeat_rescue_interval: Duration,
    deadline_rescue_interval: Duration,
    callback_rescue_interval: Duration,
    promote_interval: Duration,
    cleanup_interval: Duration,
    cron_sync_interval: Duration,
    cron_eval_interval: Duration,
    leader_check_interval: Duration,
    leader_election_interval: Duration,
    heartbeat_staleness: Duration,
    completed_retention: Duration,
    failed_retention: Duration,
    cleanup_batch_size: i64,
    queue_retention_overrides: HashMap<String, RetentionPolicy>,
    queue_stats_interval: Duration,
    dlq_retention: Duration,
    dlq_cleanup_batch_size: i64,
    dlq_policy: DlqPolicy,
    dirty_key_recompute_interval: Duration,
    metadata_reconciliation_interval: Duration,
    /// Interval for priority aging — jobs waiting longer than this have their
    /// priority improved by one level per interval elapsed (default: 60s).
    priority_aging_interval: Duration,
    batch_operations_interval: Duration,
    terminal_count_rollup_interval: Duration,
    /// How long a descriptor catalog row can sit without being refreshed
    /// before the maintenance leader deletes it. Zero disables cleanup.
    /// Default: 30 days.
    descriptor_retention: Duration,
    /// #371 staged upgrade: when true, the maintenance leader auto-flips the
    /// ring-cursor authority `columns -> ledger` once the whole fresh fleet
    /// has reported flip-aware (0.7+) continuously for
    /// `ring_authority_auto_flip_stable_period`. Default: true.
    ring_authority_auto_flip: bool,
    /// How long the fleet must be continuously flip-aware (no fresh
    /// NULL-version runtimes) before the leader auto-flips. Default: 10min.
    ring_authority_auto_flip_stable_period: Duration,
    /// Earliest wall-clock instant at which the fleet has been continuously
    /// flip-aware. `None` while any fresh runtime is not flip-aware; reset
    /// whenever a blocking runtime reappears. The auto-flip fires once
    /// `now - since >= stable_period`.
    ring_authority_flip_aware_since: std::sync::Mutex<Option<Instant>>,
}

const PROMOTE_BATCH_SIZE: i64 = 4_096;
const PROMOTE_MAX_BATCHES_PER_TICK: usize = 32;
const CRON_CATCH_UP_LIMIT: usize = 1_000;
const TERMINAL_COUNT_ROLLUP_MAX_SLOTS_PER_TICK: usize = 4;
type QueueStorageMetricRow = (String, i64, i64, i64, i64, i64, i64, i64, Option<f64>);

impl MaintenanceService {
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        pool: PgPool,
        metrics: crate::metrics::AwaMetrics,
        leader: Arc<AtomicBool>,
        alive: Arc<AtomicBool>,
        cancel: CancellationToken,
        periodic_jobs: Arc<Vec<PeriodicJob>>,
        in_flight: InFlightMap,
        storage: RuntimeStorage,
        enqueue_specs: Arc<
            HashMap<
                crate::enqueue_specs::Outcome,
                HashMap<String, Vec<crate::enqueue_specs::BoxedEnqueueSpec>>,
            >,
        >,
        lifecycle_handlers: Arc<HashMap<String, Vec<crate::events::BoxedUntypedEventHandler>>>,
    ) -> Self {
        Self {
            pool,
            metrics,
            cancel,
            leader,
            alive,
            periodic_jobs,
            in_flight,
            storage,
            standby_queue_storage: None,
            enqueue_specs,
            lifecycle_handlers,
            heartbeat_rescue_interval: Duration::from_secs(30),
            deadline_rescue_interval: DEFAULT_DEADLINE_RESCUE_INTERVAL,
            callback_rescue_interval: Duration::from_secs(30),
            promote_interval: Duration::from_millis(250),
            cleanup_interval: Duration::from_secs(60),
            cron_sync_interval: Duration::from_secs(60),
            cron_eval_interval: Duration::from_secs(1),
            leader_check_interval: Duration::from_secs(30),
            leader_election_interval: Duration::from_secs(10),
            heartbeat_staleness: Duration::from_secs(90),
            completed_retention: Duration::from_secs(86400), // 24h
            failed_retention: Duration::from_secs(259200),   // 72h
            cleanup_batch_size: 1000,
            queue_retention_overrides: HashMap::new(),
            queue_stats_interval: Duration::from_secs(30),
            dlq_retention: Duration::from_secs(60 * 60 * 24 * 30),
            dlq_cleanup_batch_size: 1000,
            dlq_policy: DlqPolicy::default(),
            dirty_key_recompute_interval: Duration::from_secs(2),
            metadata_reconciliation_interval: Duration::from_secs(60),
            priority_aging_interval: Duration::from_secs(60),
            batch_operations_interval: Duration::from_secs(1),
            terminal_count_rollup_interval: Duration::from_secs(30),
            descriptor_retention: Duration::from_secs(30 * 86400), // 30d
            ring_authority_auto_flip: true,
            ring_authority_auto_flip_stable_period: Duration::from_secs(600), // 10min
            ring_authority_flip_aware_since: std::sync::Mutex::new(None),
        }
    }

    /// #456: register the configured queue-storage runtime of a
    /// canonical-resolved worker so a pre-flip leader can promote the
    /// queue-storage deferred backlog after the routing flip.
    pub(crate) fn standby_queue_storage(mut self, runtime: Option<QueueStorageRuntime>) -> Self {
        self.standby_queue_storage = runtime;
        self
    }

    /// #371 staged upgrade: enable/disable the maintenance auto-flip of the
    /// ring-cursor authority `columns -> ledger` (default: enabled). Disable
    /// if you want to drive the flip manually with
    /// `awa storage flip-ring-authority`.
    pub fn ring_authority_auto_flip(mut self, enabled: bool) -> Self {
        self.ring_authority_auto_flip = enabled;
        self
    }

    /// #371 staged upgrade: how long the fleet must be continuously
    /// flip-aware (every fresh runtime reporting a 0.7+ `binary_version`)
    /// before the maintenance leader auto-flips the ring-cursor authority
    /// (default: 10min). A conservative window absorbs a slow rolling deploy
    /// where an old binary briefly disappears and reappears.
    pub fn ring_authority_auto_flip_stable_period(mut self, period: Duration) -> Self {
        self.ring_authority_auto_flip_stable_period = period;
        self
    }

    /// Set the priority aging interval (default: 60s).
    ///
    /// Jobs waiting longer than this per priority level are promoted:
    /// a priority-4 job waiting 180s is treated as priority-1.
    pub fn priority_aging_interval(mut self, interval: Duration) -> Self {
        self.priority_aging_interval = interval;
        self
    }

    /// Set how often the maintenance leader processes batch-operation chunks.
    pub fn batch_operations_interval(mut self, interval: Duration) -> Self {
        self.batch_operations_interval = interval;
        self
    }

    /// Set how often terminal-count delta rows are folded into the live
    /// counter table for sealed queue segments (default: 30s).
    pub fn terminal_count_rollup_interval(mut self, interval: Duration) -> Self {
        self.terminal_count_rollup_interval = interval;
        self
    }

    /// How long a descriptor catalog row can go without being re-synced
    /// before the maintenance leader deletes it (default: 30 days). Set
    /// to `Duration::ZERO` to disable — useful if you maintain the catalog
    /// externally or want to keep historical descriptors forever.
    ///
    /// Descriptors carry no FK from jobs, so deletion is safe: a later
    /// worker restart that re-declares the same queue or kind will
    /// recreate the row from its declaration on the next snapshot tick.
    pub fn descriptor_retention(mut self, retention: Duration) -> Self {
        self.descriptor_retention = retention;
        self
    }

    /// Set the leader election retry interval (default: 10s).
    ///
    /// Controls how often a non-leader instance retries acquiring the
    /// advisory lock. Lower values speed up leader election in tests.
    pub fn leader_election_interval(mut self, interval: Duration) -> Self {
        self.leader_election_interval = interval;
        self
    }

    /// Set the leader connection health-check interval (default: 30s).
    pub fn leader_check_interval(mut self, interval: Duration) -> Self {
        self.leader_check_interval = interval;
        self
    }

    /// Set the promotion interval for scheduled/retryable jobs.
    pub fn promote_interval(mut self, interval: Duration) -> Self {
        self.promote_interval = interval;
        self
    }

    /// Set the stale-heartbeat rescue interval (default: 30s).
    pub fn heartbeat_rescue_interval(mut self, interval: Duration) -> Self {
        self.heartbeat_rescue_interval = interval;
        self
    }

    /// Set the deadline rescue interval (default: 30s).
    pub fn deadline_rescue_interval(mut self, interval: Duration) -> Self {
        self.deadline_rescue_interval = interval;
        self
    }

    /// Set the callback-timeout rescue interval (default: 30s).
    pub fn callback_rescue_interval(mut self, interval: Duration) -> Self {
        self.callback_rescue_interval = interval;
        self
    }

    /// Set how long a heartbeat must be stale before the job is rescued (default: 90s).
    ///
    /// Should be at least 3× the heartbeat interval to avoid false rescues
    /// from transient delays. The run-lease guard prevents duplicate completions
    /// even if a false rescue occurs, but wasted work is still undesirable.
    pub fn heartbeat_staleness(mut self, staleness: Duration) -> Self {
        self.heartbeat_staleness = staleness;
        self
    }

    /// Set the cleanup interval (default: 60s).
    pub fn cleanup_interval(mut self, interval: Duration) -> Self {
        self.cleanup_interval = interval;
        self
    }

    /// Set retention for completed jobs (default: 24h).
    pub fn completed_retention(mut self, retention: Duration) -> Self {
        self.completed_retention = retention;
        self
    }

    /// Set retention for failed/cancelled jobs (default: 72h).
    pub fn failed_retention(mut self, retention: Duration) -> Self {
        self.failed_retention = retention;
        self
    }

    /// Set the maximum number of jobs to delete per cleanup pass (default: 1000).
    pub fn cleanup_batch_size(mut self, batch_size: i64) -> Self {
        self.cleanup_batch_size = batch_size;
        self
    }

    /// Set the interval for publishing queue depth/lag metrics (default: 30s).
    pub fn queue_stats_interval(mut self, interval: Duration) -> Self {
        self.queue_stats_interval = interval;
        self
    }

    /// Set retention for DLQ rows (default: 30 days).
    pub fn dlq_retention(mut self, retention: Duration) -> Self {
        self.dlq_retention = retention;
        self
    }

    /// Set the maximum number of DLQ rows deleted per cleanup pass (default: 1000).
    pub fn dlq_cleanup_batch_size(mut self, batch_size: i64) -> Self {
        self.dlq_cleanup_batch_size = batch_size;
        self
    }

    /// Set the per-queue DLQ policy.
    pub(crate) fn dlq_policy(mut self, policy: DlqPolicy) -> Self {
        self.dlq_policy = policy;
        self
    }

    /// Set per-queue retention overrides.
    pub fn queue_retention_overrides(
        mut self,
        overrides: HashMap<String, RetentionPolicy>,
    ) -> Self {
        self.queue_retention_overrides = overrides;
        self
    }

    /// Run the maintenance loop. Attempts leader election first.
    pub async fn run(&self) {
        info!("Maintenance service starting");
        self.alive.store(true, Ordering::SeqCst);
        let _alive_guard = MaintenanceAliveGuard(self.alive.clone());
        self.leader.store(false, Ordering::SeqCst);

        loop {
            // Try to acquire advisory lock for leader election.
            // We get back a dedicated connection that holds the lock.
            let mut leader_conn = match self.try_become_leader().await {
                Ok(Some(conn)) => conn,
                Ok(None) => {
                    // Not leader — back off and try again
                    tokio::select! {
                        _ = self.cancel.cancelled() => {
                            debug!("Maintenance service shutting down (not leader)");
                            self.leader.store(false, Ordering::SeqCst);
                            return;
                        }
                        _ = tokio::time::sleep(self.leader_election_interval) => continue,
                    }
                }
                Err(err) => {
                    warn!(error = %err, "Failed to check leader status");
                    tokio::select! {
                        _ = self.cancel.cancelled() => {
                            debug!("Maintenance service shutting down (leader check failed)");
                            self.leader.store(false, Ordering::SeqCst);
                            return;
                        }
                        _ = tokio::time::sleep(self.leader_election_interval) => continue,
                    }
                }
            };

            debug!("Elected as maintenance leader");
            self.leader.store(true, Ordering::SeqCst);
            if let Some(runtime) = self.storage.queue_storage() {
                runtime.store.reset_rotation_observations();
            }

            // Run maintenance tasks as leader
            let mut heartbeat_rescue_timer = leader_loop_interval(self.heartbeat_rescue_interval);
            let mut deadline_rescue_timer = leader_loop_interval(self.deadline_rescue_interval);
            let mut callback_rescue_timer = leader_loop_interval(self.callback_rescue_interval);
            let mut promote_timer = leader_loop_interval(self.promote_interval);
            let mut cleanup_timer = leader_loop_interval(self.cleanup_interval);
            let mut cron_sync_timer = leader_loop_interval(self.cron_sync_interval);
            let mut leader_check_timer = leader_loop_interval(self.leader_check_interval);
            let mut queue_stats_timer = leader_loop_interval(self.queue_stats_interval);
            let mut dirty_key_timer = leader_loop_interval(self.dirty_key_recompute_interval);
            let mut metadata_reconciliation_timer =
                leader_loop_interval(self.metadata_reconciliation_interval);
            let mut priority_aging_timer = leader_loop_interval(self.priority_aging_interval);
            let mut batch_operations_timer = leader_loop_interval(self.batch_operations_interval);
            let mut terminal_count_rollup_timer =
                leader_loop_interval(self.terminal_count_rollup_interval);
            let mut vacuum_queue_timer = self
                .storage
                .queue_storage()
                .map(|runtime| leader_loop_interval(runtime.queue_rotate_interval));
            let mut vacuum_lease_timer = self
                .storage
                .queue_storage()
                .map(|runtime| leader_loop_interval(runtime.lease_rotate_interval));
            let mut vacuum_claim_timer = self
                .storage
                .queue_storage()
                .map(|runtime| leader_loop_interval(runtime.claim_rotate_interval));
            // Cache rotate intervals alongside the timers so the per-branch
            // observability calls can recover the configured period. Both
            // derive from the same `queue_storage()` source so they are
            // Some/None in lockstep with their corresponding timers.
            let vacuum_queue_interval = self
                .storage
                .queue_storage()
                .map(|runtime| runtime.queue_rotate_interval);
            let vacuum_lease_interval = self
                .storage
                .queue_storage()
                .map(|runtime| runtime.lease_rotate_interval);
            let vacuum_claim_interval = self
                .storage
                .queue_storage()
                .map(|runtime| runtime.claim_rotate_interval);
            // Per-branch overrun tracker. Issue #242 — observability only;
            // the architectural split of this `tokio::select!` is deferred
            // to v0.7 conditional on the overrun counter showing fleet hits.
            let branch_tracker = MaintenanceBranchTracker::new();
            // Per-segment prune backoff (#169). Gates the prune step of
            // each rotate branch so a pinned MVCC snapshot doesn't make
            // us repeat the ACCESS-EXCLUSIVE + count(*) attempt every
            // tick. Local to the leader loop so a re-election resets
            // backoff to zero.
            let prune_tracker = PruneBackoffTracker::new();

            // Skip the first immediate tick
            heartbeat_rescue_timer.tick().await;
            deadline_rescue_timer.tick().await;
            callback_rescue_timer.tick().await;
            promote_timer.tick().await;
            cleanup_timer.tick().await;
            cron_sync_timer.tick().await;
            leader_check_timer.tick().await;
            queue_stats_timer.tick().await;
            dirty_key_timer.tick().await;
            metadata_reconciliation_timer.tick().await;
            priority_aging_timer.tick().await;
            batch_operations_timer.tick().await;
            terminal_count_rollup_timer.tick().await;
            if let Some(timer) = &mut vacuum_queue_timer {
                timer.tick().await;
            }
            if let Some(timer) = &mut vacuum_lease_timer {
                timer.tick().await;
            }
            if let Some(timer) = &mut vacuum_claim_timer {
                timer.tick().await;
            }

            // Do an initial sync immediately on becoming leader
            self.sync_periodic_jobs_to_db().await;
            let cron_eval_cancel = self.cancel.child_token();
            let cron_eval_task = tokio::spawn(Self::run_cron_evaluator(
                self.pool.clone(),
                cron_eval_cancel.clone(),
                self.cron_eval_interval,
            ));

            loop {
                tokio::select! {
                    _ = self.cancel.cancelled() => {
                        debug!("Maintenance service shutting down");
                        self.leader.store(false, Ordering::SeqCst);
                        Self::stop_cron_evaluator(&cron_eval_cancel, &cron_eval_task);
                        // Release leader lock on the same connection that acquired it.
                        // If this fails, dropping the connection will release the lock anyway.
                        let _ = Self::release_leader(&mut leader_conn).await;
                        return;
                    }
                    _ = heartbeat_rescue_timer.tick() => {
                        if let Some(timer) = branch_tracker.try_begin("rescue_stale_heartbeats", self.heartbeat_rescue_interval, &self.metrics) {
                            self.rescue_stale_heartbeats().await;
                            timer.finish();
                        }
                    }
                    _ = deadline_rescue_timer.tick() => {
                        if let Some(timer) = branch_tracker.try_begin("rescue_expired_deadlines", self.deadline_rescue_interval, &self.metrics) {
                            self.rescue_expired_deadlines().await;
                            timer.finish();
                        }
                    }
                    _ = callback_rescue_timer.tick() => {
                        if let Some(timer) = branch_tracker.try_begin("rescue_expired_callbacks", self.callback_rescue_interval, &self.metrics) {
                            self.rescue_expired_callbacks().await;
                            timer.finish();
                        }
                    }
                    _ = promote_timer.tick() => {
                        if let Some(timer) = branch_tracker.try_begin_without_cooldown("promote_scheduled", self.promote_interval, &self.metrics) {
                            self.promote_scheduled().await;
                            timer.finish();
                        }
                    }
                    _ = cleanup_timer.tick() => {
                        if let Some(timer) = branch_tracker.try_begin("cleanup", self.cleanup_interval, &self.metrics) {
                            self.cleanup_completed().await;
                            self.cleanup_dlq_rows().await;
                            self.cleanup_batch_operations().await;
                            self.cleanup_stale_runtime_snapshots().await;
                            self.cleanup_stale_descriptors().await;
                            timer.finish();
                        }
                    }
                    _ = cron_sync_timer.tick() => {
                        if let Some(timer) = branch_tracker.try_begin("cron_sync", self.cron_sync_interval, &self.metrics) {
                            self.sync_periodic_jobs_to_db().await;
                            timer.finish();
                        }
                    }
                    _ = queue_stats_timer.tick() => {
                        if let Some(timer) = branch_tracker.try_begin("queue_stats", self.queue_stats_interval, &self.metrics) {
                            self.publish_queue_health_metrics().await;
                            timer.finish();
                        }
                    }
                    _ = dirty_key_timer.tick() => {
                        if let Some(timer) = branch_tracker.try_begin("recompute_dirty_admin_metadata", self.dirty_key_recompute_interval, &self.metrics) {
                            self.recompute_dirty_admin_metadata().await;
                            timer.finish();
                        }
                    }
                    _ = metadata_reconciliation_timer.tick() => {
                        if let Some(timer) = branch_tracker.try_begin("refresh_admin_metadata", self.metadata_reconciliation_interval, &self.metrics) {
                            self.refresh_admin_metadata().await;
                            timer.finish();
                        }
                    }
                    _ = priority_aging_timer.tick() => {
                        if let Some(timer) = branch_tracker.try_begin("priority_aging", self.priority_aging_interval, &self.metrics) {
                            self.age_waiting_priorities().await;
                            timer.finish();
                        }
                    }
                    _ = batch_operations_timer.tick() => {
                        if let Some(timer) = branch_tracker.try_begin("batch_operations", self.batch_operations_interval, &self.metrics) {
                            self.process_batch_operation().await;
                            timer.finish();
                        }
                    }
                    _ = terminal_count_rollup_timer.tick() => {
                        if let Some(timer) = branch_tracker.try_begin_without_cooldown("terminal_count_rollup", self.terminal_count_rollup_interval, &self.metrics) {
                            self.rollup_terminal_count_deltas().await;
                            timer.finish();
                        }
                    }
                    _ = async {
                        if let Some(timer) = &mut vacuum_queue_timer {
                            timer.tick().await;
                        } else {
                            std::future::pending::<()>().await;
                        }
                    }, if vacuum_queue_timer.is_some() => {
                        let interval = vacuum_queue_interval
                            .expect("vacuum_queue_interval Some iff vacuum_queue_timer Some");
                        if let Some(timer) = branch_tracker.try_begin("rotate_queue", interval, &self.metrics) {
                            self.rotate_queue_storage_queue(&prune_tracker).await;
                            timer.finish();
                        }
                    }
                    _ = async {
                        if let Some(timer) = &mut vacuum_lease_timer {
                            timer.tick().await;
                        } else {
                            std::future::pending::<()>().await;
                        }
                    }, if vacuum_lease_timer.is_some() => {
                        let interval = vacuum_lease_interval
                            .expect("vacuum_lease_interval Some iff vacuum_lease_timer Some");
                        if let Some(timer) = branch_tracker.try_begin("rotate_lease", interval, &self.metrics) {
                            self.rotate_queue_storage_leases(&prune_tracker).await;
                            timer.finish();
                        }
                    }
                    _ = async {
                        if let Some(timer) = &mut vacuum_claim_timer {
                            timer.tick().await;
                        } else {
                            std::future::pending::<()>().await;
                        }
                    }, if vacuum_claim_timer.is_some() => {
                        let interval = vacuum_claim_interval
                            .expect("vacuum_claim_interval Some iff vacuum_claim_timer Some");
                        if let Some(timer) = branch_tracker.try_begin("rotate_claim", interval, &self.metrics) {
                            self.rotate_queue_storage_claims(&prune_tracker).await;
                            timer.finish();
                        }
                    }
                    _ = leader_check_timer.tick() => {
                        // Verify leader connection is still alive.
                        // The advisory lock is session-scoped: if the connection is alive,
                        // the lock is held. If the query fails, the connection (and lock) are gone.
                        if sqlx::query("SELECT 1").execute(&mut *leader_conn).await.is_err() {
                            warn!("Leader connection lost, re-entering election loop");
                            self.leader.store(false, Ordering::SeqCst);
                            Self::stop_cron_evaluator(&cron_eval_cancel, &cron_eval_task);
                            break;
                        }
                    }
                }
            }
        }
    }

    /// Advisory lock key for Awa maintenance leader election.
    const LOCK_KEY: i64 = 0x_4157_415f_4d41_494e; // "AWA_MAIN" in hex-ish

    /// Try to acquire the advisory lock for leader election.
    ///
    /// Returns a dedicated connection holding the lock on success, or `None` if
    /// another instance already holds the lock. The lock is session-scoped in
    /// PostgreSQL, so it stays held as long as this connection is alive.
    async fn try_become_leader(&self) -> Result<Option<PoolConnection<Postgres>>, sqlx::Error> {
        let mut conn = self.pool.acquire().await?;
        let result: (bool,) = sqlx::query_as("SELECT pg_try_advisory_lock($1)")
            .bind(Self::LOCK_KEY)
            .fetch_one(&mut *conn)
            .await?;
        if result.0 {
            Ok(Some(conn))
        } else {
            Ok(None)
        }
    }

    /// Release the advisory lock on the same connection that acquired it.
    ///
    /// Dropping the connection also releases the lock (PG session-scoped behavior),
    /// so this is a best-effort explicit release.
    async fn release_leader(conn: &mut PoolConnection<Postgres>) -> Result<(), sqlx::Error> {
        sqlx::query("SELECT pg_advisory_unlock($1)")
            .bind(Self::LOCK_KEY)
            .execute(&mut **conn)
            .await?;
        Ok(())
    }

    async fn run_cron_evaluator(pool: PgPool, cancel: CancellationToken, interval: Duration) {
        let mut timer = tokio::time::interval(interval);
        timer.tick().await;

        loop {
            tokio::select! {
                _ = cancel.cancelled() => return,
                _ = timer.tick() => {
                    Self::evaluate_cron_schedules(&pool).await;
                }
            }
        }
    }

    fn stop_cron_evaluator(cancel: &CancellationToken, task: &JoinHandle<()>) {
        cancel.cancel();
        task.abort();
    }

    /// Sync all registered periodic job schedules to `awa.cron_jobs` via UPSERT.
    ///
    /// Unowned registrations stay additive. Owned schedules reconcile only from
    /// the shared fleet evidence, independently of this leader's local set.
    #[tracing::instrument(skip(self), name = "maintenance.cron_sync")]
    async fn sync_periodic_jobs_to_db(&self) {
        match awa_model::cron_reconciliation::owners(&self.pool).await {
            Ok(owners) => {
                for owner in owners {
                    match awa_model::cron_reconciliation::reconcile(&self.pool, &owner).await {
                        Ok(plan) => {
                            let outcome = plan
                                .blockers
                                .first()
                                .map(String::as_str)
                                .unwrap_or("converged")
                                .to_owned();
                            self.metrics
                                .cron_reconciliation_decisions
                                .add(1, &[opentelemetry::KeyValue::new("outcome", outcome)]);
                            if plan.applied {
                                self.metrics
                                    .cron_retired
                                    .add(plan.retirements.len() as u64, &[]);
                            }

                            if plan.applied
                                && (!plan.retirements.is_empty()
                                    || !plan.updates.is_empty()
                                    || !plan.additions.is_empty())
                            {
                                info!(owner = %owner, retired = plan.retirements.len(), updated = plan.updates.len(), added = plan.additions.len(), "Periodic reconciliation applied changes");
                            }
                            debug!(owner = %owner, hash = ?plan.desired_hash, blockers = ?plan.blockers, blocking_instances = ?plan.blocking_instances, grace_remaining_ms = plan.grace_remaining_ms, proposed_retirements = plan.retirements.len(), applied = plan.applied, "Periodic reconciliation");
                        }
                        Err(err) => {
                            error!(owner = %owner, error = %err, "Periodic reconciliation failed")
                        }
                    }
                }
            }
            Err(err) => error!(error = %err, "Failed to load cron reconciliation owners"),
        }

        if self.periodic_jobs.is_empty() {
            return;
        }

        for job in self.periodic_jobs.iter() {
            if let Err(err) = upsert_cron_job(&self.pool, job).await {
                error!(name = %job.name, error = %err, "Failed to sync periodic job");
            }
        }

        debug!(
            count = self.periodic_jobs.len(),
            "Synced periodic jobs to database"
        );
    }

    async fn process_batch_operation(&self) {
        let runner_instance = Uuid::new_v4();
        match awa_model::batch_operations::run_one_default_chunk(&self.pool, runner_instance).await
        {
            Ok(outcome) if outcome.claimed => {
                debug!(
                    processed = outcome.processed,
                    skipped = outcome.skipped,
                    errored = outcome.errored,
                    finalized = outcome.finalized,
                    "processed batch operation chunk"
                );
            }
            Ok(_) => {}
            Err(err) => warn!(error = %err, "failed to process batch operation chunk"),
        }
    }

    async fn cleanup_batch_operations(&self) {
        match awa_model::batch_operations::cleanup_expired_batch_operations(&self.pool, 1000).await
        {
            Ok(deleted) if deleted > 0 => {
                debug!(deleted, "cleaned up expired batch operations");
            }
            Ok(_) => {}
            Err(err) => warn!(error = %err, "failed to clean up expired batch operations"),
        }
    }

    async fn rollup_terminal_count_deltas(&self) {
        let Some(runtime) = self.storage.queue_storage() else {
            return;
        };

        match runtime
            .store
            .rollup_terminal_count_deltas(&self.pool, TERMINAL_COUNT_ROLLUP_MAX_SLOTS_PER_TICK)
            .await
        {
            Ok(TerminalDeltaRollupOutcome {
                rolled_slots: 0,
                delta_rows: 0,
                grouped_keys: 0,
                skipped_active_slots: 0,
                blocked_slots: 0,
                skipped_mvcc_pinned: false,
            }) => {}
            Ok(outcome) => {
                debug!(
                    rolled_slots = outcome.rolled_slots,
                    delta_rows = outcome.delta_rows,
                    grouped_keys = outcome.grouped_keys,
                    skipped_active_slots = outcome.skipped_active_slots,
                    blocked_slots = outcome.blocked_slots,
                    skipped_mvcc_pinned = outcome.skipped_mvcc_pinned,
                    "rolled up queue-storage terminal count deltas"
                );
            }
            Err(err) => warn!(error = %err, "failed to roll up terminal count deltas"),
        }

        // #371: the append-only prune-rollup deltas and ring-rotation
        // ledgers piggyback this tick. Both folds are horizon-gated
        // inside queue-storage (they stand down while another backend
        // pins the MVCC horizon), so calling them every tick is cheap.
        match runtime.store.fold_terminal_rollup_deltas(&self.pool).await {
            Ok(outcome) if outcome.folded_delta_rows > 0 || outcome.skipped_mvcc_pinned => {
                debug!(
                    folded_delta_rows = outcome.folded_delta_rows,
                    folded_keys = outcome.folded_keys,
                    skipped_mvcc_pinned = outcome.skipped_mvcc_pinned,
                    "folded queue-storage terminal rollup deltas"
                );
            }
            Ok(_) => {}
            Err(err) => warn!(error = %err, "failed to fold terminal rollup deltas"),
        }

        match runtime.store.fold_ring_rotation_ledgers(&self.pool).await {
            Ok(outcome) if outcome.trimmed_rows > 0 || outcome.skipped_mvcc_pinned => {
                debug!(
                    trimmed_rows = outcome.trimmed_rows,
                    skipped_mvcc_pinned = outcome.skipped_mvcc_pinned,
                    "trimmed queue-storage ring rotation ledgers"
                );
            }
            Ok(_) => {}
            Err(err) => warn!(error = %err, "failed to trim ring rotation ledgers"),
        }

        // #371 staged upgrade: piggyback the leader's rollup tick to
        // consider auto-flipping the ring-cursor authority once the fleet
        // has fully rolled to 0.7.
        self.maybe_auto_flip_ring_authority(runtime.store.schema())
            .await;
    }

    /// #371 staged rolling upgrade. On the maintenance leader, flip the
    /// ring-cursor authority `columns -> ledger` once every fresh-heartbeat
    /// runtime has reported flip-aware (a non-NULL `binary_version` >= the
    /// min flip version) continuously for the configured stable period.
    ///
    /// The stable window is tracked in-process: the first tick that sees a
    /// clean fleet stamps `ring_authority_flip_aware_since`; any tick that
    /// sees a blocking (fresh, not-flip-aware) runtime resets it. This
    /// absorbs a slow rolling deploy where an old binary briefly disappears
    /// and reappears without prematurely flipping out from under it. The
    /// flip itself re-checks the gate transactionally in
    /// `awa.flip_ring_authority` (without `--force`), so a race between the
    /// window expiring here and a fresh old binary registering cannot flip
    /// unsafely.
    async fn maybe_auto_flip_ring_authority(&self, schema: &str) {
        if !self.ring_authority_auto_flip {
            return;
        }

        let status = match awa_model::storage::ring_authority_status(&self.pool, schema).await {
            Ok(status) => status,
            Err(err) => {
                debug!(error = %err, schema, "ring authority status unavailable; skipping auto-flip");
                return;
            }
        };

        if status.authority == "ledger" {
            return; // already flipped (one-way); nothing to do
        }

        // Track the continuously-clean window in-process.
        let ready_since = {
            let mut since = self
                .ring_authority_flip_aware_since
                .lock()
                .expect("ring_authority_flip_aware_since poisoned");
            if status.blocking_instances > 0 || status.live_instances == 0 {
                // A pre-flip binary is live (or the fleet is momentarily
                // empty — never flip a fleet we cannot see). Reset the
                // window.
                *since = None;
                return;
            }
            *since.get_or_insert_with(Instant::now)
        };

        if ready_since.elapsed() < self.ring_authority_auto_flip_stable_period {
            return; // fleet clean, but not yet for long enough
        }

        match awa_model::storage::flip_ring_authority(&self.pool, schema, false).await {
            Ok(result) => {
                warn!(
                    schema,
                    authority = %result,
                    flip_aware_instances = status.flip_aware_instances,
                    stable_period_secs = self.ring_authority_auto_flip_stable_period.as_secs(),
                    "#371 auto-flipped ring-cursor authority columns -> ledger: fleet fully \
                     flip-aware for the stable period; dead-tuple-free rotation now active"
                );
            }
            Err(err) => {
                // A blocking binary re-registered between the window check
                // and the transactional gate: reset and wait again.
                warn!(error = %err, schema, "auto-flip refused by the transactional gate; will retry");
                *self
                    .ring_authority_flip_aware_since
                    .lock()
                    .expect("ring_authority_flip_aware_since poisoned") = None;
            }
        }
    }

    /// Evaluate all cron schedules and enqueue any that are due.
    ///
    /// For each schedule, computes due fire times ≤ now that are after
    /// `last_enqueued_at`. If fires are due, executes the atomic CTE for each
    /// fire in order so delayed evaluation catches up instead of collapsing
    /// intermediate fires.
    #[tracing::instrument(skip(pool), name = "maintenance.cron_eval")]
    async fn evaluate_cron_schedules(pool: &PgPool) {
        let cron_rows = match list_cron_jobs(pool).await {
            Ok(rows) => rows,
            Err(err) => {
                error!(error = %err, "Failed to load cron jobs for evaluation");
                return;
            }
        };

        if cron_rows.is_empty() {
            return;
        }

        let now = Utc::now();

        for row in &cron_rows {
            if row.is_paused() || row.retired_at.is_some() {
                debug!(cron_name = %row.name, "Skipping paused cron schedule");
                continue;
            }
            let fire_times = compute_fire_times(row, now, CRON_CATCH_UP_LIMIT);
            if fire_times.is_empty() {
                continue;
            }
            if fire_times.len() == CRON_CATCH_UP_LIMIT {
                warn!(
                    cron_name = %row.name,
                    catch_up_limit = CRON_CATCH_UP_LIMIT,
                    "Cron catch-up limit reached; remaining due fires will be retried on the next evaluation"
                );
            }

            let mut previous_enqueued_at = row.last_enqueued_at;
            for fire_time in fire_times {
                match atomic_enqueue(pool, &row.name, fire_time, previous_enqueued_at).await {
                    Ok(Some(job)) => {
                        previous_enqueued_at = Some(fire_time);
                        info!(
                            cron_name = %row.name,
                            job_id = job.id,
                            fire_time = %fire_time,
                            "Enqueued periodic job"
                        );
                    }
                    Ok(None) => {
                        // Another leader already claimed this fire — not an error
                        debug!(cron_name = %row.name, "Cron fire already claimed");
                        break;
                    }
                    Err(err) => {
                        error!(
                            cron_name = %row.name,
                            error = %err,
                            "Failed to enqueue periodic job"
                        );
                        break;
                    }
                }
            }
        }
    }

    /// Rescue jobs with stale heartbeats (crash detection).
    ///
    /// #456 mirror: during a mixed transition either engine's leader must
    /// sweep both planes — stuck canonical rows would otherwise pin
    /// `canonical_live_backlog` above zero and wedge `finalize --wait`,
    /// and stuck queue-storage rows would wait out a drain-only leader.
    /// Both cross-plane sweeps are cheap no-ops once their tables are empty.
    #[tracing::instrument(skip(self), name = "maintenance.rescue_stale")]
    async fn rescue_stale_heartbeats(&self) {
        match &self.storage {
            RuntimeStorage::Canonical => {
                let outcome = self.rescue_canonical_stale_heartbeats().await;
                self.process_heartbeat_rescues(outcome, true, None).await;
                if let Some(runtime) = self.active_standby_queue_storage().await {
                    let outcome = runtime
                        .store
                        .rescue_stale_heartbeats_with_dlq(
                            &self.pool,
                            self.heartbeat_staleness,
                            &|queue| self.dlq_policy.enabled_for(queue),
                        )
                        .await;
                    self.process_heartbeat_rescues(outcome, false, Some(runtime))
                        .await;
                }
            }
            RuntimeStorage::QueueStorage(runtime) => {
                let outcome = runtime
                    .store
                    .rescue_stale_heartbeats_with_dlq(
                        &self.pool,
                        self.heartbeat_staleness,
                        &|queue| self.dlq_policy.enabled_for(queue),
                    )
                    .await;
                self.process_heartbeat_rescues(outcome, true, Some(runtime))
                    .await;
                let outcome = self.rescue_canonical_stale_heartbeats().await;
                self.process_heartbeat_rescues(outcome, false, None).await;
            }
        }
    }

    /// Batched canonical stale-heartbeat rescue with the #388 row-at-a-time
    /// fallback on unique-claim conflict. Exhausted attempts fail in place,
    /// as callback timeouts do, so a job that keeps killing or hanging its
    /// worker stops at `max_attempts`; the rest move to `scheduled_jobs` as
    /// `retryable`.
    async fn rescue_canonical_stale_heartbeats(&self) -> Result<Vec<JobRow>, awa_model::AwaError> {
        let staleness_ms = self.heartbeat_staleness.as_millis() as i64;
        let outcome = sqlx::query_as::<_, JobRow>(
            r#"
                WITH candidates AS (
                    SELECT id, attempt, max_attempts FROM awa.jobs_hot
                    WHERE state = 'running'
                      AND heartbeat_at < now() - ($1 * interval '1 millisecond')
                    LIMIT 500
                    FOR UPDATE SKIP LOCKED
                ),
                failed AS (
                    UPDATE awa.jobs_hot
                    SET state = 'failed',
                        finalized_at = now(),
                        heartbeat_at = NULL,
                        deadline_at = NULL,
                        errors = errors || jsonb_build_object(
                            'error', 'heartbeat stale: worker presumed dead',
                            'attempt', attempt,
                            'at', now()
                        )::jsonb
                    WHERE id IN (SELECT id FROM candidates WHERE attempt >= max_attempts)
                    RETURNING *
                ),
                deleted AS (
                    DELETE FROM awa.jobs_hot
                    WHERE id IN (SELECT id FROM candidates WHERE attempt < max_attempts)
                    RETURNING *
                ),
                moved AS (
                    INSERT INTO awa.scheduled_jobs (
                        id, kind, queue, args, state, priority, attempt, max_attempts,
                        run_at, heartbeat_at, deadline_at, attempted_at, finalized_at,
                        created_at, errors, metadata, tags, unique_key, unique_states,
                        callback_id, callback_timeout_at, callback_filter,
                        callback_on_complete, callback_on_fail, callback_transform,
                        run_lease, progress
                    )
                    SELECT
                        id, kind, queue, args, 'retryable', priority, attempt, max_attempts,
                        run_at, NULL, NULL, attempted_at, now(),
                        created_at,
                        errors || jsonb_build_object(
                            'error', 'heartbeat stale: worker presumed dead',
                            'attempt', attempt,
                            'at', now()
                        )::jsonb,
                        metadata, tags, unique_key, unique_states,
                        NULL, NULL, NULL, NULL, NULL, NULL,
                        run_lease, progress
                    FROM deleted
                    RETURNING *
                )
                SELECT * FROM failed
                UNION ALL
                SELECT * FROM moved
                "#,
        )
        .bind(staleness_ms)
        .fetch_all(&self.pool)
        .await
        .map_err(awa_model::AwaError::Database);
        match outcome {
            Err(err) if is_unique_claim_conflict(&err) => {
                warn!(
                    error = %err,
                    "Batched heartbeat rescue hit a unique-claim conflict; retrying row-at-a-time (#388)"
                );
                self.rescue_canonical_per_row(
                    "SELECT id FROM awa.jobs_hot \
                     WHERE state = 'running' \
                       AND heartbeat_at < now() - ($1 * interval '1 millisecond') \
                     LIMIT 500",
                    HEARTBEAT_RESCUE_PER_ROW_SQL,
                    Some(staleness_ms),
                    "running",
                    "rescued as duplicate: heartbeat stale and unique claim held by a newer job",
                    "heartbeat",
                )
                .await
            }
            other => other,
        }
    }

    /// Metrics, cancellation signalling, and lifecycle events for one
    /// plane's stale-heartbeat rescue outcome. `signal_local` is false for
    /// the cross-plane sweep: those jobs never execute on this worker, and
    /// the planes' id sequences are independent, so an `(id, run_lease)`
    /// match in the in-flight map would flag an unrelated local job.
    async fn process_heartbeat_rescues(
        &self,
        outcome: Result<Vec<JobRow>, awa_model::AwaError>,
        signal_local: bool,
        dlq_runtime: Option<&QueueStorageRuntime>,
    ) {
        match outcome {
            Ok(rescued) if !rescued.is_empty() => {
                let (cancelled_duplicates, rescued): (Vec<_>, Vec<_>) = rescued
                    .into_iter()
                    .partition(|job| job.state == JobState::Cancelled);
                self.handle_duplicate_cancellations(
                    "heartbeat",
                    &cancelled_duplicates,
                    signal_local,
                )
                .await;
                if rescued.is_empty() {
                    return;
                }
                self.metrics.maintenance_rescues.add(
                    rescued.len() as u64,
                    &[opentelemetry::KeyValue::new("awa.rescue.kind", "heartbeat")],
                );
                warn!(count = rescued.len(), "Rescued stale heartbeat jobs");
                // Signal cancellation to any rescued jobs still running on this instance
                if signal_local {
                    self.signal_cancellation(&rescued).await;
                }
                for job in &rescued {
                    self.emit_rescued(job, crate::events::RescueReason::StaleHeartbeat)
                        .await;
                }
                self.record_rescued_dlq_moves(&rescued, dlq_runtime, "heartbeat_stale");
            }
            Err(err) => {
                error!(error = %err, "Failed to rescue stale heartbeat jobs");
            }
            _ => {}
        }
    }

    /// The standby queue-storage runtime, but only once routing has flipped
    /// to its schema — the same guard as the promote mirror in
    /// [`Self::promote_due_state`].
    async fn active_standby_queue_storage(&self) -> Option<&QueueStorageRuntime> {
        let runtime = self.standby_queue_storage.as_ref()?;
        match sqlx::query_scalar::<_, Option<String>>("SELECT awa.active_queue_storage_schema()")
            .fetch_one(&self.pool)
            .await
        {
            Ok(active_schema) if active_schema.as_deref() == Some(runtime.store.schema()) => {
                Some(runtime)
            }
            Ok(_) => None,
            Err(err) => {
                error!(error = %err, "Failed to resolve the active queue-storage schema");
                None
            }
        }
    }

    /// Rescue jobs that exceeded their hard deadline.
    ///
    /// #456 mirror: sweeps both storage planes; see
    /// [`Self::rescue_stale_heartbeats`].
    #[tracing::instrument(skip(self), name = "maintenance.rescue_deadline")]
    async fn rescue_expired_deadlines(&self) {
        match &self.storage {
            RuntimeStorage::Canonical => {
                let outcome = self.rescue_canonical_expired_deadlines().await;
                self.process_deadline_rescues(outcome, true, None).await;
                if let Some(runtime) = self.active_standby_queue_storage().await {
                    let outcome = runtime
                        .store
                        .rescue_expired_deadlines_with_dlq(&self.pool, &|queue| {
                            self.dlq_policy.enabled_for(queue)
                        })
                        .await;
                    self.process_deadline_rescues(outcome, false, Some(runtime))
                        .await;
                }
            }
            RuntimeStorage::QueueStorage(runtime) => {
                let outcome = runtime
                    .store
                    .rescue_expired_deadlines_with_dlq(&self.pool, &|queue| {
                        self.dlq_policy.enabled_for(queue)
                    })
                    .await;
                self.process_deadline_rescues(outcome, true, Some(runtime))
                    .await;
                let outcome = self.rescue_canonical_expired_deadlines().await;
                self.process_deadline_rescues(outcome, false, None).await;
            }
        }
    }

    /// Batched canonical deadline rescue with the #388 row-at-a-time
    /// fallback on unique-claim conflict. Exhausted attempts fail in place,
    /// as callback timeouts do, so a job that keeps killing or hanging its
    /// worker stops at `max_attempts`; the rest move to `scheduled_jobs` as
    /// `retryable`.
    async fn rescue_canonical_expired_deadlines(&self) -> Result<Vec<JobRow>, awa_model::AwaError> {
        let outcome = sqlx::query_as::<_, JobRow>(
            r#"
                WITH candidates AS (
                    SELECT id, attempt, max_attempts FROM awa.jobs_hot
                    WHERE state = 'running'
                      AND deadline_at IS NOT NULL
                      AND deadline_at < now()
                    LIMIT 500
                    FOR UPDATE SKIP LOCKED
                ),
                failed AS (
                    UPDATE awa.jobs_hot
                    SET state = 'failed',
                        finalized_at = now(),
                        heartbeat_at = NULL,
                        deadline_at = NULL,
                        errors = errors || jsonb_build_object(
                            'error', 'hard deadline exceeded',
                            'attempt', attempt,
                            'at', now()
                        )::jsonb
                    WHERE id IN (SELECT id FROM candidates WHERE attempt >= max_attempts)
                    RETURNING *
                ),
                deleted AS (
                    DELETE FROM awa.jobs_hot
                    WHERE id IN (SELECT id FROM candidates WHERE attempt < max_attempts)
                    RETURNING *
                ),
                moved AS (
                    INSERT INTO awa.scheduled_jobs (
                        id, kind, queue, args, state, priority, attempt, max_attempts,
                        run_at, heartbeat_at, deadline_at, attempted_at, finalized_at,
                        created_at, errors, metadata, tags, unique_key, unique_states,
                        callback_id, callback_timeout_at, callback_filter,
                        callback_on_complete, callback_on_fail, callback_transform,
                        run_lease, progress
                    )
                    SELECT
                        id, kind, queue, args, 'retryable', priority, attempt, max_attempts,
                        run_at, NULL, NULL, attempted_at, now(),
                        created_at,
                        errors || jsonb_build_object(
                            'error', 'hard deadline exceeded',
                            'attempt', attempt,
                            'at', now()
                        )::jsonb,
                        metadata, tags, unique_key, unique_states,
                        NULL, NULL, NULL, NULL, NULL, NULL,
                        run_lease, progress
                    FROM deleted
                    RETURNING *
                )
                SELECT * FROM failed
                UNION ALL
                SELECT * FROM moved
                "#,
        )
        .fetch_all(&self.pool)
        .await
        .map_err(awa_model::AwaError::Database);
        match outcome {
            Err(err) if is_unique_claim_conflict(&err) => {
                warn!(
                    error = %err,
                    "Batched deadline rescue hit a unique-claim conflict; retrying row-at-a-time (#388)"
                );
                self.rescue_canonical_per_row(
                    "SELECT id FROM awa.jobs_hot \
                     WHERE state = 'running' \
                       AND deadline_at IS NOT NULL \
                       AND deadline_at < now() \
                     LIMIT 500",
                    DEADLINE_RESCUE_PER_ROW_SQL,
                    None,
                    "running",
                    "rescued as duplicate: deadline expired and unique claim held by a newer job",
                    "deadline",
                )
                .await
            }
            other => other,
        }
    }

    /// One plane's deadline rescue outcome; `signal_local` as in
    /// [`Self::process_heartbeat_rescues`].
    async fn process_deadline_rescues(
        &self,
        outcome: Result<Vec<JobRow>, awa_model::AwaError>,
        signal_local: bool,
        dlq_runtime: Option<&QueueStorageRuntime>,
    ) {
        match outcome {
            Ok(rescued) if !rescued.is_empty() => {
                let (cancelled_duplicates, rescued): (Vec<_>, Vec<_>) = rescued
                    .into_iter()
                    .partition(|job| job.state == JobState::Cancelled);
                self.handle_duplicate_cancellations(
                    "deadline",
                    &cancelled_duplicates,
                    signal_local,
                )
                .await;
                if rescued.is_empty() {
                    return;
                }
                self.metrics.maintenance_rescues.add(
                    rescued.len() as u64,
                    &[opentelemetry::KeyValue::new("awa.rescue.kind", "deadline")],
                );
                warn!(count = rescued.len(), "Rescued deadline-expired jobs");
                // Signal cancellation so handlers see ctx.is_cancelled() == true
                if signal_local {
                    self.signal_cancellation(&rescued).await;
                }
                for job in &rescued {
                    self.emit_rescued(job, crate::events::RescueReason::DeadlineExceeded)
                        .await;
                }
                self.record_rescued_dlq_moves(&rescued, dlq_runtime, "deadline_exceeded");
            }
            Err(err) => {
                error!(error = %err, "Failed to rescue deadline-expired jobs");
            }
            _ => {}
        }
    }

    /// Rescue jobs whose callback timeout has expired.
    ///
    /// #456 mirror: sweeps both storage planes; see
    /// [`Self::rescue_stale_heartbeats`]. The canonical sweep matters
    /// doubly here because callback-carrying jobs never migrate off the
    /// canonical plane during a transition.
    #[tracing::instrument(skip(self), name = "maintenance.rescue_callback_timeout")]
    async fn rescue_expired_callbacks(&self) {
        match &self.storage {
            RuntimeStorage::Canonical => {
                let outcome = self.rescue_canonical_expired_callbacks().await;
                self.process_callback_rescues(outcome, true, None).await;
                if let Some(runtime) = self.active_standby_queue_storage().await {
                    let outcome = runtime
                        .store
                        .rescue_expired_callbacks_with_dlq(&self.pool, &|queue| {
                            self.dlq_policy.enabled_for(queue)
                        })
                        .await;
                    self.process_callback_rescues(outcome, false, Some(runtime))
                        .await;
                }
            }
            RuntimeStorage::QueueStorage(runtime) => {
                let outcome = runtime
                    .store
                    .rescue_expired_callbacks_with_dlq(&self.pool, &|queue| {
                        self.dlq_policy.enabled_for(queue)
                    })
                    .await;
                self.process_callback_rescues(outcome, true, Some(runtime))
                    .await;
                let outcome = self.rescue_canonical_expired_callbacks().await;
                self.process_callback_rescues(outcome, false, None).await;
            }
        }
    }

    /// Batched canonical callback-timeout rescue with the #388
    /// row-at-a-time fallback on unique-claim conflict. Exhausted attempts
    /// fail in place (`failed` rows live in `jobs_hot`); the rest move to
    /// `scheduled_jobs` as `retryable` with backoff, mirroring the view's
    /// state-routed UPDATE.
    async fn rescue_canonical_expired_callbacks(&self) -> Result<Vec<JobRow>, awa_model::AwaError> {
        let outcome = sqlx::query_as::<_, JobRow>(
            r#"
            WITH candidates AS (
                SELECT id, attempt, max_attempts FROM awa.jobs_hot
                WHERE state = 'waiting_external'
                  AND callback_timeout_at IS NOT NULL
                  AND callback_timeout_at < now()
                LIMIT 500
                FOR UPDATE SKIP LOCKED
            ),
            failed AS (
                UPDATE awa.jobs_hot
                SET state = 'failed',
                    finalized_at = now(),
                    callback_id = NULL,
                    callback_timeout_at = NULL,
                    callback_filter = NULL,
                    callback_on_complete = NULL,
                    callback_on_fail = NULL,
                    callback_transform = NULL,
                    errors = errors || jsonb_build_object(
                        'error', 'callback timed out',
                        'attempt', attempt,
                        'at', now()
                    )::jsonb
                WHERE id IN (SELECT id FROM candidates WHERE attempt >= max_attempts)
                RETURNING *
            ),
            deleted AS (
                DELETE FROM awa.jobs_hot
                WHERE id IN (SELECT id FROM candidates WHERE attempt < max_attempts)
                RETURNING *
            ),
            moved AS (
                INSERT INTO awa.scheduled_jobs (
                    id, kind, queue, args, state, priority, attempt, max_attempts,
                    run_at, heartbeat_at, deadline_at, attempted_at, finalized_at,
                    created_at, errors, metadata, tags, unique_key, unique_states,
                    callback_id, callback_timeout_at, callback_filter,
                    callback_on_complete, callback_on_fail, callback_transform,
                    run_lease, progress
                )
                SELECT
                    id, kind, queue, args, 'retryable', priority, attempt, max_attempts,
                    now() + awa.backoff_duration(attempt, max_attempts),
                    heartbeat_at, deadline_at, attempted_at, now(),
                    created_at,
                    errors || jsonb_build_object(
                        'error', 'callback timed out',
                        'attempt', attempt,
                        'at', now()
                    )::jsonb,
                    metadata, tags, unique_key, unique_states,
                    NULL, NULL, NULL, NULL, NULL, NULL,
                    run_lease, progress
                FROM deleted
                RETURNING *
            )
            SELECT * FROM failed
            UNION ALL
            SELECT * FROM moved
            "#,
        )
        .fetch_all(&self.pool)
        .await
        .map_err(awa_model::AwaError::Database);
        match outcome {
            Err(err) if is_unique_claim_conflict(&err) => {
                warn!(
                    error = %err,
                    "Batched callback-timeout rescue hit a unique-claim conflict; retrying row-at-a-time (#388)"
                );
                self.rescue_canonical_per_row(
                    "SELECT id FROM awa.jobs_hot \
                     WHERE state = 'waiting_external' \
                       AND callback_timeout_at IS NOT NULL \
                       AND callback_timeout_at < now() \
                     LIMIT 500",
                    CALLBACK_RESCUE_PER_ROW_SQL,
                    None,
                    "waiting_external",
                    "rescued as duplicate: callback timed out and unique claim held by a newer job",
                    "callback_timeout",
                )
                .await
            }
            other => other,
        }
    }

    /// One plane's callback-timeout rescue outcome; `signal_local` as in
    /// [`Self::process_heartbeat_rescues`]. `dlq_runtime` is the runtime
    /// owning the rescued rows' plane, when that plane is queue storage —
    /// canonical rows have no DLQ, and the independent id sequences mean
    /// routing them through a queue-storage DLQ move could hit an
    /// unrelated job with the same id.
    async fn process_callback_rescues(
        &self,
        outcome: Result<Vec<JobRow>, awa_model::AwaError>,
        signal_local: bool,
        dlq_runtime: Option<&QueueStorageRuntime>,
    ) {
        match outcome {
            Ok(rescued) if !rescued.is_empty() => {
                let (cancelled_duplicates, rescued): (Vec<_>, Vec<_>) = rescued
                    .into_iter()
                    .partition(|job| job.state == JobState::Cancelled);
                self.handle_duplicate_cancellations(
                    "callback_timeout",
                    &cancelled_duplicates,
                    signal_local,
                )
                .await;
                if rescued.is_empty() {
                    return;
                }
                self.metrics.maintenance_rescues.add(
                    rescued.len() as u64,
                    &[opentelemetry::KeyValue::new(
                        "awa.rescue.kind",
                        "callback_timeout",
                    )],
                );
                warn!(count = rescued.len(), "Rescued callback-timed-out jobs");
                for job in &rescued {
                    self.emit_rescued(job, crate::events::RescueReason::ExpiredCallback)
                        .await;
                }
                self.record_rescued_dlq_moves(&rescued, dlq_runtime, "callback_timeout");
            }
            Err(err) => {
                error!(error = %err, "Failed to rescue callback-timed-out jobs");
            }
            _ => {}
        }
    }

    /// Count rescued attempts that the queue-storage rescue dead-lettered.
    /// The store moves an exhausted attempt into the DLQ in the rescue
    /// transaction whenever the queue's policy enables it, so every `failed`
    /// row on a DLQ-enabled queue here is already in `dlq_entries`.
    /// `runtime` is the queue-storage runtime that owns the rows' plane;
    /// canonical rows have no DLQ (see [`Self::process_callback_rescues`]).
    fn record_rescued_dlq_moves(
        &self,
        rescued: &[JobRow],
        runtime: Option<&QueueStorageRuntime>,
        reason: &str,
    ) {
        if runtime.is_none() {
            return;
        }
        for job in rescued {
            if job.state == JobState::Failed && self.dlq_policy.enabled_for(&job.queue) {
                self.metrics.record_dlq_moved(&job.kind, &job.queue, reason);
            }
        }
    }

    /// Age priorities for jobs that have been waiting longer than `priority_aging_interval`.
    ///
    /// Decrements `priority` by 1 per pass for available jobs waiting longer than
    /// the aging interval (minimum priority 1). On the first age, stores the
    /// original priority in `metadata._awa_original_priority` so the API can
    /// report it accurately.
    #[tracing::instrument(skip(self), name = "maintenance.priority_aging")]
    async fn age_waiting_priorities(&self) {
        let aging_secs = self.priority_aging_interval.as_secs_f64();
        if aging_secs <= 0.0 {
            return;
        }
        if let Some(runtime) = self.storage.queue_storage() {
            debug!(
                schema = %runtime.store.schema(),
                "Queue storage uses claim-time priority aging; skipping physical reprioritization pass"
            );
            return;
        }

        match sqlx::query_scalar::<_, i64>(
            r#"
            WITH eligible AS (
                SELECT id FROM awa.jobs_hot
                WHERE state = 'available'
                  AND priority > 1
                  AND run_at <= now() - make_interval(secs => $1)
                LIMIT 1000
                FOR UPDATE SKIP LOCKED
            )
            UPDATE awa.jobs_hot
            SET priority = priority - 1,
                metadata = CASE
                    WHEN NOT (metadata ? '_awa_original_priority')
                    THEN metadata || jsonb_build_object('_awa_original_priority', priority)
                    ELSE metadata
                END
            FROM eligible
            WHERE awa.jobs_hot.id = eligible.id
            RETURNING awa.jobs_hot.id
            "#,
        )
        .bind(aging_secs)
        .fetch_all(&self.pool)
        .await
        {
            Ok(ids) if !ids.is_empty() => {
                debug!(count = ids.len(), "Aged job priorities");
            }
            Err(err) => {
                error!(error = %err, "Failed to age job priorities");
            }
            _ => {}
        }
    }

    /// Row-at-a-time fallback for the canonical rescue sweeps (#388).
    ///
    /// The batched rescue UPDATEs abort wholesale when any single row's
    /// `running -> retryable` (or `waiting_external -> retryable`) transition
    /// re-enters its unique-claim state set while another job holds the
    /// claim — `awa.sync_job_unique_claims` raises `idx_awa_jobs_unique`.
    /// One poisoned row must not starve rescue for the whole cluster, so on
    /// that specific failure the sweep retries per row: rows that rescue
    /// cleanly are returned as usual; a row whose rescue conflicts is
    /// cancelled instead (the claim holder — the job that superseded it —
    /// wins), and a row that can't even be cancelled (its mask claims
    /// `cancelled` too) is skipped with a loud log so it never blocks the
    /// rest of the batch.
    async fn rescue_canonical_per_row(
        &self,
        candidates_sql: &str,
        per_row_sql: &str,
        staleness_ms: Option<i64>,
        from_state: &str,
        duplicate_error: &str,
        rescue_kind: &'static str,
    ) -> Result<Vec<JobRow>, awa_model::AwaError> {
        let ids: Vec<i64> = {
            let query = sqlx::query_scalar(audited_sql(candidates_sql));
            let query = match staleness_ms {
                Some(ms) => query.bind(ms),
                None => query,
            };
            query
                .fetch_all(&self.pool)
                .await
                .map_err(awa_model::AwaError::Database)?
        };

        let mut rescued = Vec::new();
        for id in ids {
            let attempt = {
                let query = sqlx::query_as::<_, JobRow>(audited_sql(per_row_sql)).bind(id);
                let query = match staleness_ms {
                    Some(ms) => query.bind(ms),
                    None => query,
                };
                query
                    .fetch_optional(&self.pool)
                    .await
                    .map_err(awa_model::AwaError::Database)
            };
            match attempt {
                Ok(Some(row)) => rescued.push(row),
                // The row was completed/claimed/rescued by someone else
                // between candidate selection and the update.
                Ok(None) => {}
                Err(err) if is_unique_claim_conflict(&err) => {
                    let holder = self.unique_claim_holder(id).await;
                    warn!(
                        job_id = id,
                        claim_holder = ?holder,
                        rescue_kind,
                        "Rescue conflicts with a unique claim held by another job; \
                         cancelling the superseded job (#388)"
                    );
                    match self
                        .cancel_unique_conflicted_job(id, from_state, duplicate_error)
                        .await
                    {
                        Ok(Some(row)) => rescued.push(row),
                        Ok(None) => {}
                        Err(err) if is_unique_claim_conflict(&err) => {
                            error!(
                                job_id = id,
                                claim_holder = ?holder,
                                rescue_kind,
                                "Cannot rescue or cancel unique-conflicted job: its \
                                 unique_states mask claims 'cancelled' as well; skipping \
                                 so the sweep can proceed — resolve manually (see \
                                 docs/troubleshooting.md, #388)"
                            );
                        }
                        Err(err) => {
                            error!(job_id = id, error = %err, rescue_kind, "Failed to cancel unique-conflicted job");
                        }
                    }
                }
                Err(err) => {
                    error!(job_id = id, error = %err, rescue_kind, "Per-row rescue failed");
                }
            }
        }

        Ok(rescued)
    }

    /// Shared handling for rescue rows that were cancelled because a newer
    /// duplicate held their unique claim (either engine): metrics, local
    /// cancellation signal, and the same terminal lifecycle event a normal
    /// cancellation dispatches. `signal_local` as in
    /// [`Self::process_heartbeat_rescues`].
    async fn handle_duplicate_cancellations(
        &self,
        rescue_kind: &str,
        cancelled: &[JobRow],
        signal_local: bool,
    ) {
        if cancelled.is_empty() {
            return;
        }
        self.metrics.maintenance_rescues.add(
            cancelled.len() as u64,
            &[opentelemetry::KeyValue::new(
                "awa.rescue.kind",
                format!("{rescue_kind}_duplicate_cancelled"),
            )],
        );
        warn!(
            count = cancelled.len(),
            rescue_kind, "Cancelled unique-conflicted jobs superseded by a newer duplicate"
        );
        if signal_local {
            self.signal_cancellation(cancelled).await;
        }
        for job in cancelled {
            let handlers = self.lifecycle_handlers.clone();
            let kind = job.kind.clone();
            let reason = job
                .errors
                .as_ref()
                .and_then(|errors| errors.last())
                .and_then(|entry| entry.get("error"))
                .and_then(|value| value.as_str())
                .unwrap_or("rescued as duplicate: unique claim held by a newer job")
                .to_string();
            let event = crate::events::UntypedJobEvent::Cancelled {
                job: job.clone(),
                reason,
            };
            tokio::spawn(async move {
                crate::executor::dispatch_lifecycle_event(&handlers, &kind, event).await;
            });
        }
    }

    /// The job id currently holding the unique claim for `job_id`'s key,
    /// for operator-legible conflict logs.
    async fn unique_claim_holder(&self, job_id: i64) -> Option<i64> {
        sqlx::query_scalar(
            r#"
            SELECT c.job_id
            FROM awa.jobs_hot AS j
            JOIN awa.job_unique_claims AS c ON c.unique_key = j.unique_key
            WHERE j.id = $1 AND c.job_id <> j.id
            "#,
        )
        .bind(job_id)
        .fetch_optional(&self.pool)
        .await
        .ok()
        .flatten()
    }

    /// Terminal fallback for a job whose rescue transition conflicts with a
    /// unique claim: the newer claim holder wins, this job is cancelled with
    /// an error entry naming why. `cancelled` sits outside the default
    /// unique-states mask, so this transition normally releases nothing and
    /// claims nothing.
    async fn cancel_unique_conflicted_job(
        &self,
        job_id: i64,
        from_state: &str,
        error_message: &str,
    ) -> Result<Option<JobRow>, awa_model::AwaError> {
        let row = sqlx::query_as::<_, JobRow>(
            r#"
            UPDATE awa.jobs_hot
            SET state = 'cancelled',
                finalized_at = now(),
                heartbeat_at = NULL,
                deadline_at = NULL,
                callback_id = NULL,
                callback_timeout_at = NULL,
                callback_filter = NULL,
                callback_on_complete = NULL,
                callback_on_fail = NULL,
                callback_transform = NULL,
                errors = errors || jsonb_build_object(
                    'error', $3::text,
                    'attempt', attempt,
                    'at', now()
                )::jsonb
            WHERE id = $1 AND state = $2::awa.job_state
            RETURNING *
            "#,
        )
        .bind(job_id)
        .bind(from_state)
        .bind(error_message)
        .fetch_optional(&self.pool)
        .await?;
        Ok(row)
    }

    /// Signal cancellation to any rescued jobs that are still running on this instance.
    async fn signal_cancellation(&self, rescued_jobs: &[JobRow]) {
        for job in rescued_jobs {
            if let Some(flag) = self.in_flight.get_cancel((job.id, job.run_lease)) {
                flag.store(true, Ordering::SeqCst);
                debug!(job_id = job.id, "Signalled cancellation for rescued job");
            }
        }
    }

    /// Emit a Rescued notification: dispatch the durable follow-up specs
    /// (best-effort, separate tx — see [`Self::dispatch_rescued_followups`])
    /// then fire the in-process hook detached so a slow observer does not
    /// gate the side-effect path or any work the caller does after this
    /// (e.g. the callback-rescue DLQ move in
    /// [`Self::rescue_expired_callbacks`]).
    async fn emit_rescued(&self, job: &JobRow, reason: crate::events::RescueReason) {
        self.dispatch_rescued_followups(job, reason).await;
        let handlers = self.lifecycle_handlers.clone();
        let kind = job.kind.clone();
        let event = crate::events::UntypedJobEvent::Rescued {
            job: job.clone(),
            reason,
        };
        tokio::spawn(async move {
            crate::executor::dispatch_lifecycle_event(&handlers, &kind, event).await;
        });
    }

    /// ADR-029 best-effort dispatch of `Rescued` follow-up specs.
    ///
    /// Like callback-resolution follow-ups, this runs in a *separate*
    /// transaction from the rescue UPDATE — the rescue UPDATE has already
    /// committed by the time we're here, so we can't be atomic without
    /// teaching every rescue path to take a `&mut tx`. If a spec INSERT
    /// fails it's logged; the rescue itself remains valid.
    async fn dispatch_rescued_followups(&self, job: &JobRow, reason: crate::events::RescueReason) {
        let Some(specs) = self
            .enqueue_specs
            .get(&crate::enqueue_specs::Outcome::Rescued)
            .and_then(|by_kind| by_kind.get(&job.kind))
            .cloned()
        else {
            return;
        };
        if specs.is_empty() {
            return;
        }
        let mut tx = match self.pool.begin().await {
            Ok(tx) => tx,
            Err(err) => {
                error!(
                    job_id = job.id,
                    kind = %job.kind,
                    rescue_reason = reason.as_str(),
                    error = %err,
                    "Rescued follow-up dispatch: failed to begin transaction"
                );
                return;
            }
        };
        let outcome_ctx = crate::enqueue_specs::OutcomeContext::Rescued { reason };
        let result =
            crate::enqueue_specs::dispatch_specs_in_tx(&mut tx, job, &specs, Some(&outcome_ctx))
                .await;
        match result {
            Ok(()) => {
                if let Err(err) = tx.commit().await {
                    error!(
                        job_id = job.id,
                        kind = %job.kind,
                        rescue_reason = reason.as_str(),
                        error = %err,
                        "Rescued follow-up dispatch: commit failed"
                    );
                }
            }
            Err(err) => {
                error!(
                    job_id = job.id,
                    kind = %job.kind,
                    rescue_reason = reason.as_str(),
                    error = %err,
                    "Rescued follow-up dispatch: spec INSERT failed; rolling back"
                );
                let _ = tx.rollback().await;
            }
        }
    }

    /// Promote scheduled jobs that are now due.
    #[tracing::instrument(skip(self), name = "maintenance.promote")]
    async fn promote_scheduled(&self) {
        if let Err(err) = self.promote_due_state("scheduled", "scheduled jobs").await {
            error!(error = %err, "Failed to promote scheduled jobs");
        }
        if let Err(err) = self
            .promote_due_state("retryable", "retryable jobs (backoff elapsed)")
            .await
        {
            error!(error = %err, "Failed to promote retryable jobs");
        }
    }

    async fn promote_due_state(
        &self,
        state: &'static str,
        label: &'static str,
    ) -> Result<(), awa_model::AwaError> {
        let mut promoted_total = 0usize;
        let mut notified_queues = HashSet::new();

        for _ in 0..PROMOTE_MAX_BATCHES_PER_TICK {
            if self.cancel.is_cancelled() {
                break;
            }

            match &self.storage {
                RuntimeStorage::Canonical => {
                    let (promoted, queues) = self
                        .promote_due_batch(state)
                        .await
                        .map_err(awa_model::AwaError::Database)?;
                    if promoted == 0 {
                        break;
                    }

                    promoted_total += promoted;
                    notified_queues.extend(queues);

                    if promoted < PROMOTE_BATCH_SIZE as usize {
                        break;
                    }
                }
                RuntimeStorage::QueueStorage(runtime) => {
                    let job_state = match state {
                        "scheduled" => awa_model::JobState::Scheduled,
                        "retryable" => awa_model::JobState::Retryable,
                        other => {
                            return Err(awa_model::AwaError::Validation(format!(
                                "unsupported queue storage promote state: {other}"
                            )));
                        }
                    };
                    let promote_start = std::time::Instant::now();
                    let promoted = runtime
                        .store
                        .promote_due(&self.pool, job_state, PROMOTE_BATCH_SIZE)
                        .await?;
                    self.metrics.record_promotion_batch(
                        state,
                        promoted as u64,
                        promote_start.elapsed(),
                    );
                    if promoted == 0 {
                        break;
                    }

                    promoted_total += promoted;

                    if promoted < PROMOTE_BATCH_SIZE as usize {
                        break;
                    }
                }
            }
        }

        // #456: while a storage transition drains, due jobs can still sit in
        // the canonical deferred backlog. Drain-only runtimes execute
        // canonical work but stop leading maintenance once a post-flip
        // runtime wins the election, so a queue-storage leader must promote
        // the canonical plane too. Both queries are cheap no-ops once the
        // canonical tables are empty (every finalized or fresh install).
        if matches!(&self.storage, RuntimeStorage::QueueStorage(_)) {
            for _ in 0..PROMOTE_MAX_BATCHES_PER_TICK {
                if self.cancel.is_cancelled() {
                    break;
                }
                let (promoted, queues) = self
                    .promote_due_batch(state)
                    .await
                    .map_err(awa_model::AwaError::Database)?;
                if promoted == 0 {
                    break;
                }
                promoted_total += promoted;
                notified_queues.extend(queues);
                if promoted < PROMOTE_BATCH_SIZE as usize {
                    break;
                }
            }
        }

        // #456 mirror: a canonical-resolved (pre-flip) leader must promote
        // the queue-storage deferred backlog once routing has flipped —
        // re-scheduled canonical jobs migrate there, and post-flip runtimes
        // that would promote them may not hold leadership. The active-schema
        // guard keeps this a no-op until the flip.
        if matches!(&self.storage, RuntimeStorage::Canonical) {
            if let Some(runtime) = &self.standby_queue_storage {
                let active_schema: Option<String> =
                    sqlx::query_scalar("SELECT awa.active_queue_storage_schema()")
                        .fetch_one(&self.pool)
                        .await
                        .map_err(awa_model::AwaError::Database)?;
                if active_schema.as_deref() == Some(runtime.store.schema()) {
                    let job_state = match state {
                        "scheduled" => awa_model::JobState::Scheduled,
                        "retryable" => awa_model::JobState::Retryable,
                        other => {
                            return Err(awa_model::AwaError::Validation(format!(
                                "unsupported queue storage promote state: {other}"
                            )));
                        }
                    };
                    for _ in 0..PROMOTE_MAX_BATCHES_PER_TICK {
                        if self.cancel.is_cancelled() {
                            break;
                        }
                        let promote_start = std::time::Instant::now();
                        let promoted = runtime
                            .store
                            .promote_due(&self.pool, job_state, PROMOTE_BATCH_SIZE)
                            .await?;
                        self.metrics.record_promotion_batch(
                            state,
                            promoted as u64,
                            promote_start.elapsed(),
                        );
                        if promoted == 0 {
                            break;
                        }
                        promoted_total += promoted;
                        if promoted < PROMOTE_BATCH_SIZE as usize {
                            break;
                        }
                    }
                }
            }
        }

        if promoted_total > 0 {
            debug!(
                count = promoted_total,
                queues = notified_queues.len(),
                state,
                "Promoted {label}"
            );
        }

        Ok(())
    }

    /// SQL template for promotion. The state literal is injected directly
    /// (not as a parameter) so the planner can match the partial index on
    /// `(run_at, id) WHERE state = '<state>'`. With a parameter, the planner
    /// cannot prove the partial index applies and falls back to a full
    /// bitmap scan on multi-million-row tables.
    fn promote_sql(state: &'static str) -> String {
        format!(
            r#"
            WITH due AS (
                DELETE FROM awa.scheduled_jobs
                WHERE id IN (
                    SELECT id
                    FROM awa.scheduled_jobs
                    WHERE state = '{state}'::awa.job_state
                      AND run_at <= now()
                    ORDER BY run_at ASC, id ASC
                    LIMIT $1
                    FOR UPDATE SKIP LOCKED
                )
                RETURNING *
            ),
            promoted AS (
                INSERT INTO awa.jobs_hot (
                    id, kind, queue, args, state, priority, attempt, max_attempts,
                    run_at, heartbeat_at, deadline_at, attempted_at, finalized_at,
                    created_at, errors, metadata, tags, unique_key, unique_states,
                    callback_id, callback_timeout_at, callback_filter, callback_on_complete,
                    callback_on_fail, callback_transform, run_lease, progress
                )
                SELECT
                    id,
                    kind,
                    queue,
                    args,
                    'available'::awa.job_state,
                    priority,
                    attempt,
                    max_attempts,
                    now(),
                    NULL,
                    NULL,
                    attempted_at,
                    finalized_at,
                    created_at,
                    errors,
                    metadata,
                    tags,
                    unique_key,
                    unique_states,
                    NULL,
                    NULL,
                    NULL,
                    NULL,
                    NULL,
                    NULL,
                    run_lease,
                    progress
                FROM due
                RETURNING queue
            )
            SELECT queue FROM promoted
            "#
        )
    }

    async fn promote_due_batch(
        &self,
        state: &'static str,
    ) -> Result<(usize, HashSet<String>), sqlx::Error> {
        let mut tx = self.pool.begin().await?;
        let promote_start = std::time::Instant::now();
        let sql = Self::promote_sql(state);
        let promoted_rows: Vec<(String,)> = sqlx::query_as(audited_sql(sql))
            .bind(PROMOTE_BATCH_SIZE)
            .fetch_all(&mut *tx)
            .await?;

        let promoted = promoted_rows.len();
        self.metrics
            .record_promotion_batch(state, promoted as u64, promote_start.elapsed());
        if promoted == 0 {
            tx.commit().await?;
            return Ok((0, HashSet::new()));
        }

        let queues: HashSet<String> = promoted_rows.into_iter().map(|(queue,)| queue).collect();

        tx.commit().await?;
        Ok((promoted, queues))
    }

    async fn rotate_queue_storage_queue(&self, prune_tracker: &PruneBackoffTracker) {
        let Some(runtime) = self.storage.queue_storage() else {
            return;
        };

        let idle_cursor = match runtime
            .store
            .rotate_with_cadence(&self.pool, runtime.rotation_cadence)
            .await
        {
            Ok(outcome) => {
                self.metrics.record_rotate_outcome("queue", &outcome);
                let idle_cursor = match &outcome {
                    RotateOutcome::SkippedIdle { slot, generation } => Some((*slot, *generation)),
                    _ => None,
                };
                match outcome {
                    RotateOutcome::Rotated { slot, generation } => {
                        debug!(slot, generation, "Rotated queue storage queue segment");
                    }
                    RotateOutcome::SkippedBusy { slot, busy } => {
                        debug!(
                            slot,
                            ready_rows = busy.queue_ready,
                            claim_attempt_batches = busy.queue_claim_attempt_batches,
                            done_rows = busy.queue_done,
                            ready_segments = busy.queue_ready_segments,
                            receipt_completion_batches = busy.queue_receipt_completion_batches,
                            receipt_completion_tombstones =
                                busy.queue_receipt_completion_tombstones,
                            "Skipped busy queue storage queue segment",
                        );
                    }
                    RotateOutcome::SkippedIdle { slot, generation } => {
                        debug!(slot, generation, "Skipped idle queue storage queue segment");
                    }
                    RotateOutcome::SkippedFilling { slot, generation } => {
                        debug!(
                            slot,
                            generation, "Queue storage queue segment still filling"
                        );
                    }
                }
                idle_cursor
            }
            Err(err) => {
                error!(error = %err, "Failed to rotate queue storage queue segments");
                return;
            }
        };

        if let Some((slot, generation)) = idle_cursor.and_then(|(slot, generation)| {
            prune_tracker.already_pruned_for_idle_cursor(
                PRUNE_BRANCH_QUEUE,
                slot,
                generation,
                runtime.store.queue_slot_count(),
            )
        }) {
            self.metrics.record_prune_already_pruned("queue");
            debug!(
                slot,
                generation, "Queue storage queue segment already pruned"
            );
            return;
        }

        if prune_tracker.should_skip(PRUNE_BRANCH_QUEUE) {
            debug!(branch = PRUNE_BRANCH_QUEUE, "Prune backed off this tick");
            self.metrics.record_prune_backed_off("queue");
            return;
        }

        match runtime
            .store
            .prune_oldest(&self.pool, self.failed_retention)
            .await
        {
            Ok(outcome) => {
                self.metrics.record_prune_outcome("queue", &outcome);
                prune_tracker.record_outcome(PRUNE_BRANCH_QUEUE, &outcome);
                match outcome {
                    PruneOutcome::Noop => {}
                    PruneOutcome::Pruned {
                        slot,
                        generation,
                        carried_failed_rows,
                        durations,
                    } => {
                        debug!(
                            slot,
                            generation,
                            carried_failed_rows,
                            lock_ms = durations.lock.as_secs_f64() * 1000.0,
                            truncate_ms = durations.truncate.as_secs_f64() * 1000.0,
                            commit_ms = durations.commit.as_secs_f64() * 1000.0,
                            "Pruned queue storage queue segment"
                        );
                    }
                    PruneOutcome::Blocked { slot } => {
                        debug!(slot, "Queue storage queue segment prune blocked");
                    }
                    PruneOutcome::SkippedActive {
                        slot,
                        reason,
                        count,
                    } => {
                        debug!(
                            slot,
                            reason = reason.as_str(),
                            count,
                            "Queue storage queue segment still active",
                        );
                    }
                }
            }
            Err(err) => {
                self.metrics.record_prune_error("queue");
                error!(error = %err, "Failed to prune queue storage queue segments");
            }
        }
    }

    async fn rotate_queue_storage_leases(&self, prune_tracker: &PruneBackoffTracker) {
        let Some(runtime) = self.storage.queue_storage() else {
            return;
        };

        let idle_cursor = match runtime
            .store
            .rotate_leases_with_cadence(&self.pool, runtime.rotation_cadence)
            .await
        {
            Ok(outcome) => {
                self.metrics.record_rotate_outcome("lease", &outcome);
                let idle_cursor = match &outcome {
                    RotateOutcome::SkippedIdle { slot, generation } => Some((*slot, *generation)),
                    _ => None,
                };
                match outcome {
                    RotateOutcome::Rotated { slot, generation } => {
                        debug!(slot, generation, "Rotated queue storage lease segment");
                    }
                    RotateOutcome::SkippedBusy { slot, busy } => {
                        debug!(
                            slot,
                            lease_rows = busy.leases,
                            "Skipped busy queue storage lease segment",
                        );
                    }
                    RotateOutcome::SkippedIdle { slot, generation } => {
                        debug!(slot, generation, "Skipped idle queue storage lease segment");
                    }
                    RotateOutcome::SkippedFilling { slot, generation } => {
                        debug!(
                            slot,
                            generation, "Queue storage lease segment still filling"
                        );
                    }
                }
                idle_cursor
            }
            Err(err) => {
                error!(error = %err, "Failed to rotate queue storage lease segments");
                return;
            }
        };

        if let Some((slot, generation)) = idle_cursor.and_then(|(slot, generation)| {
            prune_tracker.already_pruned_for_idle_cursor(
                PRUNE_BRANCH_LEASE,
                slot,
                generation,
                runtime.store.lease_slot_count(),
            )
        }) {
            self.metrics.record_prune_already_pruned("lease");
            debug!(
                slot,
                generation, "Queue storage lease segment already pruned"
            );
            return;
        }

        if prune_tracker.should_skip(PRUNE_BRANCH_LEASE) {
            debug!(branch = PRUNE_BRANCH_LEASE, "Prune backed off this tick");
            self.metrics.record_prune_backed_off("lease");
            return;
        }

        match runtime.store.prune_oldest_leases(&self.pool).await {
            Ok(outcome) => {
                self.metrics.record_prune_outcome("lease", &outcome);
                prune_tracker.record_outcome(PRUNE_BRANCH_LEASE, &outcome);
                match outcome {
                    PruneOutcome::Noop => {}
                    PruneOutcome::Pruned { slot, .. } => {
                        debug!(slot, "Pruned queue storage lease segment");
                    }
                    PruneOutcome::Blocked { slot } => {
                        debug!(slot, "Queue storage lease segment prune blocked");
                    }
                    PruneOutcome::SkippedActive {
                        slot,
                        reason,
                        count,
                    } => {
                        debug!(
                            slot,
                            reason = reason.as_str(),
                            count,
                            "Queue storage lease segment still active",
                        );
                    }
                }
            }
            Err(err) => {
                self.metrics.record_prune_error("lease");
                error!(error = %err, "Failed to prune queue storage lease segments");
            }
        }
    }

    /// Claim-ring maintenance tick (see ADR-023). Rotates the claim-ring
    /// cursor and prunes the oldest fully-closed partition, mirroring the
    /// lease-ring rotate/prune pair above.
    async fn rotate_queue_storage_claims(&self, prune_tracker: &PruneBackoffTracker) {
        let Some(runtime) = self.storage.queue_storage() else {
            return;
        };

        let idle_cursor = match runtime
            .store
            .rotate_claims_with_cadence(&self.pool, runtime.rotation_cadence)
            .await
        {
            Ok(outcome) => {
                self.metrics.record_rotate_outcome("claim", &outcome);
                let idle_cursor = match &outcome {
                    RotateOutcome::SkippedIdle { slot, generation } => Some((*slot, *generation)),
                    _ => None,
                };
                match outcome {
                    RotateOutcome::Rotated { slot, generation } => {
                        debug!(slot, generation, "Rotated queue storage claim segment");
                    }
                    RotateOutcome::SkippedBusy { slot, busy } => {
                        debug!(
                            slot,
                            claim_rows = busy.claims,
                            closure_rows = busy.closures,
                            closure_batch_rows = busy.closure_batches,
                            "Skipped busy queue storage claim segment",
                        );
                    }
                    RotateOutcome::SkippedIdle { slot, generation } => {
                        debug!(slot, generation, "Skipped idle queue storage claim segment");
                    }
                    RotateOutcome::SkippedFilling { slot, generation } => {
                        debug!(
                            slot,
                            generation, "Queue storage claim segment still filling"
                        );
                    }
                }
                idle_cursor
            }
            Err(err) => {
                error!(error = %err, "Failed to rotate queue storage claim segments");
                return;
            }
        };

        if let Some((slot, generation)) = idle_cursor.and_then(|(slot, generation)| {
            prune_tracker.already_pruned_for_idle_cursor(
                PRUNE_BRANCH_CLAIM,
                slot,
                generation,
                runtime.store.claim_slot_count(),
            )
        }) {
            self.metrics.record_prune_already_pruned("claim");
            debug!(
                slot,
                generation, "Queue storage claim segment already pruned"
            );
            return;
        }

        if prune_tracker.should_skip(PRUNE_BRANCH_CLAIM) {
            debug!(branch = PRUNE_BRANCH_CLAIM, "Prune backed off this tick");
            self.metrics.record_prune_backed_off("claim");
            return;
        }

        match runtime.store.prune_oldest_claims(&self.pool).await {
            Ok(outcome) => {
                self.metrics.record_prune_outcome("claim", &outcome);
                prune_tracker.record_outcome(PRUNE_BRANCH_CLAIM, &outcome);
                match outcome {
                    PruneOutcome::Noop => {}
                    PruneOutcome::Pruned { slot, .. } => {
                        debug!(slot, "Pruned queue storage claim segment");
                    }
                    PruneOutcome::Blocked { slot } => {
                        debug!(slot, "Queue storage claim segment prune blocked");
                    }
                    PruneOutcome::SkippedActive {
                        slot,
                        reason,
                        count,
                    } => {
                        debug!(
                            slot,
                            reason = reason.as_str(),
                            count,
                            "Queue storage claim segment still active",
                        );
                    }
                }
            }
            Err(err) => {
                self.metrics.record_prune_error("claim");
                error!(error = %err, "Failed to prune queue storage claim segments");
            }
        }
    }

    /// Clean up completed/failed/cancelled jobs past retention.
    ///
    /// Targets `jobs_hot` directly (bypassing the `awa.jobs` INSTEAD OF trigger)
    /// since terminal-state jobs always reside in `jobs_hot`.
    /// Runs a global pass for queues without overrides, then per-queue passes
    /// for queues with custom retention.
    #[tracing::instrument(skip(self), name = "maintenance.cleanup")]
    async fn cleanup_completed(&self) {
        if matches!(self.storage, RuntimeStorage::QueueStorage(_)) {
            // Queue storage uses rotation/prune rather than row-by-row cleanup.
            return;
        }

        let mut total_deleted: u64 = 0;

        // Collect override queue names for the exclusion clause
        let override_queues: Vec<String> = self.queue_retention_overrides.keys().cloned().collect();

        // Global pass: delete jobs in queues that do NOT have overrides
        let completed_retention_secs =
            i64::try_from(self.completed_retention.as_secs()).unwrap_or(i64::MAX);
        let failed_retention_secs =
            i64::try_from(self.failed_retention.as_secs()).unwrap_or(i64::MAX);

        let global_result = if override_queues.is_empty() {
            sqlx::query(
                r#"
                DELETE FROM awa.jobs_hot
                WHERE id IN (
                    SELECT id FROM awa.jobs_hot
                    WHERE (state = 'completed' AND finalized_at < now() - make_interval(secs => $1::bigint))
                       OR (state IN ('failed', 'cancelled') AND finalized_at < now() - make_interval(secs => $2::bigint))
                    LIMIT $3
                )
                "#,
            )
            .bind(completed_retention_secs)
            .bind(failed_retention_secs)
            .bind(self.cleanup_batch_size)
            .execute(&self.pool)
            .await
        } else {
            sqlx::query(
                r#"
                DELETE FROM awa.jobs_hot
                WHERE id IN (
                    SELECT id FROM awa.jobs_hot
                    WHERE ((state = 'completed' AND finalized_at < now() - make_interval(secs => $1::bigint))
                       OR (state IN ('failed', 'cancelled') AND finalized_at < now() - make_interval(secs => $2::bigint)))
                      AND queue != ALL($4::text[])
                    LIMIT $3
                )
                "#,
            )
            .bind(completed_retention_secs)
            .bind(failed_retention_secs)
            .bind(self.cleanup_batch_size)
            .bind(&override_queues)
            .execute(&self.pool)
            .await
        };

        match global_result {
            Ok(result) if result.rows_affected() > 0 => {
                total_deleted += result.rows_affected();
            }
            Err(err) => {
                error!(error = %err, "Failed to clean up old jobs (global pass)");
            }
            _ => {}
        }

        // Per-queue override passes
        for (queue_name, policy) in &self.queue_retention_overrides {
            let queue_completed_secs =
                i64::try_from(policy.completed.as_secs()).unwrap_or(i64::MAX);
            let queue_failed_secs = i64::try_from(policy.failed.as_secs()).unwrap_or(i64::MAX);

            match sqlx::query(
                r#"
                DELETE FROM awa.jobs_hot
                WHERE id IN (
                    SELECT id FROM awa.jobs_hot
                    WHERE queue = $4
                      AND ((state = 'completed' AND finalized_at < now() - make_interval(secs => $1::bigint))
                        OR (state IN ('failed', 'cancelled') AND finalized_at < now() - make_interval(secs => $2::bigint)))
                    LIMIT $3
                )
                "#,
            )
            .bind(queue_completed_secs)
            .bind(queue_failed_secs)
            .bind(self.cleanup_batch_size)
            .bind(queue_name)
            .execute(&self.pool)
            .await
            {
                Ok(result) if result.rows_affected() > 0 => {
                    total_deleted += result.rows_affected();
                    debug!(
                        queue = %queue_name,
                        messaging.destination.name = %queue_name,
                        count = result.rows_affected(),
                        "Cleaned up old jobs (queue override)"
                    );
                }
                Err(err) => {
                    error!(
                        queue = %queue_name,
                        messaging.destination.name = %queue_name,
                        error = %err,
                        "Failed to clean up old jobs (queue override)"
                    );
                }
                _ => {}
            }
        }

        if total_deleted > 0 {
            info!(count = total_deleted, "Cleaned up old jobs");
        }
    }

    #[tracing::instrument(skip(self), name = "maintenance.cleanup_dlq")]
    async fn cleanup_dlq_rows(&self) {
        let RuntimeStorage::QueueStorage(runtime) = &self.storage else {
            return;
        };

        let schema = runtime.store.schema();
        let override_queues: Vec<&str> = self
            .queue_retention_overrides
            .iter()
            .filter(|(_, policy)| policy.dlq.is_some())
            .map(|(queue, _)| queue.as_str())
            .collect();
        let retention_secs = i64::try_from(self.dlq_retention.as_secs()).unwrap_or(i64::MAX);

        let global_result = if override_queues.is_empty() {
            sqlx::query(audited_sql(format!(
                r#"
                DELETE FROM {schema}.dlq_entries
                WHERE job_id IN (
                    SELECT job_id FROM {schema}.dlq_entries
                    WHERE dlq_at < now() - make_interval(secs => $1::bigint)
                    LIMIT $2
                )
                "#
            )))
            .bind(retention_secs)
            .bind(self.dlq_cleanup_batch_size)
            .execute(&self.pool)
            .await
        } else {
            sqlx::query(audited_sql(format!(
                r#"
                DELETE FROM {schema}.dlq_entries
                WHERE job_id IN (
                    SELECT job_id FROM {schema}.dlq_entries
                    WHERE dlq_at < now() - make_interval(secs => $1::bigint)
                      AND queue != ALL($3::text[])
                    LIMIT $2
                )
                "#
            )))
            .bind(retention_secs)
            .bind(self.dlq_cleanup_batch_size)
            .bind(&override_queues)
            .execute(&self.pool)
            .await
        };

        match global_result {
            Ok(result) if result.rows_affected() > 0 => {
                self.metrics.record_dlq_purged(None, result.rows_affected());
            }
            Err(err) => {
                error!(error = %err, "Failed to clean up DLQ rows (global pass)");
            }
            _ => {}
        }

        for (queue, policy) in &self.queue_retention_overrides {
            let Some(retention) = policy.dlq else {
                continue;
            };
            let retention_secs = i64::try_from(retention.as_secs()).unwrap_or(i64::MAX);
            match sqlx::query(audited_sql(format!(
                r#"
                DELETE FROM {schema}.dlq_entries
                WHERE job_id IN (
                    SELECT job_id FROM {schema}.dlq_entries
                    WHERE queue = $3
                      AND dlq_at < now() - make_interval(secs => $1::bigint)
                    LIMIT $2
                )
                "#
            )))
            .bind(retention_secs)
            .bind(self.dlq_cleanup_batch_size)
            .bind(queue)
            .execute(&self.pool)
            .await
            {
                Ok(result) if result.rows_affected() > 0 => {
                    self.metrics
                        .record_dlq_purged(Some(queue), result.rows_affected());
                }
                Err(err) => {
                    error!(queue, error = %err, "Failed to clean up DLQ rows");
                }
                _ => {}
            }
        }
    }
}

struct MaintenanceAliveGuard(Arc<AtomicBool>);

impl Drop for MaintenanceAliveGuard {
    fn drop(&mut self) {
        self.0.store(false, Ordering::SeqCst);
    }
}

/// Compute due fire times for a cron job row, using its expression and timezone.
///
/// Existing schedules can catch up missed fires up to `limit` when configured.
/// First registration always enqueues only the latest due fire to avoid
/// backfilling before the schedule was known to AWA.
fn compute_fire_times(
    row: &CronJobRow,
    now: chrono::DateTime<Utc>,
    limit: usize,
) -> Vec<chrono::DateTime<Utc>> {
    let cron = match Cron::new(&row.cron_expr).with_seconds_optional().parse() {
        Ok(c) => c,
        Err(err) => {
            error!(cron_name = %row.name, error = %err, "Invalid cron expression in database");
            return Vec::new();
        }
    };

    let tz: chrono_tz::Tz = match row.timezone.parse() {
        Ok(tz) => tz,
        Err(err) => {
            error!(cron_name = %row.name, error = %err, "Invalid timezone in database");
            return Vec::new();
        }
    };

    let search_start = match row.last_enqueued_at {
        Some(last) => last.with_timezone(&tz),
        // First registration: search from one interval before created_at
        // so that the current minute's fire is found. Without this,
        // a schedule created at HH:MM:30 won't find the HH:MM:00 fire
        // because created_at > fire_time, causing up to 60s delay.
        None => (row.created_at - chrono::Duration::minutes(1)).with_timezone(&tz),
    };

    let missed_fire_policy = match CronMissedFirePolicy::parse(&row.missed_fire_policy) {
        Ok(policy) => policy,
        Err(err) => {
            error!(cron_name = %row.name, error = %err, "Invalid cron missed-fire policy in database");
            return Vec::new();
        }
    };
    let should_catch_up =
        row.last_enqueued_at.is_some() && missed_fire_policy == CronMissedFirePolicy::CatchUp;

    if !should_catch_up {
        return latest_due_fire(&cron, tz, search_start, row.last_enqueued_at, now)
            .into_iter()
            .collect();
    }

    let mut fire_times = Vec::new();
    for fire_time in cron.iter_from(search_start) {
        let fire_utc = fire_time.with_timezone(&Utc);

        if fire_utc > now {
            break;
        }

        if let Some(last) = row.last_enqueued_at {
            if fire_utc <= last {
                continue;
            }
        }

        fire_times.push(fire_utc);
        if fire_times.len() >= limit {
            break;
        }
    }

    fire_times
}

fn latest_due_fire(
    cron: &Cron,
    tz: chrono_tz::Tz,
    search_start: chrono::DateTime<chrono_tz::Tz>,
    last_enqueued_at: Option<chrono::DateTime<Utc>>,
    now: chrono::DateTime<Utc>,
) -> Option<chrono::DateTime<Utc>> {
    let first_due = first_due_fire(cron, search_start, last_enqueued_at, now)?;
    let total_span_seconds = now.signed_duration_since(first_due).num_seconds().max(1);
    let mut lookback_seconds = 1_i64;

    loop {
        // Croner has no previous-occurrence iterator. Search backward from
        // `now` with an exponentially growing window, then scan only that
        // small window to preserve coalesced/latest-only semantics.
        let window_start_utc = (now - chrono::Duration::seconds(lookback_seconds)).max(first_due);
        let window_start = window_start_utc.with_timezone(&tz);
        let next_in_window = cron
            .iter_from(window_start)
            .next()
            .map(|fire_time| fire_time.with_timezone(&Utc));

        if next_in_window.is_some_and(|fire_utc| {
            fire_utc <= now && last_enqueued_at.is_none_or(|last| fire_utc > last)
        }) {
            return latest_due_fire_in_window(cron, window_start, last_enqueued_at, now)
                .or(Some(first_due));
        }

        if lookback_seconds >= total_span_seconds {
            return Some(first_due);
        }

        lookback_seconds = lookback_seconds.saturating_mul(2).min(total_span_seconds);
    }
}

fn first_due_fire(
    cron: &Cron,
    search_start: chrono::DateTime<chrono_tz::Tz>,
    last_enqueued_at: Option<chrono::DateTime<Utc>>,
    now: chrono::DateTime<Utc>,
) -> Option<chrono::DateTime<Utc>> {
    for fire_time in cron.iter_from(search_start) {
        let fire_utc = fire_time.with_timezone(&Utc);
        if fire_utc > now {
            return None;
        }
        if last_enqueued_at.is_none_or(|last| fire_utc > last) {
            return Some(fire_utc);
        }
    }

    None
}

fn latest_due_fire_in_window(
    cron: &Cron,
    window_start: chrono::DateTime<chrono_tz::Tz>,
    last_enqueued_at: Option<chrono::DateTime<Utc>>,
    now: chrono::DateTime<Utc>,
) -> Option<chrono::DateTime<Utc>> {
    let mut latest_fire = None;

    for fire_time in cron.iter_from(window_start) {
        let fire_utc = fire_time.with_timezone(&Utc);
        if fire_utc > now {
            break;
        }
        if last_enqueued_at.is_none_or(|last| fire_utc > last) {
            latest_fire = Some(fire_utc);
        }
    }

    latest_fire
}

impl MaintenanceService {
    /// Clean up runtime snapshots older than 24 hours.
    /// Runs as part of the leader's cleanup cycle (not on every snapshot publish).
    #[tracing::instrument(skip(self), name = "maintenance.cleanup_runtime_snapshots")]
    async fn cleanup_stale_runtime_snapshots(&self) {
        if let Err(err) = awa_model::admin::cleanup_runtime_snapshots(
            &self.pool,
            chrono::TimeDelta::try_hours(24).unwrap(),
        )
        .await
        {
            tracing::warn!(error = %err, "Failed to clean up stale runtime snapshots");
        }
    }

    /// Delete catalog rows whose last_seen_at is older than
    /// `descriptor_retention`. Runs alongside the existing cleanup cycle.
    /// When retention is zero this is a no-op, so this stays cheap for
    /// operators who don't want descriptor GC.
    #[tracing::instrument(skip(self), name = "maintenance.cleanup_stale_descriptors")]
    async fn cleanup_stale_descriptors(&self) {
        if self.descriptor_retention.is_zero() {
            return;
        }
        let max_age = chrono::TimeDelta::from_std(self.descriptor_retention)
            .unwrap_or_else(|_| chrono::TimeDelta::try_days(30).unwrap());
        for table in ["awa.queue_descriptors", "awa.job_kind_descriptors"] {
            match awa_model::admin::cleanup_stale_descriptors(&self.pool, table, max_age).await {
                Ok(deleted) if deleted > 0 => {
                    tracing::info!(table, deleted, "Cleaned up stale descriptor rows");
                }
                Ok(_) => {}
                Err(err) => {
                    tracing::warn!(table, error = %err, "Failed to clean up stale descriptors");
                }
            }
        }
    }

    /// Drain dirty keys and recompute exact cached rows for recently-touched
    /// queues and kinds. This is the primary cache update mechanism — called
    /// every ~2s to keep dashboard counters fresh.
    #[tracing::instrument(skip(self), name = "maintenance.recompute_dirty_metadata")]
    async fn recompute_dirty_admin_metadata(&self) {
        if self.storage.queue_storage().is_some() {
            return;
        }
        match awa_model::admin::recompute_dirty_admin_metadata(&self.pool).await {
            Ok(count) if count > 0 => {
                tracing::debug!(count, "Recomputed dirty admin metadata keys");
            }
            Err(err) => {
                tracing::warn!(error = %err, "Failed to recompute dirty admin metadata");
            }
            _ => {}
        }
    }

    /// Full reconciliation of admin metadata from base tables.
    /// Safety net for any drift — runs infrequently (~60s).
    #[tracing::instrument(skip(self), name = "maintenance.refresh_admin_metadata")]
    async fn refresh_admin_metadata(&self) {
        if self.storage.queue_storage().is_some() {
            return;
        }
        if let Err(err) = awa_model::admin::refresh_admin_metadata(&self.pool).await {
            tracing::warn!(error = %err, "Failed to refresh admin metadata");
        }
    }

    /// Publish queue depth and lag as OTel gauge metrics.
    #[tracing::instrument(skip(self), name = "maintenance.queue_stats")]
    async fn publish_queue_health_metrics(&self) {
        if let RuntimeStorage::QueueStorage(runtime) = &self.storage {
            self.publish_queue_storage_health_metrics(runtime).await;
            return;
        }

        let stats = match awa_model::admin::queue_overviews(&self.pool).await {
            Ok(stats) => stats,
            Err(err) => {
                tracing::warn!(error = %err, "Failed to query queue stats for metrics");
                return;
            }
        };

        for queue_stat in &stats {
            let queue = &queue_stat.queue;

            // Depth per state
            self.metrics
                .record_queue_depth(queue, "available", queue_stat.available);
            self.metrics
                .record_queue_depth(queue, "running", queue_stat.running);
            self.metrics
                .record_queue_depth(queue, "failed", queue_stat.failed);
            self.metrics
                .record_queue_depth(queue, "scheduled", queue_stat.scheduled);
            self.metrics
                .record_queue_depth(queue, "retryable", queue_stat.retryable);
            self.metrics
                .record_queue_depth(queue, "waiting_external", queue_stat.waiting_external);

            // Lag
            if let Some(lag_seconds) = queue_stat.lag_seconds {
                self.metrics.record_queue_lag(queue, lag_seconds);
            }
        }
    }

    async fn publish_queue_storage_health_metrics(
        &self,
        runtime: &crate::storage::QueueStorageRuntime,
    ) {
        let schema = runtime.store.schema();
        // This is a high-cadence metrics path, not an admin exact-count
        // endpoint, and it MUST NOT read the receipt-claim tables
        // (`lease_claims` / `lease_claim_batches` / the closure tables):
        // claim-ring rotation/prune takes them ACCESS EXCLUSIVE, and a gauge
        // that blocked on that lock would stall the whole maintenance loop.
        // The `queue_storage_metrics_query_uses_bounded_observability_path`
        // unit test enforces this by locking those tables and requiring this
        // function to still finish within 3s. So the `running` gauge below is
        // intentionally `leases`-only (materialised claims). It therefore
        // UNDER-reports queues whose in-flight work is still on compact
        // `lease_claim_batches` (#246) and has not materialised — that is an
        // accepted approximation for the cheap gauge. Surfaces that need the
        // exact running count (which DOES expand lease_claim_batches) go
        // through QueueStorage::queue_counts_exact() / admin::state_counts().
        let rows: Vec<QueueStorageMetricRow> = match sqlx::query_as(audited_sql(format!(
            "{} SELECT queue, available, running, waiting_external, scheduled, retryable, \
             failed_done, failed_dlq, lag_seconds FROM signals ORDER BY queue",
            awa_model::queue_storage::queue_signals_sql(
                schema,
                awa_model::queue_storage::QueueSignalReads::Gauge,
            )
        )))
        .fetch_all(&self.pool)
        .await
        {
            Ok(rows) => rows,
            Err(err) => {
                tracing::warn!(error = %err, "Failed to query queue storage stats for metrics");
                return;
            }
        };

        for (
            queue,
            available,
            running,
            waiting_external,
            scheduled,
            retryable,
            failed_done,
            failed_dlq,
            lag_seconds,
        ) in rows
        {
            self.metrics
                .record_queue_depth(&queue, "available", available);
            self.metrics.record_queue_depth(&queue, "running", running);
            self.metrics
                .record_queue_depth(&queue, "failed", failed_done + failed_dlq);
            self.metrics
                .record_queue_depth(&queue, "scheduled", scheduled);
            self.metrics
                .record_queue_depth(&queue, "retryable", retryable);
            self.metrics
                .record_queue_depth(&queue, "waiting_external", waiting_external);
            self.metrics.record_dlq_depth(&queue, failed_dlq);

            if let Some(lag_seconds) = lag_seconds {
                self.metrics.record_queue_lag(&queue, lag_seconds);
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use awa_model::{migrations, QueueStorage, QueueStorageConfig};
    use chrono::TimeZone;
    use sqlx::postgres::PgPoolOptions;
    use std::sync::OnceLock;

    fn cron_row(
        cron_expr: &str,
        created_at: chrono::DateTime<Utc>,
        last_enqueued_at: Option<chrono::DateTime<Utc>>,
        missed_fire_policy: CronMissedFirePolicy,
    ) -> CronJobRow {
        CronJobRow {
            owner_id: None,
            retired_at: None,
            retired_by: None,
            retired_revision: None,
            name: "test_cron".to_string(),
            cron_expr: cron_expr.to_string(),
            timezone: "UTC".to_string(),
            kind: "test_job".to_string(),
            queue: "default".to_string(),
            args: serde_json::json!({}),
            priority: 2,
            max_attempts: 25,
            tags: Vec::new(),
            metadata: serde_json::json!({}),
            missed_fire_policy: missed_fire_policy.as_str().to_string(),
            last_enqueued_at,
            created_at,
            updated_at: created_at,
            paused_at: None,
            paused_by: None,
        }
    }

    #[test]
    fn compute_fire_times_coalesces_missed_existing_fires_by_default() {
        let last = Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 0).unwrap();
        let now = Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 20).unwrap();
        let row = cron_row(
            "*/5 * * * * *",
            last,
            Some(last),
            CronMissedFirePolicy::Coalesce,
        );

        let fires = compute_fire_times(&row, now, CRON_CATCH_UP_LIMIT);

        assert_eq!(
            fires,
            vec![Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 20).unwrap()]
        );
    }

    #[test]
    fn compute_fire_times_coalesces_to_latest_fire_after_long_outage() {
        let last = Utc.with_ymd_and_hms(2026, 5, 6, 12, 0, 0).unwrap();
        let now = Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 20).unwrap();
        let row = cron_row(
            "*/1 * * * * *",
            last,
            Some(last),
            CronMissedFirePolicy::Coalesce,
        );

        let fires = compute_fire_times(&row, now, 2);

        assert_eq!(
            fires,
            vec![Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 20).unwrap()]
        );
    }

    #[test]
    fn compute_fire_times_catches_up_when_policy_requests_it() {
        let last = Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 0).unwrap();
        let now = Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 20).unwrap();
        let row = cron_row(
            "*/5 * * * * *",
            last,
            Some(last),
            CronMissedFirePolicy::CatchUp,
        );

        let fires = compute_fire_times(&row, now, CRON_CATCH_UP_LIMIT);

        assert_eq!(
            fires,
            vec![
                Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 5).unwrap(),
                Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 10).unwrap(),
                Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 15).unwrap(),
                Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 20).unwrap(),
            ]
        );
    }

    #[test]
    fn compute_fire_times_limits_catch_up_work() {
        let last = Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 0).unwrap();
        let now = Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 30).unwrap();
        let row = cron_row(
            "*/5 * * * * *",
            last,
            Some(last),
            CronMissedFirePolicy::CatchUp,
        );

        let fires = compute_fire_times(&row, now, 2);

        assert_eq!(
            fires,
            vec![
                Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 5).unwrap(),
                Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 10).unwrap(),
            ]
        );
    }

    // ── MaintenanceBranchTracker (#242) ──────────────────────────────

    fn metrics_for_test() -> crate::metrics::AwaMetrics {
        crate::metrics::AwaMetrics::from_global()
    }

    fn database_url() -> String {
        std::env::var("DATABASE_URL")
            .unwrap_or_else(|_| "postgres://postgres:test@localhost:15432/awa_test".to_string())
    }

    fn db_test_mutex() -> &'static tokio::sync::Mutex<()> {
        static MUTEX: OnceLock<tokio::sync::Mutex<()>> = OnceLock::new();
        MUTEX.get_or_init(|| tokio::sync::Mutex::new(()))
    }

    async fn ensure_database_exists(url: &str) {
        let parts = url
            .rsplit_once('/')
            .expect("DATABASE_URL must include a database name");
        let database_name = parts.1.to_string();
        let admin_url = format!("{}/postgres", parts.0);
        let admin_pool = PgPoolOptions::new()
            .max_connections(1)
            .connect(&admin_url)
            .await
            .expect("Failed to connect to admin database for maintenance tests");
        let create_sql = format!("CREATE DATABASE {database_name}");
        match sqlx::query(audited_sql(create_sql))
            .execute(&admin_pool)
            .await
        {
            Ok(_) => {}
            Err(sqlx::Error::Database(db_err)) if db_err.code().as_deref() == Some("42P04") => {}
            Err(err) => panic!("Failed to create maintenance test database {database_name}: {err}"),
        }
    }

    async fn setup_pool(max_connections: u32) -> PgPool {
        let url = database_url();
        ensure_database_exists(&url).await;
        PgPoolOptions::new()
            .max_connections(max_connections)
            .acquire_timeout(Duration::from_secs(5))
            .connect(&url)
            .await
            .expect("Failed to connect to maintenance test database")
    }

    async fn reset_schema(pool: &PgPool) {
        sqlx::raw_sql("DROP SCHEMA IF EXISTS awa CASCADE")
            .execute(pool)
            .await
            .expect("Failed to drop awa schema");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn queue_storage_metrics_query_uses_bounded_observability_path() {
        let _guard = db_test_mutex().lock().await;
        let pool = setup_pool(4).await;
        reset_schema(&pool).await;
        migrations::run(&pool)
            .await
            .expect("migrations should succeed");

        let store = QueueStorage::new(QueueStorageConfig::default()).expect("queue storage");
        store.install(&pool).await.expect("queue storage install");

        let runtime = crate::storage::QueueStorageRuntime::new(
            QueueStorageConfig::default(),
            Duration::from_millis(1000),
            Duration::from_millis(250),
        )
        .expect("queue storage runtime");
        let service = MaintenanceService::new(
            pool,
            metrics_for_test(),
            Arc::new(AtomicBool::new(true)),
            Arc::new(AtomicBool::new(true)),
            CancellationToken::new(),
            Arc::new(Vec::new()),
            InFlightMap::default(),
            RuntimeStorage::QueueStorage(runtime.clone()),
            Arc::new(HashMap::new()),
            Arc::new(HashMap::new()),
        );

        let mut receipt_lock_tx = service
            .pool
            .begin()
            .await
            .expect("begin receipt lock transaction");
        sqlx::query(
            "LOCK TABLE awa.lease_claims, awa.lease_claim_batches, awa.lease_claim_closures, awa.lease_claim_closure_batches IN ACCESS EXCLUSIVE MODE",
        )
        .execute(receipt_lock_tx.as_mut())
        .await
        .expect("lock receipt tables");

        tokio::time::timeout(
            Duration::from_secs(3),
            service.publish_queue_storage_health_metrics(&runtime),
        )
        .await
        .expect("queue-storage metrics must not wait on receipt tables");

        receipt_lock_tx
            .rollback()
            .await
            .expect("release receipt table locks");
    }

    /// #456 mirror: a queue-storage-resolved maintenance leader must keep
    /// rescuing stuck canonical rows until finalize — the drain-only
    /// runtimes that execute them may no longer hold leadership, and
    /// unrescued rows pin `canonical_live_backlog` above zero.
    ///
    /// Runs on its own database: it holds the storage transition in
    /// `mixed_transition` with live canonical rows, which would trip the
    /// ADR-037 migrate gate for tests sharing the default database.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn queue_storage_leader_rescues_canonical_plane() {
        let base_url = database_url();
        let url = format!("{}_rescue_mirror", base_url.trim_end_matches('/'));
        ensure_database_exists(&url).await;
        let pool = PgPoolOptions::new()
            .max_connections(4)
            .acquire_timeout(Duration::from_secs(5))
            .connect(&url)
            .await
            .expect("Failed to connect to rescue-mirror test database");
        reset_schema(&pool).await;
        migrations::run(&pool)
            .await
            .expect("migrations should succeed");

        let store = QueueStorage::new(QueueStorageConfig::default()).expect("queue storage");
        store.install(&pool).await.expect("queue storage install");

        // A fresh install auto-finalizes straight to queue storage, but the
        // scenario under test is a *mixed transition*: canonical rows still
        // live while routing has flipped. Rewind to canonical, seed the
        // stuck rows, then drive prepare → mixed_transition the way the
        // staged-upgrade SQL does.
        sqlx::query(
            r#"
            UPDATE awa.storage_transition_state
            SET current_engine = 'canonical',
                prepared_engine = NULL,
                state = 'canonical',
                transition_epoch = transition_epoch + 1,
                details = '{}'::jsonb,
                updated_at = now(),
                finalized_at = NULL
            WHERE singleton
            "#,
        )
        .execute(&pool)
        .await
        .expect("rewind transition state to canonical");

        // A canonical job whose worker died mid-run…
        let stale_running: i64 = sqlx::query_scalar(
            r#"
            INSERT INTO awa.jobs_hot (kind, queue, state, attempt, heartbeat_at, attempted_at)
            VALUES ('rescue_mirror', 'rescue_mirror_q', 'running', 1, now() - interval '1 hour', now())
            RETURNING id
            "#,
        )
        .fetch_one(&pool)
        .await
        .expect("seed stale canonical running job");

        // …and one parked on a callback that has timed out. Callback-carrying
        // jobs never migrate off the canonical plane during a transition, so
        // this sweep has no other owner once drain-only runtimes lose
        // leadership.
        let expired_callback: i64 = sqlx::query_scalar(
            r#"
            INSERT INTO awa.jobs_hot (
                kind, queue, state, attempt, attempted_at,
                callback_id, callback_timeout_at
            )
            VALUES (
                'rescue_mirror', 'rescue_mirror_q', 'waiting_external', 1, now(),
                gen_random_uuid(), now() - interval '1 hour'
            )
            RETURNING id
            "#,
        )
        .fetch_one(&pool)
        .await
        .expect("seed expired-callback canonical job");

        sqlx::query(
            r#"
            INSERT INTO awa.runtime_instances (
                instance_id, hostname, pid, version, storage_capability,
                transition_role, started_at, last_seen_at, snapshot_interval_ms,
                healthy, postgres_connected, poll_loop_alive, heartbeat_alive,
                maintenance_alive, shutting_down, leader, global_max_workers,
                queues, queue_descriptor_hashes, job_kind_descriptor_hashes
            )
            VALUES (
                gen_random_uuid(), 'rescue-mirror-test', 0, '0.0.0-test',
                'queue_storage', 'queue_storage_target',
                now(), now(), 5000, TRUE, TRUE, TRUE, TRUE, TRUE,
                FALSE, FALSE, 1,
                '[]'::jsonb, '[]'::jsonb, '[]'::jsonb
            )
            "#,
        )
        .execute(&pool)
        .await
        .expect("stamp queue_storage_target runtime");
        sqlx::query("SELECT awa.storage_prepare('queue_storage', '{\"schema\": \"awa\"}'::jsonb)")
            .execute(&pool)
            .await
            .expect("storage_prepare");
        sqlx::query("SELECT awa.storage_enter_mixed_transition()")
            .execute(&pool)
            .await
            .expect("storage_enter_mixed_transition");

        let runtime = crate::storage::QueueStorageRuntime::new(
            QueueStorageConfig::default(),
            Duration::from_millis(1000),
            Duration::from_millis(250),
        )
        .expect("queue storage runtime");
        let service = MaintenanceService::new(
            pool.clone(),
            metrics_for_test(),
            Arc::new(AtomicBool::new(true)),
            Arc::new(AtomicBool::new(true)),
            CancellationToken::new(),
            Arc::new(Vec::new()),
            InFlightMap::default(),
            RuntimeStorage::QueueStorage(runtime),
            Arc::new(HashMap::new()),
            Arc::new(HashMap::new()),
        );

        service.rescue_stale_heartbeats().await;
        service.rescue_expired_callbacks().await;

        // Post-flip the `awa.jobs` view exposes only the queue-storage
        // plane, so assert against the physical canonical tables: both
        // rescues move `retryable` rows into `scheduled_jobs`.
        let state: String =
            sqlx::query_scalar("SELECT state::text FROM awa.scheduled_jobs WHERE id = $1")
                .bind(stale_running)
                .fetch_one(&pool)
                .await
                .expect("read rescued stale-heartbeat job");
        assert_eq!(state, "retryable");

        let (state, callback_id): (String, Option<uuid::Uuid>) =
            sqlx::query_as("SELECT state::text, callback_id FROM awa.scheduled_jobs WHERE id = $1")
                .bind(expired_callback)
                .fetch_one(&pool)
                .await
                .expect("read rescued callback job");
        assert_eq!(state, "retryable");
        assert_eq!(callback_id, None, "rescue must clear the callback park");

        let leftover: i64 =
            sqlx::query_scalar("SELECT count(*) FROM awa.jobs_hot WHERE kind = 'rescue_mirror'")
                .fetch_one(&pool)
                .await
                .expect("count leftover canonical hot rows");
        assert_eq!(leftover, 0, "no stuck canonical rows may remain");
    }

    #[test]
    fn branch_tracker_initial_state_has_no_history() {
        let tracker = MaintenanceBranchTracker::new();
        assert_eq!(tracker.snapshot("promote_scheduled"), None);
    }

    #[test]
    fn branch_tracker_finish_records_last_duration() {
        let tracker = MaintenanceBranchTracker::new();
        let metrics = metrics_for_test();
        let timer = tracker
            .try_begin("promote_scheduled", Duration::from_secs(1), &metrics)
            .expect("first tick should not be skipped");
        // Body would run here in production; for the test we just finish.
        timer.finish();
        let (last_duration, is_delayed) = tracker
            .snapshot("promote_scheduled")
            .expect("snapshot should exist after one finish");
        assert!(last_duration.is_some());
        assert!(
            !is_delayed,
            "first tick has no prior duration → not delayed"
        );
    }

    /// Drive `n` consecutive ticks on a virtual clock that advances by
    /// `tick_interval` before each one. When the body would have run
    /// (`try_begin_at` returns Some), call `record_finish` with
    /// `body_duration` to simulate the body completing in that time.
    /// When the body is skipped (cooldown returns None), `last_duration`
    /// is intentionally NOT updated — matching production where a
    /// skipped tick doesn't produce a new sample. Returns one bool per
    /// tick: true if the body ran, false if it was skipped.
    ///
    /// This helper does NOT seed `last_duration` itself. Tests that need
    /// an initial sample (e.g., to drive the very first overrun
    /// observation) must call `seed_last_duration` first.
    fn replay_ticks(
        tracker: &MaintenanceBranchTracker,
        clock: &mut Instant,
        branch: &'static str,
        body_duration: Duration,
        tick_interval: Duration,
        n: u32,
    ) -> Vec<bool> {
        let metrics = metrics_for_test();
        let mut ran = Vec::with_capacity(n as usize);
        for _ in 0..n {
            *clock += tick_interval;
            let did_run = tracker
                .try_begin_at(branch, tick_interval, &metrics, true, *clock)
                .is_some();
            ran.push(did_run);
            if did_run {
                tracker.record_finish(branch, body_duration);
            }
        }
        ran
    }

    /// Explicitly seed `last_duration` as if a prior body completed in
    /// `dur`. Use to set up the initial state for tests that need
    /// the first `try_begin` to see a sample.
    fn seed_last_duration(tracker: &MaintenanceBranchTracker, branch: &'static str, dur: Duration) {
        tracker
            .branches
            .lock()
            .unwrap()
            .entry(branch)
            .or_default()
            .last_duration = Some(dur);
    }

    const TICK_100MS: Duration = Duration::from_millis(100);

    /// Tracker for "cleanup" driven through the K-th overrun, so the next
    /// tick falls inside the armed cooldown. Returns the tracker, the
    /// virtual clock at the flip, and the cooldown deadline it armed.
    fn tracker_flipped_by_250ms_bodies() -> (MaintenanceBranchTracker, Instant, Instant) {
        let tracker = MaintenanceBranchTracker::new();
        let mut clock = Instant::now();
        seed_last_duration(&tracker, "cleanup", Duration::from_millis(250));
        replay_ticks(
            &tracker,
            &mut clock,
            "cleanup",
            Duration::from_millis(250),
            TICK_100MS,
            OVERRUN_HYSTERESIS_K,
        );
        let (deadline, _, _) = tracker.cooldown_snapshot("cleanup").expect("snapshot");
        let deadline = deadline.expect("flip arms a cooldown");
        (tracker, clock, deadline)
    }

    #[test]
    fn branch_tracker_single_overrun_does_not_flip() {
        // A single overrun sample (clearly above the upper threshold)
        // shouldn't flip is_delayed — K-consecutive is required.
        let tracker = MaintenanceBranchTracker::new();
        let mut clock = Instant::now();
        seed_last_duration(&tracker, "cleanup", Duration::from_millis(200));
        replay_ticks(
            &tracker,
            &mut clock,
            "cleanup",
            Duration::from_millis(200), // upper threshold for 100ms tick is 150
            TICK_100MS,
            1,
        );
        assert!(
            !tracker.snapshot("cleanup").unwrap().1,
            "single overrun must not flip is_delayed (K=3 required)"
        );
    }

    #[test]
    fn branch_tracker_deadband_sample_does_not_advance_counters() {
        // Samples between LOWER (70ms) and UPPER (150ms) of a 100ms
        // tick — e.g., 101ms — must NOT advance either counter. This
        // is the fix the bench post-mortem prescribed: 49ms-vs-51ms
        // flap at the boundary no longer accumulates toward a flip.
        let tracker = MaintenanceBranchTracker::new();
        let mut clock = Instant::now();
        seed_last_duration(&tracker, "cleanup", Duration::from_millis(101));
        replay_ticks(
            &tracker,
            &mut clock,
            "cleanup",
            Duration::from_millis(101),
            TICK_100MS,
            5,
        );
        let (cooldown_until, overrun, ontime) =
            tracker.cooldown_snapshot("cleanup").expect("snapshot");
        assert!(
            cooldown_until.is_none(),
            "deadband samples should not arm cooldown"
        );
        assert_eq!(overrun, 0, "deadband sample 101ms must not advance overrun");
        assert_eq!(ontime, 0, "deadband sample 101ms must not advance ontime");
    }

    #[test]
    fn branch_tracker_k_consecutive_overruns_flips_and_arms_cooldown() {
        // After K=3 consecutive samples clearly above UPPER, flip to
        // delayed AND arm cooldown. The third try_begin consumes the
        // K-th sample, crosses the threshold, and returns None (the
        // flip-tick skips the body so we don't immediately do more
        // expensive work).
        let tracker = MaintenanceBranchTracker::new();
        let mut clock = Instant::now();
        seed_last_duration(&tracker, "cleanup", Duration::from_millis(250));
        let ran = replay_ticks(
            &tracker,
            &mut clock,
            "cleanup",
            Duration::from_millis(250),
            TICK_100MS,
            OVERRUN_HYSTERESIS_K,
        );
        assert_eq!(ran, vec![true, true, false], "flip-tick skips body");
        let (cooldown_until, overrun, _) = tracker.cooldown_snapshot("cleanup").expect("snapshot");
        assert!(tracker.snapshot("cleanup").unwrap().1, "flipped to delayed");
        assert_eq!(overrun, OVERRUN_HYSTERESIS_K);
        // The cooldown is a wall-clock duration paced by the offending
        // body: 2 × 250ms from the flip tick.
        assert_eq!(
            cooldown_until,
            Some(clock + Duration::from_millis(500)),
            "cooldown is 2× the overrunning body, from the flip"
        );
    }

    /// A branch whose body always overruns keeps a bounded duty cycle:
    /// each post-cooldown run re-arms a duration-paced cooldown, so the
    /// body still executes roughly every `3×duration` of wall time —
    /// never once per many seconds. This is the regression shape behind
    /// the starved deadline-rescue sweep: a ~250ms sweep on a 100ms
    /// interval must keep rescuing.
    #[test]
    fn branch_tracker_persistent_overrun_keeps_bounded_duty_cycle() {
        let (tracker, mut clock, _) = tracker_flipped_by_250ms_bodies();

        // 40 further 100ms ticks (4s) with every executed body still
        // overrunning.
        let ran = replay_ticks(
            &tracker,
            &mut clock,
            "cleanup",
            Duration::from_millis(250),
            TICK_100MS,
            40,
        );
        let executed = ran.iter().filter(|&&r| r).count();
        assert!(
            executed >= 5,
            "an always-overrunning branch must keep a ~1-in-6 duty cycle, ran {executed}/40"
        );
    }

    /// Pathologically long bodies clamp at BRANCH_COOLDOWN_MAX.
    #[test]
    fn branch_tracker_cooldown_clamps_at_cap() {
        let tracker = MaintenanceBranchTracker::new();
        let mut clock = Instant::now();
        seed_last_duration(&tracker, "cleanup", Duration::from_secs(60));
        replay_ticks(
            &tracker,
            &mut clock,
            "cleanup",
            Duration::from_secs(60),
            TICK_100MS,
            OVERRUN_HYSTERESIS_K,
        );
        let (cooldown_until, _, _) = tracker.cooldown_snapshot("cleanup").expect("snapshot");
        assert_eq!(
            cooldown_until,
            Some(clock + BRANCH_COOLDOWN_MAX),
            "cap bounds the cooldown"
        );
    }

    #[test]
    fn branch_tracker_without_cooldown_tracks_overrun_but_keeps_running() {
        let tracker = MaintenanceBranchTracker::new();
        let metrics = metrics_for_test();
        let mut clock = Instant::now();
        seed_last_duration(
            &tracker,
            "terminal_count_rollup",
            Duration::from_millis(250),
        );

        // Well past the point where a cooling-down branch would have been
        // skipped several times over.
        let ticks = OVERRUN_HYSTERESIS_K * 10;
        for _ in 0..ticks {
            clock += TICK_100MS;
            let timer =
                tracker.try_begin_at("terminal_count_rollup", TICK_100MS, &metrics, false, clock);
            assert!(
                timer.is_some(),
                "no-cooldown branches should keep running on overrun"
            );
            tracker.record_finish("terminal_count_rollup", Duration::from_millis(250));
        }

        let (cooldown_until, overrun, _) = tracker
            .cooldown_snapshot("terminal_count_rollup")
            .expect("snapshot");
        assert!(
            cooldown_until.is_none(),
            "no-cooldown branch must not arm cooldown"
        );
        assert!(overrun >= OVERRUN_HYSTERESIS_K);
        assert!(
            tracker.snapshot("terminal_count_rollup").unwrap().1,
            "overrun state should still be observable"
        );
    }

    #[test]
    fn branch_tracker_cooldown_skips_body_until_deadline() {
        // Drive past the flip, then assert every tick before the
        // deadline returns None. After the flip, `last_duration` was
        // consumed by `take()`, so the cooldown gate is the only thing
        // returning None here — exactly the production shape we want.
        let (tracker, mut clock, deadline) = tracker_flipped_by_250ms_bodies();
        let ran = replay_ticks(
            &tracker,
            &mut clock,
            "cleanup",
            Duration::from_millis(50),
            TICK_100MS,
            4,
        );
        assert!(clock < deadline, "test must stay inside the cooldown");
        assert!(
            ran.iter().all(|&r| !r),
            "every tick during cooldown must skip body"
        );
        let (cooldown_until, _, _) = tracker.cooldown_snapshot("cleanup").expect("snapshot");
        assert_eq!(
            cooldown_until,
            Some(deadline),
            "skips do not move the deadline"
        );
    }

    #[test]
    fn branch_tracker_cooldown_is_wall_clock_not_tick_count() {
        // A tick-count cooldown only drains as ticks are serviced, and
        // under CPU starvation a branch is serviced rarely. The cooldown
        // here must end on time no matter how few ticks arrive, and must
        // not end early no matter how many do.
        let (tracker, clock, deadline) = tracker_flipped_by_250ms_bodies();
        let metrics = metrics_for_test();

        // A burst of ticks all inside the cooldown does not shorten it.
        for _ in 0..1_000 {
            assert!(
                tracker
                    .try_begin_at("cleanup", TICK_100MS, &metrics, true, clock)
                    .is_none(),
                "ticks inside the cooldown are skipped however many arrive"
            );
        }
        assert!(
            tracker
                .try_begin_at(
                    "cleanup",
                    TICK_100MS,
                    &metrics,
                    true,
                    deadline - Duration::from_millis(1)
                )
                .is_none(),
            "still cooling down 1ms before the deadline"
        );

        // A single tick after the deadline runs, though only one tick
        // was serviced since the flip.
        assert!(
            tracker
                .try_begin_at("cleanup", TICK_100MS, &metrics, true, deadline)
                .is_some(),
            "first tick at the deadline runs the body"
        );
        let (cooldown_until, _, _) = tracker.cooldown_snapshot("cleanup").expect("snapshot");
        assert!(cooldown_until.is_none(), "expired cooldown is cleared");
    }

    #[test]
    fn branch_tracker_cooldown_expires_then_body_runs() {
        // After cooldown expires, the first post-cooldown try_begin
        // sees `last_duration = None` (consumed at the flip, no body
        // ran during cooldown), so no hysteresis check fires; the
        // body simply runs. The counters do NOT advance on this
        // first post-cooldown tick — the sample this body produces
        // is evaluated on the *next* try_begin.
        let (tracker, _, deadline) = tracker_flipped_by_250ms_bodies();
        let mut clock = deadline;
        let ran = replay_ticks(
            &tracker,
            &mut clock,
            "cleanup",
            Duration::from_millis(50),
            TICK_100MS,
            1,
        );
        assert_eq!(ran, vec![true], "post-cooldown body runs");
        let (cooldown_until, overrun, ontime) =
            tracker.cooldown_snapshot("cleanup").expect("snapshot");
        assert!(
            cooldown_until.is_none(),
            "cooldown stays cleared after a single fast body"
        );
        assert_eq!(
            overrun, OVERRUN_HYSTERESIS_K,
            "consecutive_overrun preserved across cooldown (no eval on this tick)"
        );
        assert_eq!(
            ontime, 0,
            "ontime advances on the next tick — this one had no sample to evaluate"
        );
    }

    #[test]
    fn branch_tracker_cooldown_rearms_on_continued_overrun() {
        // After cooldown expires, the first post-cooldown body runs
        // and is slow. That slow body's sample is evaluated on the
        // *second* post-cooldown try_begin, where it re-arms cooldown
        // because consecutive_overrun is already at K (preserved
        // across the cooldown).
        let (tracker, _, deadline) = tracker_flipped_by_250ms_bodies();
        let mut clock = deadline;

        // Tick 1 post-cooldown: take None, no eval, body runs slow.
        // Tick 2: take Some(250), consecutive_overrun saturates past
        // K, is_delayed && cross → re-arm cooldown.
        let ran = replay_ticks(
            &tracker,
            &mut clock,
            "cleanup",
            Duration::from_millis(250),
            TICK_100MS,
            2,
        );
        assert_eq!(
            ran,
            vec![true, false],
            "first tick runs body; second tick re-arms cooldown"
        );
        let (cooldown_until, _, _) = tracker.cooldown_snapshot("cleanup").expect("snapshot");
        assert_eq!(
            cooldown_until,
            Some(clock + Duration::from_millis(500)),
            "cooldown re-armed, paced by the slow body"
        );
        assert!(
            tracker.snapshot("cleanup").unwrap().1,
            "still delayed across re-arm"
        );
    }

    #[test]
    fn branch_tracker_intermittent_overrun_does_not_flip() {
        // Pattern: 3 clearly-overrun samples interleaved with
        // clearly-on-time samples. Non-consecutive overruns never
        // accumulate K=3 because each on-time tick resets the counter.
        let tracker = MaintenanceBranchTracker::new();
        let mut clock = Instant::now();
        seed_last_duration(&tracker, "cleanup", Duration::from_millis(200));
        for over in [true, false, true, false, true] {
            let dur = if over {
                Duration::from_millis(200)
            } else {
                Duration::from_millis(50)
            };
            replay_ticks(&tracker, &mut clock, "cleanup", dur, TICK_100MS, 1);
        }
        assert!(
            !tracker.snapshot("cleanup").unwrap().1,
            "intermittent overruns must not flip"
        );
    }

    #[test]
    fn branch_tracker_recovers_only_after_k_ontime_ticks_post_cooldown() {
        // After cooldown expires, K consecutive on-time samples must be
        // *evaluated* before is_delayed clears. The very first
        // post-cooldown body has no sample to evaluate (last_duration
        // was consumed at the flip), so K evaluable samples require
        // K+1 post-cooldown ticks: one to seed, K to advance ontime.
        let (tracker, _, deadline) = tracker_flipped_by_250ms_bodies();
        let mut clock = deadline;

        // K post-cooldown ticks: tick 1 seeds, ticks 2..K evaluate K-1
        // on-time samples. Still delayed.
        replay_ticks(
            &tracker,
            &mut clock,
            "cleanup",
            Duration::from_millis(50),
            TICK_100MS,
            OVERRUN_HYSTERESIS_K,
        );
        assert!(
            tracker.snapshot("cleanup").unwrap().1,
            "still delayed after only K-1 evaluations"
        );

        // (K+1)-th tick evaluates the K-th on-time sample → recovery.
        replay_ticks(
            &tracker,
            &mut clock,
            "cleanup",
            Duration::from_millis(50),
            TICK_100MS,
            1,
        );
        assert!(
            !tracker.snapshot("cleanup").unwrap().1,
            "recovered after K evaluable on-time samples"
        );
    }

    #[test]
    fn branch_tracker_per_branch_state_is_independent() {
        // Drive "cleanup" past the K=3 overrun threshold while
        // "promote_scheduled" stays on-time the whole time. Branches
        // must not share state.
        let tracker = MaintenanceBranchTracker::new();
        let mut clock = Instant::now();
        seed_last_duration(&tracker, "cleanup", Duration::from_millis(500));
        replay_ticks(
            &tracker,
            &mut clock,
            "cleanup",
            Duration::from_millis(500),
            TICK_100MS,
            OVERRUN_HYSTERESIS_K,
        );
        seed_last_duration(&tracker, "promote_scheduled", Duration::from_millis(10));
        replay_ticks(
            &tracker,
            &mut clock,
            "promote_scheduled",
            Duration::from_millis(10),
            Duration::from_millis(250),
            OVERRUN_HYSTERESIS_K,
        );
        assert!(tracker.snapshot("cleanup").unwrap().1);
        assert!(!tracker.snapshot("promote_scheduled").unwrap().1);
    }

    #[tokio::test(start_paused = true)]
    async fn leader_loop_interval_does_not_burst_after_a_stall() {
        let period = Duration::from_millis(100);
        let mut interval = leader_loop_interval(period);
        interval.tick().await;

        // The loop is stalled for ten periods; a `Burst` interval would
        // now hand back ten immediate catch-up ticks.
        tokio::time::sleep(period * 10).await;
        interval.tick().await;

        let before_next = tokio::time::Instant::now();
        interval.tick().await;
        assert!(
            before_next.elapsed() >= period,
            "the tick after a late one waits a full period instead of catching up"
        );
    }

    // ── PruneBackoffTracker (#169) ────────────────────────────────────

    fn skip_active(slot: i32) -> PruneOutcome {
        PruneOutcome::SkippedActive {
            slot,
            reason: SkipReason::LeaseActive,
            count: 1,
        }
    }

    #[test]
    fn prune_backoff_initial_state_does_not_skip() {
        let tracker = PruneBackoffTracker::new();
        assert!(!tracker.should_skip(PRUNE_BRANCH_LEASE));
        assert_eq!(
            tracker.snapshot(PRUNE_BRANCH_LEASE),
            Some((0, 0)),
            "polling once must not introduce backoff"
        );
    }

    #[test]
    fn prune_backoff_skipped_active_doubles_then_resets_on_pruned() {
        let tracker = PruneBackoffTracker::new();

        // First failure: backoff_level=1, skip the next 2 ticks.
        tracker.record_outcome(PRUNE_BRANCH_LEASE, &skip_active(0));
        assert_eq!(tracker.snapshot(PRUNE_BRANCH_LEASE), Some((2, 1)));
        assert!(tracker.should_skip(PRUNE_BRANCH_LEASE));
        assert!(tracker.should_skip(PRUNE_BRANCH_LEASE));
        assert!(!tracker.should_skip(PRUNE_BRANCH_LEASE));

        // Second failure: backoff_level=2, skip the next 4.
        tracker.record_outcome(PRUNE_BRANCH_LEASE, &skip_active(0));
        assert_eq!(tracker.snapshot(PRUNE_BRANCH_LEASE), Some((4, 2)));

        // Recovery: Pruned clears everything.
        tracker.record_outcome(
            PRUNE_BRANCH_LEASE,
            &PruneOutcome::Pruned {
                slot: 0,
                generation: 0,
                carried_failed_rows: 0,
                durations: Default::default(),
            },
        );
        assert_eq!(tracker.snapshot(PRUNE_BRANCH_LEASE), Some((0, 0)));
        assert!(!tracker.should_skip(PRUNE_BRANCH_LEASE));
    }

    #[test]
    fn prune_backoff_lock_contention_retries_next_tick() {
        let tracker = PruneBackoffTracker::new();
        tracker.record_outcome(PRUNE_BRANCH_QUEUE, &skip_active(0));
        tracker.record_outcome(PRUNE_BRANCH_QUEUE, &skip_active(0));
        assert!(tracker.should_skip(PRUNE_BRANCH_QUEUE));
        for _ in 0..64 {
            tracker.record_outcome(PRUNE_BRANCH_QUEUE, &PruneOutcome::Blocked { slot: 0 });
            assert_eq!(tracker.snapshot(PRUNE_BRANCH_QUEUE), Some((0, 0)));
            assert!(!tracker.should_skip(PRUNE_BRANCH_QUEUE));
        }
        tracker.record_outcome(PRUNE_BRANCH_QUEUE, &skip_active(0));
        assert_eq!(tracker.snapshot(PRUNE_BRANCH_QUEUE), Some((2, 1)));
    }

    #[test]
    fn prune_backoff_noop_is_neutral() {
        let tracker = PruneBackoffTracker::new();
        // Build up some backoff first so we have observable state.
        tracker.record_outcome(PRUNE_BRANCH_LEASE, &skip_active(0));
        let before = tracker.snapshot(PRUNE_BRANCH_LEASE);
        tracker.record_outcome(PRUNE_BRANCH_LEASE, &PruneOutcome::Noop);
        let after = tracker.snapshot(PRUNE_BRANCH_LEASE);
        assert_eq!(
            before, after,
            "Noop must not change backoff state — there was nothing to do, not a failure"
        );
    }

    #[test]
    fn idle_prune_suppression_requires_exact_oldest_generation() {
        let tracker = PruneBackoffTracker::new();
        tracker.record_outcome(
            PRUNE_BRANCH_QUEUE,
            &PruneOutcome::Pruned {
                slot: 0,
                generation: 0,
                carried_failed_rows: 0,
                durations: Default::default(),
            },
        );

        assert_eq!(
            tracker.already_pruned_for_idle_cursor(PRUNE_BRANCH_QUEUE, 3, 3, 4),
            Some((0, 0)),
            "a frozen fully initialized cursor must suppress the repeated prune"
        );
        assert_eq!(
            tracker.already_pruned_for_idle_cursor(PRUNE_BRANCH_QUEUE, 0, 4, 4),
            None,
            "a cursor advance changes the oldest target and must re-enable prune"
        );
        assert_eq!(
            tracker.already_pruned_for_idle_cursor(PRUNE_BRANCH_QUEUE, 1, 1, 4),
            Some((0, 0)),
            "partial-ring startup derives the same generation as storage prune"
        );
    }

    #[test]
    fn prune_backoff_caps_at_max_level() {
        let tracker = PruneBackoffTracker::new();
        // Drive past the cap to make sure backoff_level saturates.
        for _ in 0..(MAX_PRUNE_BACKOFF_LEVEL as u32 + 5) {
            tracker.record_outcome(PRUNE_BRANCH_LEASE, &skip_active(0));
        }
        let (skip_remaining, backoff_level) =
            tracker.snapshot(PRUNE_BRANCH_LEASE).expect("snapshot");
        assert_eq!(backoff_level, MAX_PRUNE_BACKOFF_LEVEL);
        assert_eq!(skip_remaining, 1u32 << MAX_PRUNE_BACKOFF_LEVEL);
    }

    #[test]
    fn prune_backoff_per_branch_state_is_independent() {
        let tracker = PruneBackoffTracker::new();
        tracker.record_outcome(PRUNE_BRANCH_LEASE, &skip_active(0));
        tracker.record_outcome(PRUNE_BRANCH_LEASE, &skip_active(0));
        // Claim branch is untouched.
        assert_eq!(tracker.snapshot(PRUNE_BRANCH_LEASE), Some((4, 2)));
        assert_eq!(tracker.snapshot(PRUNE_BRANCH_CLAIM), None);
        assert!(!tracker.should_skip(PRUNE_BRANCH_CLAIM));
    }

    #[test]
    fn compute_fire_times_keeps_first_registration_latest_only() {
        let created_at = Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 30).unwrap();
        let now = Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 55).unwrap();
        let row = cron_row(
            "*/5 * * * * *",
            created_at,
            None,
            CronMissedFirePolicy::CatchUp,
        );

        let fires = compute_fire_times(&row, now, CRON_CATCH_UP_LIMIT);

        assert_eq!(
            fires,
            vec![Utc.with_ymd_and_hms(2026, 5, 7, 12, 0, 55).unwrap()]
        );
    }
}