autumn-web 0.9.0

//! The replication loop: the in-process worker that ships `SQLite`'s WAL to an
//! offsite destination, continuously, while the app serves traffic (#1628).
//!
//! # The invariant everything else hangs off
//!
//! In WAL journal mode `SQLite` **never writes the main database file except
//! during a checkpoint**. Every committed page goes to the `-wal` sidecar until
//! someone checkpoints it back. So if exactly one component is allowed to
//! checkpoint, that component can also take a byte-faithful copy of the main
//! database file at any other moment and know it is a consistent base for the
//! WAL frames that follow it.
//!
//! That component is this one. Autumn's `SQLite` pool sets
//! `PRAGMA wal_autocheckpoint = 0` when replication is enabled, and the
//! replicator holds one connection open for the process's lifetime — which also
//! stops `SQLite`'s *last connection closing* from checkpointing behind our back.
//!
//! # The tick
//!
//! ```text
//! read the -wal header
//!   no generation yet, or the salt changed unexpectedly, or the WAL shrank
//!       → open a new generation: copy + gzip the database file, upload it,
//!         then write snapshot.json as the commit marker
//! scan the frame chain from the last shipped commit
//!   → ship [shipped_offset, last commit boundary) as one segment
//! everything shipped, and the WAL is over its size budget
//!   → checkpoint(TRUNCATE), then either open the next WAL *index* inside this
//!     generation (cheap) or, once the generation is older than the snapshot
//!     interval, start a new generation with a fresh base snapshot
//! ```
//!
//! The index/generation split is what keeps this affordable. A checkpoint is
//! unavoidable — the `-wal` cannot grow without bound — but treating every
//! checkpoint as a new generation would re-upload the entire database each time
//! `max_wal_bytes` is reached, which on a busy database is gigabytes of write
//! amplification per hour. A checkpoint therefore costs one index bump; a full
//! base snapshot happens on the snapshot interval, which is also what bounds how
//! much WAL a restore has to replay.
//!
//! Two orderings carry the whole durability story:
//!
//! * a segment ends on a **commit boundary**, so a replica is never a
//!   half-transaction;
//! * a checkpoint is attempted **only when nothing is un-shipped**, so
//!   truncating the WAL can never discard bytes the destination has not
//!   acknowledged. A destination that is down therefore stalls checkpoints — the
//!   WAL grows, lag climbs, the health indicator goes `Down`, and data is kept.
//!   Disk is the thing that gets sacrificed, never durability.

// autumn-determinism-gate: production code in this module must read time and
// mint identifiers through the framework's injected seams (ClockSource /
// Entropy), never `Instant::now()` / `Utc::now()` / `SystemTime::now()` /
// `Uuid::new_v4()` directly. See CONTRIBUTING.md "Determinism seam gate"
// (issue #1797). Justify exceptions with
// #[allow(clippy::disallowed_methods, reason = "…")] at the narrowest scope.
#![cfg_attr(not(test), deny(clippy::disallowed_methods))]
// autumn-panic-gate: durability-critical module — production code path must be
// panic-free. See CONTRIBUTING.md "Request-path panic gate". Justify exceptions
// with #[allow(clippy::<lint>, reason = "…")] at the narrowest scope.
#![cfg_attr(
    not(test),
    deny(
        clippy::unwrap_used,
        clippy::expect_used,
        clippy::panic,
        clippy::unreachable,
        clippy::todo,
        clippy::unimplemented,
        clippy::indexing_slicing,
        clippy::string_slice,
        clippy::arithmetic_side_effects,
    )
)]

use std::collections::BTreeMap;
use std::fmt;
use std::io::{Read as _, Seek as _, SeekFrom, Write as _};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use chrono::{DateTime, Utc};
use diesel::SqliteConnection;
use diesel::connection::SimpleConnection as _;
use diesel::prelude::*;
use diesel::sql_types::Text;

use super::destination::{DestinationError, ReplicaDestination};
use super::restore::{self, RestoreError, RestoreOutcome};
use super::segment::{self, SegmentError, SegmentHeader, SnapshotMeta};
use super::sqlite::{self, CheckpointOutcome, SqliteError};
use super::status::ReplicationStatus;
use super::wal::{self, ScanCursor, WalError, WalHeader};
use crate::time::{ClockSource, SystemClock};

/// How long the loop sleeps between shutdown checks while waiting for the next
/// tick. Short enough that shutdown is prompt, long enough to be free.
const SHUTDOWN_POLL: Duration = Duration::from_millis(100);

/// Complete generations the retention sweep always keeps, whatever the clock
/// says. A host whose clock steps forward past the retention window would
/// otherwise expire the entire history in one sweep.
const MIN_KEPT_GENERATIONS: usize = 2;

/// How often a replicator with an ownership epoch reads the destination's
/// epochs between generations (issue #3233). Each read is one `list`.
pub const FENCE_CHECK_INTERVAL: Duration = Duration::from_secs(60);

/// Everything the replication loop needs, already resolved from config.
#[derive(Debug, Clone)]
pub struct ReplicationSettings {
    /// The `SQLite` database file being replicated.
    pub database_path: PathBuf,
    /// Destination key prefix for this app/profile (see
    /// [`segment::root_prefix`]).
    pub root: String,
    /// How often the loop ships. The effective RPO is roughly this plus the
    /// time one upload takes.
    pub sync_interval: Duration,
    /// How old a generation may get before the next checkpoint starts a fresh
    /// one with a new base snapshot, bounding how much WAL a restore replays.
    pub snapshot_interval: Duration,
    /// How large the `-wal` may get before the loop checkpoints it and opens the
    /// next WAL index inside the current generation.
    pub max_wal_bytes: u64,
    /// How far back the replica must stay restorable. Older generations are
    /// pruned, never the one needed to reconstruct the start of the window.
    pub retention: Duration,
    /// How often to prove the replica restorable by actually restoring it.
    /// `None` disables periodic verification.
    pub verify_interval: Option<Duration>,
}

/// Why a replication tick failed.
#[derive(Debug)]
#[non_exhaustive]
pub enum ReplicationError {
    /// The configured database file does not exist.
    DatabaseMissing {
        /// The path that was expected.
        path: String,
    },
    /// The database is not in WAL journal mode, so there is nothing to ship.
    NotWalMode {
        /// The journal mode `SQLite` reported.
        mode: String,
    },
    /// A generation could not be opened because the checkpoint it needs first
    /// was blocked by another connection. Transient: the next tick retries.
    CheckpointBlocked {
        /// Bytes still in the `-wal` file.
        wal_bytes: u64,
    },
    /// The `-wal` file could not be read as a WAL.
    Wal(WalError),
    /// A segment could not be framed.
    Segment(SegmentError),
    /// The destination refused or failed.
    Destination(DestinationError),
    /// Local I/O failed.
    Io {
        /// What was being attempted.
        op: &'static str,
        /// I/O detail.
        detail: String,
    },
    /// `SQLite` refused an operation.
    Sqlite(SqliteError),
    /// A newer ownership epoch writes to this replica (issue #3233). This
    /// replicator ships nothing more.
    Fenced {
        /// The epoch of this replicator.
        epoch: u64,
        /// The newest epoch at the destination.
        newest: u64,
    },
}

impl fmt::Display for ReplicationError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::DatabaseMissing { path } => write!(
                f,
                "the database file {path} does not exist, so there is nothing to replicate"
            ),
            Self::NotWalMode { mode } => write!(
                f,
                "continuous replication needs WAL journal mode, but the database reports \
                 {mode:?}. Autumn's SQLite pool sets `journal_mode = WAL`; a read-only or \
                 in-memory database cannot be replicated."
            ),
            Self::CheckpointBlocked { wal_bytes } => write!(
                f,
                "could not open a replication generation: the checkpoint it needs first was \
                 blocked by another connection ({wal_bytes} byte(s) still in the -wal). \
                 This is transient; the next tick retries."
            ),
            Self::Wal(e) => write!(f, "{e}"),
            Self::Segment(e) => write!(f, "{e}"),
            Self::Destination(e) => write!(f, "{e}"),
            Self::Io { op, detail } => write!(f, "replication {op} failed: {detail}"),
            Self::Sqlite(e) => write!(f, "{e}"),
            Self::Fenced { epoch, newest } => write!(
                f,
                "replication stopped: epoch {newest} owns this replica now, and this \
                 replicator has epoch {epoch}"
            ),
        }
    }
}

impl std::error::Error for ReplicationError {}

impl From<WalError> for ReplicationError {
    fn from(e: WalError) -> Self {
        Self::Wal(e)
    }
}
impl From<SegmentError> for ReplicationError {
    fn from(e: SegmentError) -> Self {
        Self::Segment(e)
    }
}
impl From<DestinationError> for ReplicationError {
    fn from(e: DestinationError) -> Self {
        Self::Destination(e)
    }
}
impl From<SqliteError> for ReplicationError {
    fn from(e: SqliteError) -> Self {
        Self::Sqlite(e)
    }
}

impl ReplicationError {
    fn io(op: &'static str) -> impl Fn(std::io::Error) -> Self {
        move |e| Self::Io {
            op,
            detail: e.to_string(),
        }
    }
}

/// What one tick did.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct TickReport {
    /// A new generation was opened (a base snapshot was uploaded).
    pub snapshot_taken: bool,
    /// Segments shipped this tick (at most one).
    pub segments: u64,
    /// Uncompressed WAL bytes shipped this tick.
    pub bytes: u64,
    /// WAL bytes written but not yet shipped after this tick.
    pub pending_bytes: u64,
    /// The WAL was checkpointed and truncated this tick.
    pub checkpointed: bool,
    /// The checkpoint opened the next WAL index inside the same generation
    /// (rather than forcing a fresh base snapshot).
    pub index_rotated: bool,
    /// Generations pruned by the retention sweep this tick.
    pub pruned_generations: u64,
}

/// Live state for the generation currently being shipped.
#[derive(Debug)]
struct GenerationState {
    /// Generation id (also its key prefix).
    id: String,
    /// WAL index inside this generation. `0` at the base snapshot; bumped by
    /// each of the replicator's own checkpoints.
    index: u32,
    /// The WAL salt this generation's segments must all carry. `None` until the
    /// first write after the generation opened reveals it.
    salt: Option<(u32, u32)>,
    /// Database page size, from the WAL header.
    page_size: u32,
    /// When the generation opened.
    started_at: DateTime<Utc>,
    /// Sequence number for the next segment.
    next_seq: u64,
    /// `-wal` bytes already shipped. Starts at `0`, so segment `0` carries the
    /// 32-byte WAL header and a restore can rebuild the sidecar from objects
    /// alone.
    shipped_offset: u64,
    /// Where the frame scan resumes, with its rolling checksum. `None` until a
    /// WAL header has been seen.
    cursor: Option<ScanCursor>,
}

impl GenerationState {
    /// Continue this generation after the replicator's own checkpoint.
    ///
    /// The checkpoint folded every shipped frame into the database file and
    /// restarted the WAL at offset `0` under a new salt, so no fresh base
    /// snapshot is needed: a restore replays index by index, checkpointing each
    /// one in before the next is applied.
    const fn begin_next_index(&mut self) {
        self.index = self.index.saturating_add(1);
        self.salt = None;
        self.page_size = 0;
        self.next_seq = 0;
        self.shipped_offset = 0;
        self.cursor = None;
    }
}

/// Row shape for `PRAGMA journal_mode`.
#[derive(QueryableByName)]
struct JournalModeRow {
    #[diesel(sql_type = Text)]
    journal_mode: String,
}

/// The continuous replicator.
///
/// Drive it with [`Replicator::run`] on a dedicated thread, or step it manually
/// with [`Replicator::tick`] (which is what the tests do, so the loop's
/// behaviour is asserted without sleeping).
pub struct Replicator {
    settings: ReplicationSettings,
    destination: Arc<dyn ReplicaDestination>,
    status: Arc<ReplicationStatus>,
    /// Held open for the process's lifetime so `SQLite` never runs its
    /// "last connection closing" checkpoint behind the replicator's back.
    connection: Option<SqliteConnection>,
    state: Option<GenerationState>,
    /// Set while a verification restore is running on its own thread, so a slow
    /// verification cannot pile up behind itself.
    verifying: Arc<AtomicBool>,
    /// The injected wall clock. Every artifact timestamp is read from here at
    /// the moment that artifact's contents are fenced, never earlier.
    clock: Arc<dyn ClockSource>,
    /// The ownership epoch this replicator ships under (issue #3233).
    epoch: Option<u64>,
    /// When the destination's epochs were last read.
    fence_checked_at: Option<DateTime<Utc>>,
    /// The newer epoch that stopped this replicator.
    fenced_by: Option<u64>,
}

impl fmt::Debug for Replicator {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Replicator")
            .field("settings", &self.settings)
            .field("destination", &self.destination.describe())
            .field("generation", &self.state.as_ref().map(|s| &s.id))
            .finish_non_exhaustive()
    }
}

impl Replicator {
    /// Build a replicator. Nothing happens until [`tick`](Self::tick) or
    /// [`run`](Self::run) is called.
    #[must_use]
    pub fn new(
        settings: ReplicationSettings,
        destination: Arc<dyn ReplicaDestination>,
        status: Arc<ReplicationStatus>,
    ) -> Self {
        Self {
            settings,
            destination,
            status,
            connection: None,
            state: None,
            verifying: Arc::new(AtomicBool::new(false)),
            clock: Arc::new(SystemClock),
            epoch: None,
            fence_checked_at: None,
            fenced_by: None,
        }
    }

    /// Ship under an ownership epoch (issue #3233). Each generation then
    /// records the epoch, and the replicator stops (with
    /// [`ReplicationError::Fenced`]) once the destination shows a newer one.
    #[must_use]
    pub fn with_epoch(mut self, epoch: Option<u64>) -> Self {
        self.set_epoch(epoch);
        self
    }

    /// Change the epoch. A new epoch starts a new generation on the next
    /// tick.
    pub fn set_epoch(&mut self, epoch: Option<u64>) {
        if self.epoch != epoch {
            self.epoch = epoch;
            self.state = None;
            self.fence_checked_at = None;
            self.fenced_by = None;
        }
    }

    /// The newer epoch that stopped this replicator, if one did.
    #[must_use]
    pub const fn fenced_by(&self) -> Option<u64> {
        self.fenced_by
    }

    /// Read the destination's epochs. A newer one than ours stops this
    /// replicator for good.
    fn check_fence(&mut self) -> Result<(), ReplicationError> {
        let Some(epoch) = self.epoch else {
            return Ok(());
        };
        if let Some(newest) = self.fenced_by {
            return Err(ReplicationError::Fenced { epoch, newest });
        }
        let root = self.settings.root.clone();
        let newest = self
            .destination
            .list(&segment::epochs_prefix(&root))?
            .iter()
            .filter_map(|key| segment::parse_epoch_marker_key(&root, key))
            .max();
        self.fence_checked_at = Some(self.clock.now());
        match newest {
            Some(newest) if newest > epoch => {
                self.fenced_by = Some(newest);
                Err(ReplicationError::Fenced { epoch, newest })
            }
            _ => Ok(()),
        }
    }

    /// The epoch to record on a new generation. With no epoch of its own, a
    /// replicator takes the newest epoch at the destination: ownership was
    /// turned off, and the new generations must still order after the
    /// fenced ones.
    fn generation_epoch(&self) -> Result<Option<u64>, ReplicationError> {
        if self.epoch.is_some() {
            return Ok(self.epoch);
        }
        let root = &self.settings.root;
        Ok(self
            .destination
            .list(&segment::epochs_prefix(root))?
            .iter()
            .filter_map(|key| segment::parse_epoch_marker_key(root, key))
            .max())
    }

    /// Whether the periodic fence check is due.
    fn fence_check_due(&self) -> bool {
        if self.fenced_by.is_some() {
            return true;
        }
        self.epoch.is_some()
            && self.fence_checked_at.is_none_or(|at| {
                self.clock
                    .now()
                    .signed_duration_since(at)
                    .to_std()
                    // A clock that stepped back checks now.
                    .map_or(true, |age| age >= FENCE_CHECK_INTERVAL)
            })
    }

    /// Read time from `clock` instead of the system clock.
    ///
    /// The app wires in its injected clock, and a test wires in one it can step,
    /// so a point-in-time restore can be exercised over a compressed timeline.
    #[must_use]
    pub fn with_clock(mut self, clock: Arc<dyn ClockSource>) -> Self {
        self.clock = clock;
        self
    }

    /// The destination this replicator ships to.
    #[must_use]
    pub fn destination(&self) -> &Arc<dyn ReplicaDestination> {
        &self.destination
    }

    /// Run until `shutdown` is cancelled, then ship one final time so a clean
    /// stop leaves nothing behind.
    ///
    /// Blocking: call this on a dedicated thread, never on the async runtime.
    pub fn run(mut self, shutdown: &tokio_util::sync::CancellationToken) {
        let mut last_verify = self.clock.now();
        loop {
            let now = self.clock.now();
            self.tick_and_record();

            if let Some(interval) = self.settings.verify_interval {
                let due = now
                    .signed_duration_since(last_verify)
                    .to_std()
                    .is_ok_and(|elapsed| elapsed >= interval);
                if due {
                    last_verify = now;
                    self.spawn_verification();
                }
            }

            if sleep_or_cancelled(shutdown, self.settings.sync_interval) {
                break;
            }
        }
        // Final flush: the app is stopping, so ship the last committed frames
        // before the process (and possibly the machine) goes away.
        self.tick_and_record();
    }

    /// Run one tick, folding the outcome into the shared status.
    fn tick_and_record(&mut self) {
        match self.tick() {
            Ok(report) => {
                if report.segments > 0 || report.snapshot_taken {
                    tracing::debug!(
                        segments = report.segments,
                        bytes = report.bytes,
                        snapshot = report.snapshot_taken,
                        checkpointed = report.checkpointed,
                        "SQLite replication tick"
                    );
                }
            }
            // `tick` has already folded the failure into the shared status, so
            // lag keeps growing; this arm only surfaces it in the log.
            Err(e) => tracing::warn!(error = %e, "SQLite replication tick failed"),
        }
    }

    /// Perform one replication tick.
    ///
    /// Whatever the outcome, it is recorded in the shared [`ReplicationStatus`]
    /// before returning — success advances the replication point, failure does
    /// not — so every caller (the loop, a test, a future operator command) sees
    /// the same observable state.
    ///
    /// # Errors
    ///
    /// See [`ReplicationError`]. A failed tick never advances the replication
    /// point, so lag keeps growing until shipping works again.
    pub fn tick(&mut self) -> Result<TickReport, ReplicationError> {
        let result = self.tick_inner();
        if let Err(e) = &result {
            self.status
                .record_tick_error(e.to_string(), self.clock.now());
        }
        result
    }

    /// The tick body. Separated so [`tick`](Self::tick) can record the outcome
    /// on every path without threading the status through each `?`.
    fn tick_inner(&mut self) -> Result<TickReport, ReplicationError> {
        let db = self.settings.database_path.clone();
        if !db.is_file() {
            return Err(ReplicationError::DatabaseMissing {
                path: db.display().to_string(),
            });
        }
        // A new generation checks in `start_generation`.
        if self.state.is_some() && self.fence_check_due() {
            self.check_fence()?;
        }
        self.ensure_connection(&db)?;

        let wal_path = wal::wal_path(&db);
        let wal_len = std::fs::metadata(&wal_path).map_or(0, |m| m.len());
        let header = read_wal_header(&wal_path, wal_len)?;

        let mut report = TickReport::default();
        let mut header = header;
        let mut wal_len = wal_len;
        if self.needs_new_generation(header.as_ref(), wal_len) {
            // Ship what the outgoing generation still owes before rotating.
            //
            // `start_generation` checkpoints, which folds every unshipped frame
            // into the database file the new base snapshot copies. Those commits
            // are not lost, but they leave the *old* generation's chain — so a
            // point-in-time restore aimed just before the rollover would replay
            // that chain and legitimately miss them. They would also stay
            // offsite-less for the whole compress-and-upload of a full snapshot,
            // which on a large database is minutes against an advertised
            // ten-second objective.
            //
            // Shipping first closes both: the frames land in the generation that
            // owns them, at the moment they were committed.
            if self.state.is_some()
                && let Some(pending) = header.as_ref()
            {
                let shipped = self.ship(&wal_path, pending, wal_len)?;
                report.segments = report.segments.saturating_add(shipped.0);
                report.bytes = report.bytes.saturating_add(shipped.1);
            }
            self.start_generation(&db)?;
            report.snapshot_taken = true;
            // The generation opened from a checkpointed database file, so the
            // WAL was reset underneath us; re-read it before shipping.
            wal_len = std::fs::metadata(&wal_path).map_or(0, |m| m.len());
            header = read_wal_header(&wal_path, wal_len)?;
        }

        if let Some(header) = header {
            let shipped = self.ship(&wal_path, &header, wal_len)?;
            report.segments = shipped.0;
            report.bytes = shipped.1;
        }

        // Re-measured, never reused from the top of the tick. `wal_len` was
        // sampled before an upload that can take seconds, so commits made during
        // the transfer are absent from it — reporting that stale figure as
        // `pending_bytes`, and stamping a fresh `last_success_at` beside it, made
        // `/actuator/health` show near-zero lag while the destination was a whole
        // transfer behind. A run of slow uploads could hold the indicator `Up`
        // indefinitely past the configured RPO. (`maybe_checkpoint` already
        // re-measures for the same reason; the status path did not.)
        let (pending, caught_up) = self.settled_tail(&wal_path);
        report.pending_bytes = pending;

        // Read *after* anything this tick may have opened or shipped, never
        // before. Both judgements below compare this instant against a timestamp
        // the tick itself may have just written — a generation's `started_at`,
        // an artifact's creation instant — and each of those is sampled at the
        // moment its contents were fenced, which is later than the top of the
        // tick. A reading taken up there would sit *behind* them, and
        // `generation_expired` reads a negative span as "expired": every fresh
        // generation would be retired on the tick that opened it.
        let now = self.clock.now();
        let mut index_rotated = false;
        report.checkpointed =
            self.maybe_checkpoint(&db, &wal_path, pending, now, &mut index_rotated)?;
        report.index_rotated = index_rotated;

        // Retention runs every tick, not only when a generation opens: on a
        // quiet database a generation can outlive many ticks, and a destination
        // that grows without bound is its own outage. A prune failure (a revoked
        // DeleteObject, a throttle) must not fail the tick — the data is still
        // safely offsite — so it is logged and retried next time.
        match self.prune(now) {
            Ok(pruned) => report.pruned_generations = pruned,
            Err(e) => tracing::warn!(error = %e, "SQLite replica retention sweep failed"),
        }

        if caught_up {
            self.status.record_tick_ok(pending, self.clock.now());
        } else {
            self.status.record_tick_behind(pending);
        }
        Ok(report)
    }

    /// Re-measure the tail after this tick's uploads: `(pending_bytes, caught_up)`.
    ///
    /// Only a *committed* transaction counts as owed. Bytes from a write still
    /// in flight are not data the destination is behind on, so treating any
    /// non-zero tail as "behind" would trip the health indicator on every busy
    /// database. Scanning tells the two apart.
    ///
    /// Every failure path answers "caught up". An unreadable or re-salted `-wal`
    /// is not evidence that the replica is behind, and inventing lag from it
    /// would raise an alarm nothing can clear; the tick's own error paths
    /// already record genuine failures.
    fn settled_tail(&self, wal_path: &Path) -> (u64, bool) {
        let Some(state) = self.state.as_ref() else {
            return (0, true);
        };
        let wal_len = std::fs::metadata(wal_path).map_or(0, |m| m.len());
        let pending = wal_len.saturating_sub(state.shipped_offset);
        if pending == 0 {
            return (0, true);
        }
        let (Ok(Some(header)), Some(cursor)) = (read_wal_header(wal_path, wal_len), state.cursor)
        else {
            return (pending, true);
        };
        if state.salt != Some(header.salt()) {
            return (pending, true);
        }
        let Ok(buffer) = read_range(wal_path, state.shipped_offset, wal_len) else {
            return (pending, true);
        };
        let scan_at = usize::try_from(cursor.offset.saturating_sub(state.shipped_offset))
            .unwrap_or(usize::MAX);
        let committed = wal::scan_from(&header, &cursor, buffer.get(scan_at..).unwrap_or(&[]))
            .is_ok_and(|outcome| outcome.last_commit_end > state.shipped_offset);
        (pending, !committed)
    }

    /// Checkpoint the WAL when it is safe and due, returning whether it happened.
    ///
    /// # The interlock
    ///
    /// A checkpoint folds the WAL into the database file and truncates it, so
    /// anything in the WAL that has not been shipped is gone from every artifact
    /// the replica has. `pending == 0` is computed from a `wal_len` measured
    /// *before* the tick's upload, which can take seconds — so by itself it
    /// proves nothing. Two more things make this safe:
    ///
    /// * the `-wal` is re-measured here, immediately before the checkpoint, and
    ///   must still be exactly what was shipped;
    /// * `PRAGMA data_version` brackets the checkpoint. It moves if — and only
    ///   if — another connection committed, so an unchanged value proves no
    ///   write slipped into the gap between that measurement and the checkpoint
    ///   taking `SQLite`'s write lock.
    ///
    /// When the bracket *does* move, the checkpoint may have folded away a
    /// transaction that was never shipped. That is not treated as loss: the
    /// generation is retired instead, so the next tick takes a fresh base
    /// snapshot of the database file the checkpoint just completed — which
    /// contains that transaction. The cost is one snapshot; the alternative
    /// would be silent data loss.
    fn maybe_checkpoint(
        &mut self,
        db: &Path,
        wal_path: &Path,
        pending: u64,
        now: DateTime<Utc>,
        index_rotated: &mut bool,
    ) -> Result<bool, ReplicationError> {
        let expired = self.generation_expired(now);
        let over_budget = self
            .state
            .as_ref()
            .is_some_and(|state| state.shipped_offset >= self.settings.max_wal_bytes);
        if pending != 0 || !(over_budget || expired) {
            return Ok(false);
        }
        let Some(shipped_offset) = self.state.as_ref().map(|state| state.shipped_offset) else {
            return Ok(false);
        };
        if shipped_offset == 0 {
            // Nothing in this index yet; an expired generation with an empty WAL
            // needs no checkpoint at all, just a fresh snapshot next tick.
            if expired {
                self.state = None;
            }
            return Ok(false);
        }

        let Some(conn) = self.connection.as_mut() else {
            return Ok(false);
        };
        let before = sqlite::data_version(conn)?;
        // Re-measure: the WAL may have grown while this tick was uploading.
        if std::fs::metadata(wal_path).map_or(0, |m| m.len()) != shipped_offset {
            return Ok(false);
        }
        if sqlite::checkpoint_truncate(conn, db)? != CheckpointOutcome::Truncated {
            return Ok(false);
        }
        let raced = sqlite::data_version(conn)? != before;

        if expired || raced {
            if raced {
                tracing::info!(
                    "a write landed while the SQLite WAL was being checkpointed; taking a \
                     fresh base snapshot rather than assuming it was replicated"
                );
            }
            self.state = None;
        } else if let Some(state) = self.state.as_mut() {
            // Cheap path: same base snapshot, next WAL index.
            state.begin_next_index();
            *index_rotated = true;
        }
        Ok(true)
    }

    /// Restore this replica into a scratch directory and integrity-check it, so
    /// "verified" means a restore actually ran (phase 3).
    ///
    /// # Errors
    ///
    /// See [`RestoreError`].
    pub fn verify(&self) -> Result<RestoreOutcome, RestoreError> {
        verify_replica(
            self.destination.as_ref(),
            &self.settings.root,
            &self.settings.database_path,
        )
    }

    /// Run a verification on its own thread.
    ///
    /// Verification restores the entire replica, which on a large database takes
    /// far longer than a tick. Running it inline would stall shipping for that
    /// whole time and blow the RPO the rest of this module exists to hold, so it
    /// gets its own thread — and a flag, so a verification slower than the
    /// verification interval cannot pile up behind itself.
    fn spawn_verification(&self) {
        if self.verifying.swap(true, Ordering::SeqCst) {
            tracing::warn!(
                "skipping a SQLite replica verification: the previous one is still running"
            );
            return;
        }
        let destination = Arc::clone(&self.destination);
        let status = Arc::clone(&self.status);
        let verifying = Arc::clone(&self.verifying);
        let root = self.settings.root.clone();
        let database_path = self.settings.database_path.clone();
        let clock = Arc::clone(&self.clock);
        let spawned = std::thread::Builder::new()
            .name("autumn-replica-verify".to_owned())
            .spawn(move || {
                // Cleared on drop, so a panic anywhere in the restore path
                // cannot leave the flag set and silently disable every future
                // verification.
                let _guard = InFlightGuard(verifying);
                let result = verify_replica(destination.as_ref(), &root, &database_path)
                    .map(|_| ())
                    .map_err(|e| e.to_string());
                if let Err(detail) = &result {
                    tracing::error!(
                        error = %detail,
                        "SQLite replica verification FAILED — the replica may not be restorable"
                    );
                }
                status.record_verification(result, clock.now());
            });
        if let Err(e) = spawned {
            tracing::warn!("could not start the SQLite replica verification thread: {e}");
            self.verifying.store(false, Ordering::SeqCst);
        }
    }

    /// Open (once) a private connection and prove the database is in WAL mode.
    fn ensure_connection(&mut self, db: &Path) -> Result<(), ReplicationError> {
        if self.connection.is_some() {
            return Ok(());
        }
        let mut conn = sqlite::open(db)?;
        let mode = journal_mode(&mut conn)?;
        if !mode.eq_ignore_ascii_case("wal") {
            // The app's pool sets WAL; do it here too so a replicator started
            // against a freshly created file does not have to wait for a write.
            conn.batch_execute("PRAGMA journal_mode = WAL;")
                .map_err(|e| {
                    ReplicationError::Sqlite(SqliteError::Query {
                        op: "journal_mode = WAL",
                        detail: e.to_string(),
                    })
                })?;
            let mode = journal_mode(&mut conn)?;
            if !mode.eq_ignore_ascii_case("wal") {
                return Err(ReplicationError::NotWalMode { mode });
            }
        }
        self.connection = Some(conn);
        Ok(())
    }

    /// Whether a fresh base snapshot is required before anything can be shipped.
    fn needs_new_generation(&self, header: Option<&WalHeader>, wal_len: u64) -> bool {
        match (&self.state, header) {
            // Nothing has been replicated yet.
            (None, _) => true,
            // The WAL restarted under a new salt: previously shipped byte
            // offsets no longer mean anything.
            (Some(state), Some(header)) => {
                state.salt.is_some_and(|salt| salt != header.salt())
                    || wal_len < state.shipped_offset
            }
            // The WAL vanished or shrank below what we had shipped — a
            // checkpoint happened (ours, on the previous tick, or someone
            // else's).
            (Some(state), None) => state.shipped_offset > 0 || wal_len < state.shipped_offset,
        }
    }

    /// Open a new generation: checkpoint, then upload a byte-faithful gzip of
    /// the database file, then its `snapshot.json` commit marker.
    ///
    /// The checkpoint comes first and is not optional. In WAL mode the main
    /// database file holds the state as of the *last* checkpoint, so snapshotting
    /// it while the `-wal` still carries committed frames would publish a base
    /// that is silently behind — and `snapshot.json` makes that generation the
    /// newest complete one, so a restore would then regress past everything the
    /// previous generation had already shipped. Checkpointing first makes the
    /// database file current by construction, and the new WAL starts empty at
    /// offset `0`, which is exactly where this generation's index `0` begins.
    /// The greatest `created_ms` among the generations already at the
    /// destination, or `None` when this is the first one.
    ///
    /// Only the id is parsed — no object is downloaded — so this costs the one
    /// `list` that opening a generation can well afford next to the snapshot
    /// upload that follows it. Incomplete generations count too: a generation
    /// whose snapshot metadata never landed still owns its slice of the
    /// ordering, and reusing its stamp would make two generations compare equal.
    fn newest_generation_ms(&self) -> Result<Option<i64>, ReplicationError> {
        let keys = self
            .destination
            .list(&segment::generations_prefix(&self.settings.root))?;
        Ok(keys
            .iter()
            .filter_map(|key| segment::generation_of_key(&self.settings.root, key))
            .filter_map(|id| segment::parse_generation_id(&id))
            .map(|info| info.created_ms)
            .max())
    }

    fn start_generation(&mut self, db: &Path) -> Result<(), ReplicationError> {
        // Before the checkpoint: a fenced replicator leaves the WAL alone.
        self.check_fence()?;
        if let Some(epoch) = self.epoch {
            self.destination.put(
                &segment::epoch_marker_key(&self.settings.root, epoch),
                b"{}",
            )?;
        }
        if let Some(conn) = self.connection.as_mut() {
            let outcome = sqlite::checkpoint_truncate(conn, db)?;
            if let CheckpointOutcome::Busy { wal_bytes } = outcome {
                return Err(ReplicationError::CheckpointBlocked { wal_bytes });
            }
        }
        // Stamped only now: the checkpoint is what fences the database file this
        // snapshot copies. A reading taken before it would claim the snapshot
        // holds a moment it does not yet cover, and a point-in-time restore
        // landing in that window would replay a transaction from *after* the
        // instant the operator asked for. Reading after the fence can only
        // over-state the snapshot's age, which costs the restore an extra
        // generation of WAL replay and never correctness.
        let now = self.clock.now();
        // Generation ids order by this stamp, and both "restore the latest" and
        // retention take the greatest one. A host clock that steps backward
        // between generations would stamp the *newer* generation lower, so a
        // latest-restore would silently select the older one and omit every
        // write captured after the step until wall time caught up again. That is
        // not a remote case here: `generation_expired` deliberately treats a
        // negative elapsed duration as expiry, so a backward step is precisely
        // when a new generation gets opened.
        //
        // The stamp is therefore floored one millisecond past the newest
        // generation already at the destination. Reading the floor from the
        // destination rather than from memory is what makes it durable: a
        // process that restarts with an already-rolled-back clock still orders
        // its new generation after the ones it finds offsite.
        let floor_ms = self.newest_generation_ms()?;
        let created_ms = now
            .timestamp_millis()
            .max(floor_ms.map_or(i64::MIN, |ms| ms.saturating_add(1)));
        let id = segment::generation_id(created_ms, rand::random::<u64>());

        // Before the upload: a retention sweep of another replicator orders
        // generations by epoch, and must not take this one for epoch 0 while
        // its snapshot uploads.
        if let Some(epoch) = self.generation_epoch()? {
            self.destination.put(
                &segment::generation_epoch_key(&self.settings.root, &id, epoch),
                b"{}",
            )?;
        }

        let staging = staging_path(db);
        let (sha256, uncompressed_len) = gzip_file(db, &staging)?;
        let upload = self
            .destination
            .put_file(&segment::snapshot_key(&self.settings.root, &id), &staging);
        let _ = std::fs::remove_file(&staging);
        upload?;

        let meta = SnapshotMeta {
            version: segment::LAYOUT_VERSION,
            generation: id.clone(),
            created_at: now.to_rfc3339(),
            created_ms,
            sha256,
            uncompressed_len,
        };
        let body = serde_json::to_vec(&meta).map_err(|e| ReplicationError::Io {
            op: "encode snapshot metadata",
            detail: e.to_string(),
        })?;
        // Written LAST: its presence is what makes the generation restorable.
        self.destination
            .put(&segment::snapshot_meta_key(&self.settings.root, &id), &body)?;

        tracing::info!(
            generation = %id,
            bytes = uncompressed_len,
            destination = %self.destination.describe(),
            "opened a new SQLite replication generation"
        );
        self.status.record_generation(&id, now);
        self.state = Some(GenerationState {
            id,
            index: 0,
            salt: None,
            page_size: 0,
            started_at: now,
            next_seq: 0,
            shipped_offset: 0,
            cursor: None,
        });
        Ok(())
    }

    /// Ship everything committed since the last segment. Returns
    /// `(segments, bytes)`.
    fn ship(
        &mut self,
        wal_path: &Path,
        header: &WalHeader,
        wal_len: u64,
    ) -> Result<(u64, u64), ReplicationError> {
        let root = self.settings.root.clone();
        let clock = Arc::clone(&self.clock);
        let Some(state) = self.state.as_mut() else {
            return Ok((0, 0));
        };
        if state.salt.is_none() {
            state.salt = Some(header.salt());
            state.page_size = header.page_size;
            state.cursor = Some(ScanCursor::start(header));
        }
        if state.salt != Some(header.salt()) {
            // The salt changed mid-tick; the next tick opens a new generation.
            return Ok((0, 0));
        }
        let Some(cursor) = state.cursor else {
            return Ok((0, 0));
        };
        if wal_len <= cursor.offset {
            return Ok((0, 0));
        }

        // Never read past what the decoder will accept. `decode_segment` refuses
        // any `uncompressed_len` over `MAX_SEGMENT_BYTES` — a bound it needs,
        // because the periodic verifier inflates segments inside the app process
        // — but nothing capped the *write* side. After an outage, a long sync
        // interval, or one enormous transaction, this read took the whole
        // committed tail and published a segment that no restore could ever
        // decode, while the tick went on to checkpoint the local WAL away: the
        // generation is then unrestorable until the next base snapshot.
        //
        // Capping the window instead makes the scan below stop at the last commit
        // that fits, and the next tick ships the rest. A single transaction
        // larger than the cap ships nothing, which grows lag and trips the health
        // indicator — loudly stuck rather than quietly unrestorable.
        let window_end = wal_len.min(
            state
                .shipped_offset
                .saturating_add(segment::MAX_SEGMENT_BYTES),
        );
        let buffer = read_range(wal_path, state.shipped_offset, window_end)?;
        let scan_at = usize::try_from(cursor.offset.saturating_sub(state.shipped_offset))
            .unwrap_or(usize::MAX);
        let outcome = wal::scan_from(header, &cursor, buffer.get(scan_at..).unwrap_or(&[]))?;
        if outcome.last_commit_end <= cursor.offset {
            // Frames are present but no transaction has committed yet — shipping
            // them would put half a transaction offsite.
            return Ok((0, 0));
        }

        let take = usize::try_from(outcome.last_commit_end.saturating_sub(state.shipped_offset))
            .unwrap_or(usize::MAX);
        let raw = buffer.get(..take).ok_or_else(|| ReplicationError::Io {
            op: "slice the WAL range",
            detail: format!(
                "the -wal shrank while it was being read (wanted {take} bytes, have {})",
                buffer.len()
            ),
        })?;

        // Stamped only now, once the bytes this segment carries have been read
        // out of the `-wal` and sliced at their commit boundary. `wal_len` is
        // measured at the top of the tick and the read follows it, so a
        // transaction can commit into that window and land in `raw`; a timestamp
        // sampled before the read would date the segment earlier than a change
        // it actually contains, and a restore to an instant in between would
        // replay that change. Sampling after can only date the segment late,
        // which excludes it from such a restore instead.
        let now = clock.now();
        let segment_header = SegmentHeader {
            version: segment::LAYOUT_VERSION,
            index: state.index,
            seq: state.next_seq,
            start_offset: state.shipped_offset,
            end_offset: outcome.last_commit_end,
            frame_count: outcome.frames,
            commit_count: outcome.commits,
            page_size: header.page_size,
            db_size_pages: outcome.last_commit_db_pages,
            salt1: header.salt1,
            salt2: header.salt2,
            created_at: now.to_rfc3339(),
            created_ms: now.timestamp_millis(),
            sha256: segment::sha256_hex(raw),
            uncompressed_len: raw.len() as u64,
        };
        let payload = segment::encode_segment(&segment_header, raw)?;
        let key = segment::segment_key(
            &root,
            &state.id,
            state.index,
            state.next_seq,
            segment_header.created_ms,
        );
        self.destination.put(&key, &payload)?;

        // Written after the segment it describes, never before: a head that
        // runs ahead of the objects would fail an honest restore, while a head
        // that lags one segment only under-claims, which a restore tolerates.
        let head = segment::GenerationHead {
            version: segment::LAYOUT_VERSION,
            index: state.index,
            seq: state.next_seq,
            created_ms: segment_header.created_ms,
        };
        let head_body = serde_json::to_vec(&head).map_err(|e| ReplicationError::Io {
            op: "encode the generation head",
            detail: e.to_string(),
        })?;
        self.destination
            .put(&segment::head_key(&root, &state.id), &head_body)?;

        let bytes = raw.len() as u64;
        state.shipped_offset = outcome.last_commit_end;
        state.cursor = Some(outcome.last_commit_cursor);
        state.next_seq = state.next_seq.saturating_add(1);
        self.status.record_segment(bytes, now);
        Ok((1, bytes))
    }

    /// Whether the current generation has outlived the snapshot interval, so the
    /// next checkpoint should start a new one with a fresh base snapshot rather
    /// than another index.
    fn generation_expired(&self, now: DateTime<Utc>) -> bool {
        self.state.as_ref().is_some_and(|state| {
            let elapsed = now.signed_duration_since(state.started_at);
            // A clock that stepped backwards leaves a negative span. Treat that
            // as "expired" rather than "forever young", so a bad clock cannot
            // pin one generation open indefinitely.
            elapsed
                .to_std()
                .map_or(true, |age| age >= self.settings.snapshot_interval)
        })
    }

    /// Drop generations that are entirely outside the retention window.
    ///
    /// The newest generation that opened at or before the cutoff is **kept**:
    /// without it the start of the window is not reconstructable. Returns how
    /// many generations were removed.
    fn prune(&self, now: DateTime<Utc>) -> Result<u64, ReplicationError> {
        let Ok(retention) = chrono::Duration::from_std(self.settings.retention) else {
            return Ok(0);
        };
        let cutoff_ms = now
            .checked_sub_signed(retention)
            .unwrap_or(now)
            .timestamp_millis();
        let keys = self
            .destination
            .list(&segment::generations_prefix(&self.settings.root))?;

        let open_generation = self.state.as_ref().map(|state| state.id.clone());
        let droppable = prunable(
            &self.settings.root,
            keys,
            cutoff_ms,
            open_generation.as_deref(),
        );
        let mut pruned: u64 = 0;
        for (id, keys) in &droppable {
            // Delete the commit marker FIRST: a prune interrupted halfway then
            // leaves an *uncommitted* generation, which restore ignores, rather
            // than a complete-looking generation with no segments.
            let marker = segment::snapshot_meta_key(&self.settings.root, id);
            if keys.contains(&marker) {
                self.destination.delete(&marker)?;
            }
            for key in keys {
                if *key == marker {
                    continue;
                }
                self.destination.delete(key)?;
            }
            pruned = pruned.saturating_add(1);
            tracing::debug!(generation = %id, "pruned an expired SQLite replication generation");
        }
        if pruned > MIN_KEPT_GENERATIONS as u64 {
            tracing::warn!(
                pruned,
                "the SQLite replica retention sweep removed an unusual number of generations \
                 at once — check the host clock if this was not expected"
            );
        }
        Ok(pruned)
    }
}

/// The generations a retention sweep at `cutoff_ms` removes, with their keys.
///
/// Ordered as a restore orders them: epoch, then time (issue #3233). The floor
/// is the generation a restore to the cutoff picks. A generation is removed
/// when it orders before the floor, is not one of the
/// [`MIN_KEPT_GENERATIONS`] newest complete ones, and is not `open`. A
/// generation with no `snapshot.json` that is newer in time than the floor
/// is never removed: it can be another replicator's upload in progress, whose
/// epoch marker is not there yet.
fn prunable(
    root: &str,
    keys: Vec<String>,
    cutoff_ms: i64,
    open: Option<&str>,
) -> Vec<(String, Vec<String>)> {
    let epochs = restore::generation_epochs(root, &keys);
    let mut generations: BTreeMap<(u64, i64, String), Vec<String>> = BTreeMap::new();
    let mut complete: Vec<(u64, i64, String)> = Vec::new();
    for key in keys {
        let Some(id) = segment::generation_of_key(root, &key) else {
            continue;
        };
        let Some(info) = segment::parse_generation_id(&id) else {
            continue;
        };
        let epoch = epochs.get(&id).copied().unwrap_or(0);
        if key.ends_with(segment::SNAPSHOT_META_OBJECT) {
            complete.push((epoch, info.created_ms, id.clone()));
        }
        generations
            .entry((epoch, info.created_ms, id))
            .or_default()
            .push(key);
    }
    complete.sort_unstable();
    let Some(floor) = complete
        .iter()
        .rev()
        .find(|(_, ms, _)| *ms <= cutoff_ms)
        .cloned()
    else {
        return Vec::new();
    };
    // Keep the newest few complete generations unconditionally. A clock that
    // steps forward past the retention window would otherwise expire the
    // whole history in one sweep and leave nothing to restore from.
    let keep_newest: std::collections::BTreeSet<&(u64, i64, String)> =
        complete.iter().rev().take(MIN_KEPT_GENERATIONS).collect();
    let is_complete: std::collections::BTreeSet<&str> =
        complete.iter().map(|(_, _, id)| id.as_str()).collect();
    generations
        .into_iter()
        .filter(|(order, _)| {
            let (_, ms, id) = order;
            *order < floor
                && !keep_newest.contains(order)
                && open != Some(id.as_str())
                && (is_complete.contains(id.as_str()) || *ms <= floor.1)
        })
        .map(|((_, _, id), keys)| (id, keys))
        .collect()
}

/// Clears the "verification in flight" flag when dropped, panic or not.
struct InFlightGuard(Arc<AtomicBool>);

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

/// Prove a replica restorable by restoring it into a scratch directory beside
/// `database_path` — the same filesystem, so the restore's rename is cheap — and
/// integrity-checking the result. The scratch directory is removed either way.
fn verify_replica(
    destination: &dyn ReplicaDestination,
    root: &str,
    database_path: &Path,
) -> Result<RestoreOutcome, RestoreError> {
    let parent = database_path
        .parent()
        .filter(|p| !p.as_os_str().is_empty())
        .unwrap_or_else(|| Path::new("."));
    let scratch = parent.join(format!(
        ".autumn-replica-verify-{}-{:016x}",
        std::process::id(),
        rand::random::<u64>()
    ));
    // Created here, owner-only, rather than left to `restore`'s `create_dir_all`:
    // that lands 0755 under the usual umask, and what goes inside is a complete
    // copy of production. On a host where the database file is protected but its
    // parent directory is traversable, every verification would hand a readable
    // snapshot to any other local user — and a crash would leave it there. The
    // CLI verifier already creates its scratch directory this way.
    if let Err(e) = create_private_dir(&scratch) {
        return Err(RestoreError::Io {
            op: "create the verification scratch directory",
            detail: e.to_string(),
        });
    }
    let outcome = restore::restore(destination, root, None, &scratch.join("verified.db"));
    let _ = std::fs::remove_dir_all(&scratch);
    outcome
}

/// Create a directory only its owner can enter, on platforms that have modes.
fn create_private_dir(path: &Path) -> std::io::Result<()> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::DirBuilderExt as _;
        std::fs::DirBuilder::new()
            .recursive(true)
            .mode(0o700)
            .create(path)
    }
    #[cfg(not(unix))]
    {
        std::fs::create_dir_all(path)
    }
}

/// Read `PRAGMA journal_mode`.
fn journal_mode(conn: &mut SqliteConnection) -> Result<String, ReplicationError> {
    let rows: Vec<JournalModeRow> = diesel::sql_query("PRAGMA journal_mode")
        .load(conn)
        .map_err(|e| {
            ReplicationError::Sqlite(SqliteError::Query {
                op: "journal_mode",
                detail: e.to_string(),
            })
        })?;
    Ok(rows
        .first()
        .map_or_else(|| "unknown".to_owned(), |row| row.journal_mode.clone()))
}

/// Read and validate the `-wal` header, or `None` when the WAL is too short to
/// hold one (an empty WAL, right after a checkpoint).
fn read_wal_header(wal_path: &Path, wal_len: u64) -> Result<Option<WalHeader>, ReplicationError> {
    if wal_len < wal::WAL_HEADER_SIZE as u64 {
        return Ok(None);
    }
    let mut file =
        std::fs::File::open(wal_path).map_err(ReplicationError::io("open the -wal file"))?;
    let mut buf = [0u8; wal::WAL_HEADER_SIZE];
    file.read_exact(&mut buf)
        .map_err(ReplicationError::io("read the -wal header"))?;
    match WalHeader::parse(&buf) {
        Ok(header) => Ok(Some(header)),
        // A WAL being rewritten under us is transient, not fatal: skip this tick
        // rather than failing the loop.
        Err(WalError::HeaderChecksum | WalError::TooShort { .. }) => Ok(None),
        Err(e) => Err(ReplicationError::Wal(e)),
    }
}

/// Read `[start, end)` of a file into memory.
fn read_range(path: &Path, start: u64, end: u64) -> Result<Vec<u8>, ReplicationError> {
    let len = end.saturating_sub(start);
    let mut file = std::fs::File::open(path).map_err(ReplicationError::io("open the -wal file"))?;
    file.seek(SeekFrom::Start(start))
        .map_err(ReplicationError::io("seek the -wal file"))?;
    let mut buffer = Vec::new();
    file.take(len)
        .read_to_end(&mut buffer)
        .map_err(ReplicationError::io("read the -wal file"))?;
    Ok(buffer)
}

/// The staging file a base snapshot is gzip'd into, next to the database so the
/// rename-free upload stays on the same filesystem.
fn staging_path(db: &Path) -> PathBuf {
    let mut name = db.as_os_str().to_os_string();
    name.push(format!(".autumn-replica-{}.gz", std::process::id()));
    PathBuf::from(name)
}

/// Gzip `source` into `target`, returning the **uncompressed** digest and length.
///
/// Streams in fixed-size chunks so a multi-GB database is never buffered.
fn gzip_file(source: &Path, target: &Path) -> Result<(String, u64), ReplicationError> {
    use sha2::{Digest as _, Sha256};

    let mut input =
        std::fs::File::open(source).map_err(ReplicationError::io("open the database file"))?;
    let output =
        std::fs::File::create(target).map_err(ReplicationError::io("create the snapshot"))?;
    let mut encoder = flate2::write::GzEncoder::new(output, flate2::Compression::fast());
    let mut hasher = Sha256::new();
    let mut buffer = vec![0u8; 256 * 1024];
    let mut total: u64 = 0;
    loop {
        let read = input
            .read(&mut buffer)
            .map_err(ReplicationError::io("read the database file"))?;
        if read == 0 {
            break;
        }
        let chunk = buffer.get(..read).unwrap_or(&[]);
        hasher.update(chunk);
        encoder
            .write_all(chunk)
            .map_err(ReplicationError::io("compress the snapshot"))?;
        total = total.saturating_add(read as u64);
    }
    let file = encoder
        .finish()
        .map_err(ReplicationError::io("finish the snapshot"))?;
    file.sync_all()
        .map_err(ReplicationError::io("flush the snapshot"))?;
    Ok((hex::encode(hasher.finalize()), total))
}

/// Sleep `total`, waking early if `shutdown` is cancelled. Returns `true` when
/// cancelled.
#[allow(
    clippy::disallowed_methods,
    reason = "the deadline paces a real thread sleep, and a virtual clock does not move during it"
)]
fn sleep_or_cancelled(shutdown: &tokio_util::sync::CancellationToken, total: Duration) -> bool {
    let start = std::time::Instant::now();
    let deadline = start.checked_add(total).unwrap_or(start);
    loop {
        if shutdown.is_cancelled() {
            return true;
        }
        let now = std::time::Instant::now();
        if now >= deadline {
            return false;
        }
        std::thread::sleep(SHUTDOWN_POLL.min(deadline.saturating_duration_since(now)));
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn keys(id: &str, epoch: Option<u64>, complete: bool) -> Vec<String> {
        let mut keys = vec![segment::snapshot_key("r", id)];
        if let Some(epoch) = epoch {
            keys.push(segment::generation_epoch_key("r", id, epoch));
        }
        if complete {
            keys.push(segment::snapshot_meta_key("r", id));
        }
        keys
    }

    fn pruned(all: &[Vec<String>], cutoff_ms: i64, open: Option<&str>) -> Vec<String> {
        let mut ids: Vec<String> = prunable("r", all.concat(), cutoff_ms, open)
            .into_iter()
            .map(|(id, _)| id)
            .collect();
        ids.sort();
        ids
    }

    #[test]
    fn retention_orders_by_epoch_then_time() {
        let id = segment::generation_id;
        let (old, stale, owner) = (id(1_000, 1), id(5_000, 2), id(2_000, 3));
        let (newest, kept) = (id(6_000, 4), id(7_000, 5));
        let all = vec![
            keys(&old, Some(1), true),
            keys(&stale, Some(1), true),
            keys(&owner, Some(2), true),
            keys(&newest, Some(2), true),
            keys(&kept, Some(2), true),
        ];
        // A restore to 5_500 picks `owner` (epoch 2): the stale epoch-1
        // generation, newer in time, orders below it and goes.
        assert_eq!(pruned(&all, 5_500, None), vec![old, stale]);
    }

    #[test]
    fn retention_keeps_an_upload_in_progress_newer_than_the_floor() {
        let id = segment::generation_id;
        let (floor, uploading) = (id(1_000, 1), id(9_000, 2));
        let (later, latest) = (id(10_000, 3), id(11_000, 4));
        let all = vec![
            keys(&floor, Some(1), true),
            // Another replicator's snapshot, before its markers.
            keys(&uploading, None, false),
            keys(&later, Some(1), true),
            keys(&latest, Some(1), true),
        ];
        assert!(pruned(&all, 5_000, None).is_empty());
        // Without epochs the old rules hold: older than the floor goes.
        let (old, mid) = (id(1_000, 1), id(2_000, 2));
        let (recent, newest) = (id(9_000, 3), id(9_500, 4));
        let all = vec![
            keys(&old, None, true),
            keys(&mid, None, true),
            keys(&recent, None, true),
            keys(&newest, None, true),
        ];
        assert_eq!(pruned(&all, 5_000, None), vec![old.clone()]);
        assert!(
            pruned(&all, 5_000, Some(&old)).is_empty(),
            "the open generation stays"
        );
    }

    #[test]
    fn staging_path_sits_next_to_the_database() {
        let path = staging_path(Path::new("/srv/app.db"));
        assert_eq!(path.parent(), Some(Path::new("/srv")));
        assert!(
            path.to_string_lossy()
                .starts_with("/srv/app.db.autumn-replica-"),
            "{}",
            path.display()
        );
    }

    #[test]
    fn gzip_file_reports_the_uncompressed_digest_and_length() {
        let dir = tempfile::tempdir().expect("tempdir");
        let source = dir.path().join("app.db");
        let payload = b"autumn".repeat(1024);
        std::fs::write(&source, &payload).expect("write");

        let target = dir.path().join("app.db.gz");
        let (digest, len) = gzip_file(&source, &target).expect("gzip");
        assert_eq!(len, payload.len() as u64);
        assert_eq!(digest, segment::sha256_hex(&payload));

        let mut inflated = Vec::new();
        let file = std::fs::File::open(&target).expect("open gz");
        std::io::copy(&mut flate2::read::GzDecoder::new(file), &mut inflated).expect("inflate");
        assert_eq!(inflated, payload);
    }

    #[test]
    fn read_range_returns_exactly_the_requested_window() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("wal");
        std::fs::write(&path, (0u8..=255).collect::<Vec<_>>()).expect("write");
        assert_eq!(
            read_range(&path, 10, 20).expect("read"),
            (10u8..20).collect::<Vec<_>>()
        );
        // A window past EOF returns what exists rather than failing.
        assert_eq!(read_range(&path, 250, 300).expect("read").len(), 6);
    }

    #[test]
    fn sleep_or_cancelled_returns_immediately_when_already_cancelled() {
        let token = tokio_util::sync::CancellationToken::new();
        token.cancel();
        let start = std::time::Instant::now();
        assert!(sleep_or_cancelled(&token, Duration::from_secs(30)));
        assert!(start.elapsed() < Duration::from_secs(1));
    }

    #[test]
    fn read_wal_header_treats_a_short_wal_as_absent() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("app.db-wal");
        std::fs::write(&path, [0u8; 8]).expect("write");
        assert!(read_wal_header(&path, 8).expect("read").is_none());
    }
}