axond 0.3.33

//! The convergence loop: observe, hydrate, compile, publish.
//!
//! One task per replica, off the request path, doing one thing: making the active
//! snapshot equal the control plane's desired revision, or reporting exactly why
//! it is not.
//!
//! # Why polling is the mechanism
//!
//! Postgres `LISTEN`/`NOTIFY` is fire-and-forget: a notification delivered while
//! a replica is reconnecting is simply gone, and a replica that treated
//! notifications as its trigger would sit on a stale snapshot indefinitely with
//! nothing to report. So the poll is the *correctness* mechanism and an ordinary
//! notification only shortens the wait ([`ChangeSignal`]). Turning ordinary
//! notifications off costs latency, never convergence — and that is what the
//! missed-notification test asserts. A force-refresh notification is deliberately
//! different: it asks the same candidate path to run even when the durable
//! revision id is unchanged.
//!
//! # Why a failed candidate cannot half-apply
//!
//! [`converge_once`](Reconciler::converge_once) never holds the running snapshot.
//! It hydrates a *complete* revision, compiles it into a whole
//! [`ConfigSnapshot`], and only then hands that snapshot to the sink in one
//! atomic store. Fetch, hydration, validation, compilation, and secret resolution
//! all fail *before* anything is published, so "the previous revision keeps
//! serving" is a consequence of the control flow rather than a cleanup path.
//!
//! # Convergence targets
//!
//! With the defaults in [`ConvergenceSettings`], a healthy replica publishes a
//! new revision within one poll interval plus its compile time — under a second
//! when notifications are delivered, within five seconds when they are not. A
//! replica that cannot reach the control plane retries on a bounded exponential
//! backoff up to 30 seconds and keeps serving, and its lag is reported the whole
//! time (see [`super::status`]).

use std::future::Future;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use tokio::sync::{Mutex as AsyncMutex, Notify};

use super::backoff::Backoff;
use super::compile::{CandidateCompiler, CompileError};
use super::lkg::{LastKnownGood, LastKnownGoodError};
use super::settings::ConvergenceSettings;
use super::status::{Clock, Rejection, RevisionReport, RevisionStatus, SnapshotSource};
use crate::backends::BackendFailure;
use crate::backends::control_plane::{ControlPlaneError, ControlPlaneStore};
use crate::desired_state::{LoadedRevision, RevisionId};
use crate::policy::{ActivationRefusal, PolicyView};
use crate::state::{AppState, ConfigSnapshot};
use crate::telemetry;

/// Where a published snapshot goes.
///
/// A trait rather than a direct [`AppState`] dependency so convergence is
/// testable without a process's worth of resources — and so the *only* thing the
/// reconciler can do to the running config is replace it wholesale.
pub trait SnapshotSink: Send + Sync {
    /// Whether this candidate's stateful policy may replace what is being
    /// enforced, decided *before* anything is published (#150).
    ///
    /// The two-step exists because policy is the one part of a candidate that
    /// cannot be judged from the candidate alone: whether a change is live, needs
    /// a drain, or needs a migration depends on the holds this replica has
    /// already granted and the durable layout it booted on. A refusal here leaves
    /// both the running config and the running policy exactly as they were.
    ///
    /// The default admits everything, for sinks that hold no policy.
    fn admit(&self, _snapshot: &ConfigSnapshot) -> Result<(), ActivationRefusal> {
        Ok(())
    }

    /// Replace the serving snapshot atomically. In-flight requests keep the
    /// snapshot they already hold.
    fn publish(&self, snapshot: ConfigSnapshot);

    /// The generation currently serving, which the next candidate increments.
    fn generation(&self) -> u64;
}

impl SnapshotSink for AppState {
    fn admit(&self, snapshot: &ConfigSnapshot) -> Result<(), ActivationRefusal> {
        self.policy().plan(&PolicyView::of(&snapshot.config))?;
        Ok(())
    }

    /// The policy is installed first and the config second, both atomically. The
    /// order matters only for the instant between them, and this is the harmless
    /// one: a request in that window is priced by the new limits under the old
    /// catalogue, never routed by a new catalogue under limits nobody published.
    fn publish(&self, snapshot: ConfigSnapshot) {
        self.policy().install(PolicyView::of(&snapshot.config));
        AppState::publish(self, snapshot);
    }

    fn generation(&self) -> u64 {
        self.config().generation
    }
}

/// A hint that desired state changed, or that the active revision must be
/// recompiled even when its durable id did not change.
///
/// Optional by construction: nothing here carries the change itself, so a lost
/// signal costs at most one poll interval. A Postgres `LISTEN` task calls
/// [`ChangeSignal::notify`]; a deployment without notifications simply never
/// does. Secret-store lifecycle changes use [`ChangeSignal::force_refresh`]:
/// they do not create a durable desired-state revision, but they can change
/// whether the active revision is materializable. The signal is deliberately
/// coalescing; the next attempt reads the current revision and therefore never
/// needs to carry a secret or a lifecycle payload through this layer.
#[derive(Debug, Default)]
pub struct ChangeSignal {
    notify: Notify,
    refresh: AtomicBool,
}

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

    /// Wake the reconciler now instead of at its next poll.
    pub fn notify(&self) {
        self.notify.notify_one();
    }

    /// Wake the reconciler and require it to recompile the current desired
    /// revision, even if that revision is already active.
    ///
    /// This is the safe bridge for out-of-band lifecycle changes such as a
    /// secret being disabled or rotated. It never publishes directly: the
    /// ordinary compile-then-admit-then-publish path still decides whether the
    /// refreshed candidate is usable, and a refusal leaves the last-known-good
    /// snapshot active. A refusal also leaves the request pending, so the retry
    /// the reconciler already schedules recompiles instead of skipping on an
    /// unchanged revision id.
    pub fn force_refresh(&self) {
        self.refresh.store(true, Ordering::Release);
        self.notify.notify_one();
    }

    async fn notified(&self) -> bool {
        self.notify.notified().await;
        self.refresh.swap(false, Ordering::Acquire)
    }
}

/// What one convergence attempt did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Outcome {
    /// A candidate was compiled and published; requests started after this see
    /// it.
    Published {
        revision: RevisionId,
        generation: u64,
        took: Duration,
    },
    /// Desired state already equals what is active. The common case, and
    /// deliberately not free of a control-plane read: that read is what detects a
    /// change *and* what proves the control plane is reachable.
    AlreadyConverged { revision: Option<RevisionId> },
    /// The control plane has published nothing yet.
    Empty,
    /// A candidate was refused. The previous revision keeps serving.
    Rejected {
        revision: Option<RevisionId>,
        reason: &'static str,
    },
}

impl Outcome {
    /// A stable label for metrics.
    pub const fn as_str(&self) -> &'static str {
        match self {
            Self::Published { .. } => "published",
            Self::AlreadyConverged { .. } => "converged",
            Self::Empty => "empty",
            Self::Rejected { .. } => "rejected",
        }
    }
}

/// A revision, or a last-known-good snapshot, this build cannot read: intact
/// storage that calls for a deployment rather than a repair. A label of both
/// `axond.revision.reason` and `axond.revision.outcome`.
pub const INCOMPATIBLE_REASON: &str = "incompatible";

/// Every value the `axond.revision.reason` label can carry: the store
/// categories [`category_reason`] classifies, the refusals
/// [`AttemptError::reason`] labels ahead of them, and the compile reasons
/// [`CompileError::reason`] returns. The catalogue enumerates this list rather
/// than a copy of it, so a new category cannot ship an uncatalogued label.
pub const REVISION_REASONS: &[&str] = &[
    "unavailable",
    "conflict",
    "not_found",
    "invalid",
    "denied",
    "corrupt",
    // Named rather than spelled, because it is the one label no store category
    // produces and a second instrument also carries it.
    INCOMPATIBLE_REASON,
    "secret",
    "projection",
    "validation",
    "pricing",
    "clock",
    "snapshot",
    // The refusals a published policy can draw before it is enforced
    // ([`crate::policy::ActivationRefusal::reason`]).
    "unsupported",
    "migration",
    "refused",
    "withdrawn",
    "ungoverned",
    "invalid_policy",
];

/// The reason label for a store failure. Exhaustive, so a new
/// [`FailureCategory`](crate::backends::FailureCategory) fails the build here
/// rather than emitting a label the
/// catalogue never declared.
pub const fn category_reason(category: crate::backends::FailureCategory) -> &'static str {
    use crate::backends::FailureCategory;
    match category {
        FailureCategory::Unavailable => "unavailable",
        FailureCategory::Conflict => "conflict",
        FailureCategory::NotFound => "not_found",
        FailureCategory::Invalid => "invalid",
        FailureCategory::Denied => "denied",
        FailureCategory::Corrupt => "corrupt",
    }
}

/// Why one attempt did not produce a published snapshot.
#[derive(Debug, thiserror::Error)]
enum AttemptError {
    #[error(transparent)]
    Store(#[from] ControlPlaneError),
    #[error(transparent)]
    Compile(#[from] CompileError),
}

impl AttemptError {
    /// The low-cardinality reason label. Store failures are classified by the
    /// backend's own category, so "unavailable" (retry) and "corrupt" (page
    /// someone) never collapse into one bucket.
    ///
    /// An incompatible revision gets its own label ahead of the category mapping:
    /// its category is a refusal a retry cannot clear, like a bound, but the
    /// action it calls for is a deployment, and reporting it as `denied` or
    /// `corrupt` would send an operator looking for the wrong thing.
    fn reason(&self) -> &'static str {
        match self {
            Self::Store(ControlPlaneError::Incompatible { .. }) => INCOMPATIBLE_REASON,
            Self::Store(error) => category_reason(error.category()),
            Self::Compile(error) => error.reason(),
        }
    }

    fn revision(&self) -> Option<RevisionId> {
        match self {
            Self::Store(ControlPlaneError::Corrupt { revision, .. })
            | Self::Store(ControlPlaneError::Incompatible { revision, .. })
            | Self::Store(ControlPlaneError::RevisionNotFound(revision))
            | Self::Store(ControlPlaneError::TooLarge { revision, .. }) => Some(*revision),
            Self::Store(_) => None,
            Self::Compile(error) => Some(error.revision()),
        }
    }
}

/// Why a stateful replica could not reach a servable snapshot at boot.
#[derive(Debug, thiserror::Error)]
pub enum BootstrapError {
    /// Reachable, but nothing has ever been published. A stateful replica has no
    /// implicit empty configuration to fall back to: it would serve a gateway
    /// with no aliases and no keys while reporting itself healthy.
    #[error(
        "the control plane is reachable but has published no revision; a stateful replica \
         has nothing to serve until desired state exists"
    )]
    Empty,
    /// The control plane was unreachable and no signed cache was available.
    #[error(
        "the control plane is unreachable and no last-known-good snapshot is available: {source}"
    )]
    Unavailable {
        #[source]
        source: ControlPlaneError,
    },
    /// The control plane answered, but with something a retry cannot clear:
    /// unreadable storage, a refused read, a revision larger than this build
    /// hydrates. Never answered from cache — cached state would mask storage an
    /// operator has to repair.
    ///
    /// A desired revision this build simply cannot *read*
    /// ([`ControlPlaneError::Incompatible`]) is the exception: storage is intact,
    /// so the cache is consulted first, and this is only reached when there is no
    /// cache to restore.
    #[error("the control plane refused to yield desired state: {source}")]
    Store {
        #[source]
        source: ControlPlaneError,
    },
    /// The desired revision exists but does not compile. Fatal at boot on
    /// purpose: there is no previous revision to keep serving.
    #[error("the desired revision cannot be served: {source}")]
    Rejected {
        #[source]
        source: Box<CompileError>,
    },
    /// The cache was consulted and refused itself: unauthentic, corrupt, or
    /// unreadable. Never downgraded to "boot empty".
    #[error("the last-known-good snapshot could not be restored: {source}")]
    Cache {
        #[source]
        source: Box<LastKnownGoodError>,
    },
}

/// One replica's convergence loop.
pub struct Reconciler {
    store: Arc<dyn ControlPlaneStore>,
    compiler: Arc<dyn CandidateCompiler>,
    sink: Arc<dyn SnapshotSink>,
    status: Arc<RevisionStatus>,
    settings: ConvergenceSettings,
    cache: Option<LastKnownGood>,
    clock: Arc<dyn Clock>,
    /// The revision the sink is serving, as this reconciler last published it.
    /// Compared against desired to decide whether to hydrate at all.
    active: Mutex<Option<RevisionId>>,
    /// Only one attempt may derive and publish a next generation at a time.
    /// This matters for the explicit refresh seam, whose callers may race with
    /// one another even though the normal loop has one owner.
    attempt_lock: AsyncMutex<()>,
    /// A refresh that has been asked for and not yet served. It outlives a single
    /// attempt because a refused candidate has not refreshed anything: the
    /// desired revision id is unchanged, so without this the backoff retry would
    /// take the already-converged skip and the request would be dropped.
    refresh_pending: AtomicBool,
    backoff: Mutex<Backoff>,
    /// Whether the last export failed, so a recovering disk is logged once
    /// rather than every attempt.
    export_failing: AtomicBool,
}

impl Reconciler {
    pub fn new(
        store: Arc<dyn ControlPlaneStore>,
        compiler: Arc<dyn CandidateCompiler>,
        sink: Arc<dyn SnapshotSink>,
        settings: ConvergenceSettings,
        cache: Option<LastKnownGood>,
        clock: Arc<dyn Clock>,
    ) -> Self {
        let status = Arc::new(RevisionStatus::new(Box::new(ArcClock(Arc::clone(&clock)))));
        Self {
            store,
            compiler,
            sink,
            status,
            backoff: Mutex::new(Backoff::new(settings.backoff)),
            settings,
            cache,
            clock,
            active: Mutex::new(None),
            attempt_lock: AsyncMutex::new(()),
            refresh_pending: AtomicBool::new(false),
            export_failing: AtomicBool::new(false),
        }
    }

    /// What this replica reports about itself: desired, loaded, active, lag, and
    /// the last refusal.
    pub fn report(&self) -> RevisionReport {
        self.status.report()
    }

    /// Shared status, for the telemetry and administrative readers that observe
    /// convergence without driving it.
    pub fn status(&self) -> &Arc<RevisionStatus> {
        &self.status
    }

    /// Reach a first servable snapshot, or explain why the replica must not
    /// start.
    ///
    /// The cache is consulted for the two failures where cached state is the
    /// better answer: the control plane being unreachable, and a desired revision
    /// this build cannot read (a newer schema during a rollout). Both leave
    /// storage intact and neither is repaired by refusing to start. Corruption, a
    /// revision past this build's bounds, and a revision that does not compile are
    /// all fatal here: booting from an older cached revision would silently serve
    /// state an operator already replaced, or hide damage.
    pub async fn bootstrap(&self) -> Result<RevisionId, BootstrapError> {
        let span = telemetry::revision_convergence_span(telemetry::CONVERGENCE_BOOT);
        let result = self.bootstrap_inner().await;
        let outcome = match &result {
            Ok(revision) => Outcome::Published {
                revision: *revision,
                generation: self.status.report().generation,
                took: self.status.report().last_convergence.unwrap_or_default(),
            },
            Err(BootstrapError::Empty) => Outcome::Empty,
            Err(_) => Outcome::Rejected {
                revision: None,
                reason: self
                    .status
                    .report()
                    .last_rejection
                    .map_or("boot", |rejection| rejection.reason),
            },
        };
        telemetry::finish_revision_convergence(
            &span,
            telemetry::CONVERGENCE_BOOT,
            &outcome,
            &self.status.report(),
        );
        result
    }

    async fn bootstrap_inner(&self) -> Result<RevisionId, BootstrapError> {
        match self.attempt().await {
            Ok(Some(published)) => Ok(published),
            Ok(None) => Err(BootstrapError::Empty),
            Err(error) => {
                self.record_failure(&error);
                match error {
                    // A cache is the answer to "this replica cannot use what the
                    // control plane holds", and being unable to *read* the desired
                    // revision is that, not a repair job: during a mixed-version
                    // rollout a replica on the older build meets a revision the
                    // newer one published, and a replica that refused to start
                    // there would withdraw capacity exactly when a rollback needs
                    // it. Corruption and a revision past this build's bounds still
                    // refuse the cache, because cached state would mask storage an
                    // operator has to fix.
                    AttemptError::Store(source)
                        if source.retryable()
                            || matches!(source, ControlPlaneError::Incompatible { .. }) =>
                    {
                        self.restore_from_cache(source).await
                    }
                    AttemptError::Store(source) => Err(BootstrapError::Store { source }),
                    AttemptError::Compile(source) => Err(BootstrapError::Rejected {
                        source: Box::new(source),
                    }),
                }
            }
        }
    }

    /// One full convergence step. Deterministic and independently callable, which
    /// is what the tests drive instead of racing the loop's timers.
    pub async fn converge_once(&self, trigger: &'static str) -> Outcome {
        self.converge_once_with_refresh(trigger, false).await
    }

    /// Recompile and republish the current desired revision.
    ///
    /// A durable revision is the normal convergence key, but some runtime
    /// inputs intentionally live outside desired state. Secret-store lifecycle
    /// transitions are the important example: they must take effect without
    /// inventing a no-op revision. This method uses exactly the same atomic
    /// candidate path as ordinary convergence and is therefore safe to call
    /// from an administrative lifecycle handler or its change listener.
    pub async fn force_refresh_once(&self, trigger: &'static str) -> Outcome {
        self.converge_once_with_refresh(trigger, true).await
    }

    async fn converge_once_with_refresh(
        &self,
        trigger: &'static str,
        force_refresh: bool,
    ) -> Outcome {
        if force_refresh {
            self.refresh_pending.store(true, Ordering::Release);
        }
        let span = telemetry::revision_convergence_span(trigger);
        let outcome = match self.attempt().await {
            Ok(Some(revision)) => {
                let report = self.status.report();
                Outcome::Published {
                    revision,
                    generation: report.generation,
                    took: report.last_convergence.unwrap_or_default(),
                }
            }
            Ok(None) => match *self.active.lock().expect("not poisoned") {
                Some(revision) => Outcome::AlreadyConverged {
                    revision: Some(revision),
                },
                None => Outcome::Empty,
            },
            Err(error) => {
                let reason = self.record_failure(&error);
                Outcome::Rejected {
                    revision: error.revision(),
                    reason,
                }
            }
        };
        telemetry::finish_revision_convergence(&span, trigger, &outcome, &self.status.report());
        outcome
    }

    /// Poll, converge, and back off, until `shutdown` completes.
    ///
    /// The wait is a race between the poll interval, a change signal, and
    /// shutdown, so a notification shortens the wait without changing what the
    /// loop does when it wakes. A failing attempt replaces the poll interval with
    /// the backoff delay, which is why an outage cannot become a hot loop.
    pub async fn run(&self, signal: Arc<ChangeSignal>, shutdown: impl Future<Output = ()> + Send) {
        let shutdown = std::pin::pin!(shutdown);
        let mut shutdown = shutdown;
        loop {
            let delay = {
                let backoff = self.backoff.lock().expect("not poisoned");
                if backoff.failures() == 0 {
                    self.settings.poll_interval
                } else {
                    backoff.delay()
                }
            };
            tokio::select! {
                biased;
                () = &mut shutdown => {
                    tracing::debug!("revision convergence stopped");
                    return;
                }
                force_refresh = signal.notified() => {
                    self.converge_once_with_refresh(
                        telemetry::CONVERGENCE_NOTIFIED,
                        force_refresh,
                    ).await;
                }
                () = tokio::time::sleep(delay) => {
                    self.converge_once(telemetry::CONVERGENCE_POLLED).await;
                }
            }
        }
    }

    /// The attempt itself: `Ok(None)` means there was nothing to do.
    ///
    /// A pending refresh is consumed here, under the attempt lock, and put back
    /// if the attempt failed, so that the retry this replica already schedules
    /// recompiles rather than skipping on an unchanged revision id.
    async fn attempt(&self) -> Result<Option<RevisionId>, AttemptError> {
        let _attempt = self.attempt_lock.lock().await;
        let force_refresh = self.refresh_pending.swap(false, Ordering::AcqRel);
        let result = self.attempt_inner(force_refresh).await;
        if force_refresh && result.is_err() {
            self.refresh_pending.store(true, Ordering::Release);
        }
        result
    }

    async fn attempt_inner(&self, force_refresh: bool) -> Result<Option<RevisionId>, AttemptError> {
        let started = self.clock.now();
        // The cheap read first. It answers "is there anything to do?" without
        // hydrating bodies, and it is also this replica's liveness check against
        // the control plane, which is why it runs even when converged.
        let desired = self.store.desired_revision().await?;
        self.status.observe_desired(desired);
        let active = *self.active.lock().expect("not poisoned");
        if desired.is_none() || (!force_refresh && desired == active) {
            self.backoff.lock().expect("not poisoned").succeed();
            return Ok(None);
        }

        // Hydration is a single consistent read of the *complete* revision
        // (#166), not a re-read of the id above: if a newer revision was
        // published between the two calls, converging straight to it is correct.
        let Some(revision) = self.store.load_desired_revision().await? else {
            self.backoff.lock().expect("not poisoned").succeed();
            return Ok(None);
        };
        self.status.observe_desired(Some(revision.id()));
        self.publish(revision, SnapshotSource::ControlPlane, started)
            .await
            .map(Some)
            .map_err(AttemptError::from)
    }

    /// Compile and publish a hydrated revision.
    ///
    /// Compilation happens before the sink is touched at all, so the failure path
    /// leaves the running snapshot exactly as it was.
    async fn publish(
        &self,
        revision: LoadedRevision,
        source: SnapshotSource,
        started: std::time::Instant,
    ) -> Result<RevisionId, CompileError> {
        let id = revision.id();
        let generation = self.sink.generation().saturating_add(1);
        // Includes unwrapping this candidate's secret material, which is why it
        // awaits. A store outage or a version that will not unwrap therefore
        // fails *here*, before the sink is touched, and the previous snapshot
        // keeps serving.
        let snapshot = self.compiler.compile(&revision, generation).await?;
        // Asked before publishing, so an unsafe policy transition costs a
        // rejected candidate rather than a half-applied one.
        self.sink
            .admit(&snapshot)
            .map_err(|source| CompileError::Activation {
                revision: id,
                source,
            })?;
        self.status.observe_loaded(id);
        // How many secret versions this snapshot holds, read before it is
        // published because publishing moves it. A count, never a reference and
        // never material: enough for an operator to see that a rotation's new
        // version is in service, with nothing to leak.
        let materialized = snapshot.secrets().len();
        // Read before publication, because the snapshot is handed away whole.
        let pricing = snapshot.pricing().map(|pricing| {
            (
                pricing.book(),
                pricing.checksum(),
                pricing.catalog(),
                pricing.is_approved(),
                pricing.targets().len(),
                pricing.effective().ends(),
            )
        });

        self.sink.publish(snapshot);
        *self.active.lock().expect("not poisoned") = Some(id);
        self.backoff.lock().expect("not poisoned").succeed();
        let took = self.clock.now().saturating_duration_since(started);
        self.status.record_published(id, generation, source, took);
        tracing::info!(
            revision = %id,
            generation,
            source = source.as_str(),
            took_ms = took.as_millis(),
            materialized,
            "published desired revision"
        );
        // The identities an operator needs to answer "what was this request billed
        // at", logged at publication because a snapshot is never mutated after it.
        // `priced_until` is the instant this resolution stops being the answer: a
        // rule dated after it applies to a later compilation, not to this snapshot.
        if let Some((book, checksum, catalog, approved, targets, until)) = pricing {
            tracing::info!(
                revision = %id,
                generation,
                price_book = %book,
                price_book_checksum = %checksum,
                catalog = %catalog,
                approved,
                priced_targets = targets,
                priced_until = until.map(|until| until.millis()),
                "published approved pricing"
            );
        }
        self.export(&revision);
        Ok(id)
    }

    /// Write the revision this replica just published to its signed cache.
    ///
    /// A cache failure is logged and counted, never propagated: the replica is
    /// already serving the revision, and refusing to serve because a *cache* is
    /// unwritable would turn a full disk into an outage.
    fn export(&self, revision: &LoadedRevision) {
        let Some(cache) = &self.cache else {
            return;
        };
        match cache.export(revision) {
            Ok(()) => {
                if self.export_failing.swap(false, Ordering::Relaxed) {
                    tracing::info!(
                        path = %cache.path().display(),
                        "last-known-good snapshot is writable again"
                    );
                }
                telemetry::record_last_known_good("exported");
            }
            Err(error) => {
                telemetry::record_last_known_good("export_failed");
                if !self.export_failing.swap(true, Ordering::Relaxed) {
                    tracing::warn!(
                        path = %cache.path().display(),
                        error = %error,
                        "the last-known-good snapshot could not be written; the replica keeps \
                         serving, but a cold boot during a control-plane outage will have no \
                         cached state"
                    );
                }
            }
        }
    }

    /// Boot from the signed cache because desired state is unusable on this
    /// replica: the control plane is unreachable, or the desired revision is one
    /// this build cannot read.
    async fn restore_from_cache(
        &self,
        source: ControlPlaneError,
    ) -> Result<RevisionId, BootstrapError> {
        let Some(cache) = &self.cache else {
            return Err(Self::uncached(source));
        };
        let restored = match cache.load() {
            Ok(restored) => restored,
            // A cache this build cannot *read* is the same version skew as
            // desired state being unreadable — a rollback onto an older build
            // that reuses the volume finds a cache the newer build wrote — so the
            // refusal names the skew rather than blaming the cache file, which is
            // authentic and intact. An inconsistent or unauthentic cache is still
            // the cache's own failure.
            Err(LastKnownGoodError::Integrity(integrity)) if integrity.is_incompatible() => {
                telemetry::record_last_known_good(INCOMPATIBLE_REASON);
                tracing::warn!(
                    error = %integrity,
                    "the last-known-good snapshot was written by a build this one cannot read; \
                     it cannot stand in for desired state"
                );
                return Err(Self::uncached(source));
            }
            Err(source) => {
                return Err(BootstrapError::Cache {
                    source: Box::new(source),
                });
            }
        };
        let Some(revision) = restored else {
            return Err(Self::uncached(source));
        };
        let started = self.clock.now();
        let id = self
            .publish(revision, SnapshotSource::LastKnownGood, started)
            .await
            .map_err(|source| BootstrapError::Rejected {
                source: Box::new(source),
            })?;
        telemetry::record_last_known_good("restored");
        tracing::warn!(
            revision = %id,
            error = %source,
            "desired state is not usable on this replica; booted from the signed \
             last-known-good snapshot, which may be older than desired state"
        );
        Ok(id)
    }

    /// The refusal when the cache could not answer: an outage the replica may yet
    /// ride out, or a refusal it cannot.
    fn uncached(source: ControlPlaneError) -> BootstrapError {
        if source.retryable() {
            BootstrapError::Unavailable { source }
        } else {
            BootstrapError::Store { source }
        }
    }

    /// Record a refusal and its backoff, and return the reason label.
    fn record_failure(&self, error: &AttemptError) -> &'static str {
        let reason = error.reason();
        let (failures, delay) = {
            let mut backoff = self.backoff.lock().expect("not poisoned");
            let delay = backoff.fail();
            (backoff.failures(), delay)
        };
        self.status.record_rejection(
            Rejection {
                revision: error.revision(),
                reason,
                detail: error.to_string(),
            },
            failures,
        );
        telemetry::record_revision_rejection(reason);
        tracing::warn!(
            reason,
            failures,
            retry_in_ms = delay.as_millis(),
            error = %error,
            "desired revision was not applied; the active revision keeps serving"
        );
        reason
    }
}

/// A [`Clock`] that shares one implementation between the reconciler and the
/// status it reports through, so a test's clock governs both.
struct ArcClock(Arc<dyn Clock>);

impl Clock for ArcClock {
    fn now(&self) -> std::time::Instant {
        self.0.now()
    }
}