turso_backup/

stream.rs

//! Tier 2 — WAL-frame streaming sink (R005-F2).
//!
//! Tee WAL frames from a live turso connection to an object store, anchored
//! to a tier-1a base snapshot. Restore (R005-F3) replays the frames onto the
//! snapshot. Near-zero RPO; engine-coupled but the public seam is small —
//! see the spike findings in `.yah/docs/working/turso-s3-backup.md`.
//!
//! ## Object layout under `BackupTarget::prefix`
//!
//! ```text
//! frames/{epoch:020}/{checkpoint_seq:010}/{first:020}-{last:020}  batch of consecutive WAL frames
//! frames/{checkpoint_seq:010}/{first:020}-{last:020}              ditto, epoch 0 (unfenced)
//! frames/{epoch:020}/{checkpoint_seq:010}/{frame_no:020}          one raw frame, pre-R761-F2 layout
//! frames/{checkpoint_seq:010}/{frame_no:020}                      ditto, epoch 0 (pre-fencing)
//! generations/gen-{unix_nanos:020}.manifest                one per `tail_frames` call that uploaded
//! latest.stream-watermark    text sidecar: "<checkpoint_seq> <last_frame> <written_at_nanos> <epoch> <pointer_generation> <salt1> <salt2>"
//! ```
//!
//! R858-B19: the trailing salt pair is the WAL generation the `last_frame`
//! position belongs to. `checkpoint_seq` alone does not identify a generation —
//! a writer-process restart recreates the WAL back at sequence 0 — so the salt
//! is what `tail_frames` compares and what a generation manifest stamps. See
//! [`WalGeneration`].
//!
//! A frame object holds `n` consecutive frames, each `24 + page_size` bytes,
//! concatenated in ascending frame order — so a batch is exactly the bytes the
//! old per-frame objects held, glued together, and offset `i * frame_size`
//! within it is frame `first + i`.
//!
//! The generation manifest names the base snapshot key, the page size, the
//! frame range covered, which batch objects cover it (since R761-F2), and
//! (since R732-F2) the fencing epoch and owner label. Object keys are
//! zero-padded so lexical order matches chronological order (same convention as
//! tier 1a snapshots / tier 1b manifests; clock-skew-immune).
//!
//! ## Frame batching (R761-F2, W248/W313 §9)
//!
//! One object per WAL frame made per-write Class A ops scale with *frames*,
//! and a frame is one page — at 4 KB pages, every 4 KB of changed data was a
//! billed op, which is a pathologically small object. A tail call's frames now
//! go up as one ranged object per drain batch (bounded by
//! [`BackpressureConfig::spill_buffer_frames`], the same bound that already
//! caps how much this sink holds in memory), so an uploading call costs
//! `ceil(frames / spill_buffer_frames) + 2` PUTs instead of `frames + 2` —
//! typically **3, flat**, whatever the write volume in the interval.
//!
//! Reading stays compatible in the direction that matters: a manifest without
//! a `frame_batch` list is a pre-R761-F2 generation and its frames are fetched
//! one object each, so backups written before this change (and chains that
//! straddle it) still restore. The reverse is a loud refusal by construction —
//! the manifest header moved to `v3`, so an older binary reading a batched
//! generation says "unexpected manifest header" rather than mis-reading it.
//!
//! ## Fencing (R732-F2 / R732-T3, W245; R736-T2, W250)
//!
//! [`StreamConfig::epoch`] is a per-tenant fencing token minted by yubaba's
//! raft state machine. It is checked against the sidecar before any frame is
//! uploaded, and the sidecar advance is a compare-and-swap on the version that
//! check read — so a stale owner is *rejected* (with
//! [`StreamOutcome::Fenced`]) both when it arrives late and when it races. An
//! epoch of `0` means unfenced, which preserves single-writer behaviour but is
//! still fenced *by* a claimed sink.
//!
//! That epoch is a **local** raft counter — it fences ownership moves within
//! one cell, but two cells are independent raft groups that share no epoch
//! counter, so it is blind to a tenant moving to a *different* cell.
//! [`StreamConfig::pointer_generation`] is the second, cross-cell fence: the
//! writer's belief about the global tenant→cell pointer's generation
//! (`yah_tenant_pointer::PointerRecord`). It is checked alongside `epoch` at
//! the same two points — up front against the sidecar, and again on a lost
//! watermark CAS — so a stale generation bounces exactly like a stale epoch.
//! A node is the real owner of a tenant only when **both** fences pass.
//!
//! ## Seam isolation
//!
//! Every call into `turso_core::Connection`'s `feature = "conn_raw_api"`
//! surface goes through the [`WalSeam`] trait. If a future turso release
//! renames or reshapes those calls, the delta is a single impl block — not a
//! sed across the whole sink. The spike measured one breaking rename in three
//! months (`wal_auto_checkpoint_disable` → `wal_auto_actions_disable`); the
//! trait makes that a one-file fix.
//!
//! @yah:relay(R005, "Tier 2 — WAL-frame streaming (deferred)")
//! @yah:at(2026-05-26T22:28:31Z)
//! @yah:status(open)
//! @yah:phase(P3)
//! @yah:parent(Q002)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//!
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//!
//! @yah:ticket(R005-F2, "Frame-streaming sink: base snapshot + incremental frames + generation tracking")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-05-26T22:30:09Z)
//! @yah:status(review)
//! @yah:phase(P3)
//! @yah:parent(R005)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//! @yah:handoff("Implemented in src/stream.rs (~590 LOC). Public API: WalSeam trait (3 methods: wal_state, wal_get_frame, wal_auto_actions_disable) + real CoreWalSeam impl over turso_core::Connection + StreamConfig{base_snapshot_key, page_size} + tail_frames(seam, target, cfg) -> StreamOutcome (Empty | Streamed | Restarted) + GenerationManifest format/parse + Watermark sidecar.")
//! @yah:handoff("Object layout under prefix: frames/{checkpoint_seq:010}/{frame_no:020} for raw frames (24-byte header + page), generations/gen-{nanos:020}.manifest for per-call manifests, latest.stream-watermark for the (checkpoint_seq, last_frame) sidecar. Same zero-padded-key convention as tier 1a snapshots and tier 1b manifests — clock-skew-immune.")
//! @yah:handoff("Tier-2 invariants from the R005-T1 spike are baked in: (checkpoint_seq, frame_no) compound key (not raw frame_no), so a WAL restart -> Restarted outcome under a new seq; page_size is a required cfg parameter (no 4096 hardcode like sync_server.rs); WalSeam isolates all turso_core::Connection calls (one-file delta if upstream renames again); CoreWalSeam::open() takes WAL ownership via wal_auto_actions_disable() at construction.")
//! @yah:handoff("Cargo.toml: added turso_core = '0.6.1' with features = ['conn_raw_api'] as sibling dep to turso='0.6.1'. The friendly turso wrapper does NOT re-export the raw WAL API; pinned in lockstep — if either bumps, bump both.")
//! @yah:handoff("Verified: 6 new stream unit tests (empty-on-empty-wal, initial-tail-records-watermark, second-tail-no-new-frames-is-Empty, second-tail-uploads-only-new, wal-restart-emits-Restarted-under-new-seq, manifest-roundtrip+rejection) using a mockable WalSeam. Full crate: 18/18 tests green. cargo clippy --all-targets -- --deny=warnings clean.")
//! @yah:handoff("What's NOT verified at F2 level: live-DB ping-pong of CoreWalSeam (seed rows -> wal_state -> wal_get_frame -> assert is_commit_frame on the last frame). F3's restore path will be the natural end-to-end exercise. Optional sanity test could be added under F2 if you'd rather catch a CoreWalSeam regression here vs in F3.")
//! @yah:handoff("Generation manifest text format: 'TURSO-BACKUP STREAM v1' header, then `base_snapshot <key>`, `page_size <n>`, `checkpoint_seq <n>`, `first_frame <n>`, `last_frame <n>`. Dependency-free, same convention as dedup::Manifest. parse_generation_manifest fails loudly on any other shape.")
//! @yah:next("User: review/approve F2. If you want a live-DB CoreWalSeam sanity test before signoff, say so and I'll add it under F2; otherwise F3 picks it up naturally.")
//! @yah:next("On approval: archive F2, claim R005-F3 (Restore via frame replay onto snapshot). F3 fetches latest gen-manifest, downloads referenced base_snapshot + frames in (checkpoint_seq, frame_no) order, replays into a writable DB via wal_insert_begin/wal_insert_frame/wal_insert_end (the same WalSeam trait, extended with the insert side).")
//! @yah:next("Optional independent of F3: file a tiny upstream PR to re-export conn_raw_api from the `turso` wrapper crate so the sibling turso_core dep collapses to one.")
//!
//! @yah:ticket(R005-F3, "Restore via frame replay onto snapshot + restart/crash-consistency handling")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-05-26T22:30:10Z)
//! @yah:status(review)
//! @yah:phase(P3)
//! @yah:parent(R005)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//! @yah:depends_on(R005-F2)
//! @yah:handoff("Built restore_latest_stream(&BackupTarget, dest_path) + the WalInsertSeam trait extension on the existing WalSeam pattern. Public surface: WalInsertSeam{wal_insert_begin, wal_insert_frame, wal_insert_end} + impl for CoreWalSeam (wraps turso_core::Connection::wal_insert_*); RestoreOutcome{base_snapshot_key, checkpoint_seq, generation_count, frames_replayed, last_frame}; restore_latest_stream(target, dest_path)->RestoreOutcome.")
//! @yah:handoff("Flow: list+sort all generations/*.manifest keys lexicographically → parse each → validate_generation_chain checks all share one base, one page_size, one checkpoint_seq, frames start at 1 and are contiguous → download base_snapshot to dest_path → CoreWalSeam::open(dest) → wal_insert_begin → replay_frames_into walks (seq, frame_no) in order, downloads frames/{seq:010}/{frame_no:020}, asserts byte length = 24+page_size, calls wal_insert_frame → wal_insert_end(force_commit=false). Crash-consistency story = the engine's own truncate-to-last-commit-frame on insert_end(false).")
//! @yah:handoff("Restart handling = REFUSE: a chain spanning two checkpoint_seqs means the source engine folded WAL into main between generations, so the post-restart frames don't replay onto our pre-restart base. validate_generation_chain bails with 'WAL restart between generations, restore needs a fresh tier-1a snapshot'. Same refusal for cross-base chains and frame gaps. This is the right semantics — heroic restart-spanning replay would silently corrupt.")
//! @yah:handoff("Verified with 13 new stream tests on top of F2's 6: validate_chain accepts single/contiguous-multi, rejects empty/gap/non-one-start/restart/base-mismatch/page_size-mismatch (7 tests); replay_walks_manifests_in_frame_order with MockInsertSeam; restore_errors_when_no_generations; replay_rejects_wrong_size_frame; live_db_seed_snapshot_tail_restore_round_trips (real turso + CoreWalSeam: seed 50 → snapshot → checkpoint → seed 25 more → tail → restore → assert dest has 75 rows); manifest_with_uncommitted_tail_rolls_back_via_insert_end (live: stage a phantom non-commit frame in the sink, extend the manifest, assert restore drops it and dest=3 rows = 2 base + 1 committed, NOT 4).")
//! @yah:handoff("cargo test -p turso-backup = 31/31 green (up from 18); cargo clippy --all-targets -- --deny=warnings = clean. One clippy lint fixed in the new test code: manual_is_multiple_of (Rust 1.95 lint). The live tests use turso::Builder for writes/reads + CoreWalSeam for tailing — exclusive WAL lock means we drop the high-level conn before opening the low-level seam (matched by the snapshot tests' pattern).")
//! @yah:handoff("Two known design decisions worth flagging to the reviewer: (1) replay starts at frame 1 in the dest's fresh WAL but the source's frames could overlap content already in the base snapshot (since VACUUM INTO point-in-time includes WAL state). turso's wal_insert_frame compares-and-returns-OK on identical content, so redundant frames are no-ops — clean orchestration via checkpoint-then-snapshot-then-stream avoids them entirely (which is what the live test does). (2) on replay error we attempt a best-effort wal_insert_end(false) before bubbling the error up, so we don't leave the dest's WAL with an uncommitted suffix half-open.")
//! @yah:next("User: review/approve F3. Once green, archive F3 and the F2/F3-dependent state of R005 collapses to F4 only (concurrent-writer-safe raw copy).")
//! @yah:verify("cargo test -p turso-backup")
//! @yah:verify("cargo clippy --all-targets -- --deny=warnings")
//!
//! @yah:ticket(R005-F4, "Concurrent-writer-safe raw copy: read-only main + WAL-frame replay (no TRUNCATE-checkpoint dependency), for backing up under a live writer")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-05-27T03:05:00Z)
//! @yah:status(review)
//! @yah:phase(P3)
//! @yah:parent(R005)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//! @yah:gotcha("R004's dedup::raw_consistent_copy assumes no live writer: it folds WAL->main via PRAGMA wal_checkpoint(TRUNCATE), which a concurrent writer can make return 'busy'. The live-writer alternative copies the main file under a read txn and replays WAL frames itself (wal_get_frame seam) — tier-2 territory, gated on R005-T1's seam assessment. Filed as an R004-T4 followup.")
//! @yah:handoff("Implemented raw_consistent_copy_live(db_path, page_size) + pub(crate) replay_wal_onto_main inner. Algorithm: open CoreWalSeam (auto-actions disabled on our conn) -> wal_state for (cp_seq, max_frame) -> std::fs::read main -> walk frames 1..=max_frame collecting (page_no, db_size, page_bytes), locate the last is_commit_frame -> grow image to fit the largest page slot in the commit prefix -> overwrite each page at (page_no-1)*page_size -> truncate to db_size*page_size. Frames past the last commit are dropped wholesale (crash-consistency, matches restore's wal_insert_end(false)).")
//! @yah:next("User: review/approve F4. Once green, archive F4 and R005 collapses to closed (R005-F2/F3/F4 all in review).")
//! @yah:verify("cargo test -p turso-backup: 38/38 green (up from 31). New: 5 unit tests on replay_wal_onto_main with MockWal (empty-WAL/single-commit/uncommitted-tail-dropped/multi-commit-prefix-grows-image/only-uncommitted-frames-returns-main) + 2 live tests on raw_consistent_copy_live (live_db_consistent_copy_without_truncate_round_trips: seed -> checkpoint_truncate -> seed more uncheckpointed -> copy-live -> reopen = 35 rows; live_db_consistent_copy_reflects_new_writes: repeatable copy after subsequent writes).")
//! @yah:verify("cargo clippy --all-targets -- --deny=warnings: clean.")
//!
//! ## R574-F2 — explicit R2 backpressure
//!
//! `tail_frames`'s upload loop drains frames through
//! [`crate::backpressure::put_with_backoff`] against a bounded spill buffer
//! (see [`drain_frames_with_backpressure`]) instead of putting each frame
//! directly and bubbling any error raw. Full design in the `backpressure`
//! module doc; ticket tracked in the W248 relay, not in-source.
//! @arch:see(.yah/docs/working/W248-wal-streamer-hardening.md)
//!
//! ## R574-T4 — explicit RPO knob
//!
//! Cadence was entirely implicit in the caller's `tail_frames` invocation
//! interval (doc §10). `StreamConfig::rpo_target` states that interval as a
//! number; every `tail_frames` call reports [`RpoStatus`] (age of the last
//! durably-persisted watermark, and whether that age has drifted past the
//! target) on its [`StreamOutcome`] so an orchestrator can alert without
//! reimplementing the bookkeeping. This crate still does not schedule
//! anything itself — cadence/retention policy stays caller-driven per the
//! doc's "Not in scope" — `rpo_target` documents the contract the caller's
//! own scheduler is expected to uphold, and `RpoStatus` is the receipt.
//! @arch:see(.yah/docs/working/W248-wal-streamer-hardening.md)
//!
//! ## R761-T1 — a measured default tail cadence
//!
//! [`DEFAULT_TAIL_INTERVAL`] (60 s) and [`DEFAULT_RPO_TARGET`] (120 s) are
//! the cadence this crate recommends, derived from
//! `examples/tail_sweep_harness.rs`'s 2026-08-13 sweep rather than chosen:
//! every uploading `tail_frames` call writes two fixed objects (generation
//! manifest + watermark CAS) on top of the frames, so per-write cost was
//! `frames/write + 2/writes_per_tail` — 4.03 PUTs/write at one write per
//! tail, 2.05 at a hundred. The constants' docs carry the full table and
//! the reasoning for landing at 60 s instead of chasing the last 3 %. Still
//! no scheduler in this crate; these are numbers for the caller's.
//!
//! R761-F2 then removed the `frames/write` term those numbers were floored
//! by (see *Frame batching* above), so the cost is now `3/writes_per_tail`
//! for a call whose frames fit one batch — the cadence lever and the layout
//! lever compose, and the second is worth more the longer the interval.
//! @arch:see(.yah/docs/working/W248-wal-streamer-hardening.md)
//!
//! ## R574-F3 — one-puller-per-box fan-out + warm applier
//!
//! [`crate::puller::WalPuller`] is the read side's counterpart to
//! `tail_frames`'s write side: it pulls each newly-uploaded frame from R2
//! exactly once per box and fans it out in-process to every attached warm
//! applier (a [`WalInsertSeam`] with its page cache trimmed hard via
//! [`CoreWalSeam::trim_page_cache_kb`]), so R2 read ops are O(boxes), not
//! O(replicas). See the `puller` module doc for the full design and its v1
//! scope cut (attach is cold-start-only; no mid-stream backlog replay).
//! @arch:see(.yah/docs/working/W248-wal-streamer-hardening.md)

use anyhow::{Context, Result};
use object_store::path::Path as ObjPath;
use object_store::{ObjectStore, ObjectStoreExt, PutMode, PutOptions, UpdateVersion};
use std::collections::{HashSet, VecDeque};
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use crate::backpressure::{put_with_backoff, BackpressureConfig, BackpressurePolicy, BackpressureReport};
use crate::snapshot::BackupTarget;

/// WAL frame layout: 24-byte frame header (page_no big-endian at 0..4,
/// db_size big-endian at 4..8, salts/checksums in the remainder) followed by
/// `page_size` bytes of page data. Constant per the SQLite WAL format; the
/// page size is read from the base snapshot's header (page 1, byte 16, `u16`
/// with the value `1` meaning 65536). `sync_server.rs` hardcodes 4096 — we
/// don't, see [`StreamConfig::page_size`].
pub const WAL_FRAME_HEADER_SIZE: usize = 24;

/// Position in the WAL: a `(checkpoint_seq, last_frame)` pair. `max_frame`
/// resets to 0 every time the WAL header restarts (`WalAutoActions::Restart`
/// fires), and `checkpoint_seq_no` increments alongside it — so the pair is
/// the right primary key for sink objects, not raw frame_no.
///
/// **R858-B19 — `checkpoint_seq` alone does NOT identify a WAL generation, and
/// this doc used to claim it did ("increments monotonically across restarts").
/// That claim is false and it cost a silent wrong restore.** It holds only for
/// an *in-process* restart, where the same WAL file is reused. When the last
/// connection to a SQLite database closes, the engine checkpoints and DELETES
/// the `-wal` file; the next writer creates a fresh WAL back at
/// checkpoint-sequence `0` with a brand-new random salt. Across that fold
/// `checkpoint_seq` goes `0 -> 0` over two completely unrelated WALs (measured
/// 2026-09-06 against system sqlite3 3.51.0 by
/// `examples/foreign_checkpoint_probe.rs`, probes B and F).
///
/// Generation identity therefore lives in [`WalGeneration`], which pairs the
/// sequence with the WAL header's salt. Never compare two `Watermark`s'
/// `checkpoint_seq` to decide "same WAL" — that is exactly the inference this
/// ticket exists to delete. `last_frame` is only meaningful *relative to a
/// known generation*.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct Watermark {
    pub checkpoint_seq: u32,
    pub last_frame: u64,
}

/// R858-B19 — the WAL header's `(salt1, salt2)`, the two 32-bit values SQLite
/// re-rolls on every WAL reset. Stored big-endian at bytes 16..24 of the
/// 32-byte WAL header, and copied verbatim into bytes 8..16 of **every frame
/// header** written under that header — which is where this crate reads it
/// from, since `turso_core::WalState` exposes only `checkpoint_seq_no` and
/// `max_frame`.
///
/// Why the salt and not the sequence: the salt moves in *both* fold regimes,
/// and the sequence moves in only one.
///
/// - WAL recreated by a writer restart: fresh randomness (measured
///   `b83c03f5 -> 9ca89e39`, unrelated), while `checkpoint_seq` resets `0 -> 0`.
/// - In-process autocheckpoint: `salt1` increments in lockstep with the
///   sequence (measured `d492ea8a -> ... -> d492ea92` alongside seq `0 -> 8`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WalSalt {
    pub salt1: u32,
    pub salt2: u32,
}

impl WalSalt {
    /// Read the salt out of a WAL **frame** header (bytes 8..16 of the 24-byte
    /// header, big-endian). Panics on a short slice — every caller here sizes
    /// its buffer at `WAL_FRAME_HEADER_SIZE + page_size`.
    pub(crate) fn from_frame_header(frame: &[u8]) -> Self {
        let be = |o: usize| u32::from_be_bytes(frame[o..o + 4].try_into().unwrap());
        Self { salt1: be(8), salt2: be(12) }
    }

    /// R858-B18 — read the salt out of the **32-byte `-wal` file header**
    /// (bytes 16..24, big-endian), the copy SQLite writes once per WAL
    /// generation and duplicates into every frame header.
    ///
    /// The two constructors differ only in offset, and they live together
    /// deliberately: one type, one place, so the two ways this crate can reach
    /// the same value can never disagree. [`from_frame_header`](Self::from_frame_header)
    /// is the seam-only path ([`read_wal_salt`] — works through a
    /// [`CoreWalSeam::from_conn`] that has no path); this one is the
    /// path-only path ([`SourceFingerprint`] — works with no engine open at
    /// all, which is what makes it usable as an independent check *on* a copy
    /// the engine took).
    ///
    /// Panics on a slice shorter than 24 bytes; [`WalFileHeader::parse`] is the
    /// length-checked entry point every caller here actually uses.
    pub(crate) fn from_wal_file_header(hdr: &[u8]) -> Self {
        let be = |o: usize| u32::from_be_bytes(hdr[o..o + 4].try_into().unwrap());
        Self { salt1: be(16), salt2: be(20) }
    }
}

impl std::fmt::Display for WalSalt {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{:08x}/{:08x}", self.salt1, self.salt2)
    }
}

/// R858-B19 — the identity of one WAL generation: the checkpoint sequence
/// **and** the header salt that actually distinguishes it. This is what
/// [`tail_frames`] compares across calls, what the watermark sidecar persists,
/// and what a generation manifest stamps.
///
/// `salt: None` means **unknown generation**, and unknown is a value here, not
/// a missing one: it arises from a sidecar or manifest written before this
/// field existed, or from a WAL with no frames to read a salt out of. An
/// unknown generation is never provably equal to anything — see
/// [`WalGeneration::is_provably_same_as`] — so it forces a restart on the write
/// side and a refusal on the restore side rather than a guess.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct WalGeneration {
    pub checkpoint_seq: u32,
    pub salt: Option<WalSalt>,
}

impl WalGeneration {
    /// True only when both sides carry a **known** salt and every component
    /// agrees — i.e. only when these are *provably* the same WAL.
    ///
    /// The asymmetry is the whole point. "Not provably the same" is treated as
    /// "different", which costs a redundant re-upload (or a loud restore
    /// refusal) in the worst case. The opposite default — "no evidence of a
    /// change, so assume it is the same WAL" — is what spliced two WAL
    /// generations into one chain and restored a plausible wrong image.
    pub fn is_provably_same_as(&self, other: &WalGeneration) -> bool {
        match (self.salt, other.salt) {
            (Some(a), Some(b)) => a == b && self.checkpoint_seq == other.checkpoint_seq,
            _ => false,
        }
    }

    /// Render for an error message: `seq 4 salt 1109ca5e/7acf42a3`, or
    /// `seq 4 salt <unknown>` when the salt was never recorded.
    pub(crate) fn describe(&self) -> String {
        match self.salt {
            Some(s) => format!("seq {} salt {s}", self.checkpoint_seq),
            None => format!("seq {} salt <unknown>", self.checkpoint_seq),
        }
    }
}

/// R858-B18 — the 32-byte `-wal` file header, read straight off disk with no
/// engine open. Ground truth about which WAL generation is on disk and how far
/// it has been written, independent of anything turso caches.
///
/// Only the three fields that move are kept. The rest of the header (magic,
/// format version, the two header checksums) is either constant for a given
/// build or a function of these; a change to any of it that did *not* move one
/// of these three would not be a change this crate can act on.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WalFileHeader {
    /// Page size the WAL's frames carry (bytes 8..12).
    pub page_size: u32,
    /// Checkpoint sequence (bytes 12..16) — the field `tail_frames` used to key
    /// restart detection on before [`WalGeneration`] paired it with the salt.
    pub checkpoint_seq: u32,
    /// The generation's salt (bytes 16..24).
    pub salt: WalSalt,
}

impl WalFileHeader {
    /// SQLite's WAL header is exactly this many bytes, ahead of frame 1.
    pub const SIZE: usize = 32;

    /// Parse a WAL header out of the first [`Self::SIZE`] bytes of a `-wal`
    /// file. `None` for anything shorter — a truncated or freshly-created WAL
    /// has no generation to name yet, which is an honest unknown rather than an
    /// error (see [`WalGeneration`]'s `salt: None`).
    pub fn parse(bytes: &[u8]) -> Option<Self> {
        if bytes.len() < Self::SIZE {
            return None;
        }
        let be = |o: usize| u32::from_be_bytes(bytes[o..o + 4].try_into().unwrap());
        Some(Self {
            page_size: be(8),
            checkpoint_seq: be(12),
            salt: WalSalt::from_wal_file_header(bytes),
        })
    }

    /// Read `{db_path}-wal`'s header. `Ok(None)` when the sidecar is absent
    /// (WAL folded and deleted, or journal_mode != WAL) or too short to parse.
    pub fn read(db_path: &str) -> Result<Option<Self>> {
        match std::fs::File::open(format!("{db_path}-wal")) {
            Ok(mut f) => {
                let mut buf = [0u8; Self::SIZE];
                let mut filled = 0usize;
                loop {
                    match std::io::Read::read(&mut f, &mut buf[filled..]) {
                        Ok(0) => break,
                        Ok(n) => filled += n,
                        Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
                        Err(e) => {
                            return Err(e).with_context(|| format!("reading {db_path}-wal header"))
                        }
                    }
                    if filled == Self::SIZE {
                        break;
                    }
                }
                Ok(Self::parse(&buf[..filled]))
            }
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
            Err(e) => Err(e).with_context(|| format!("opening {db_path}-wal")),
        }
    }
}

/// R858-B18 — everything about a source database's **files** that must hold
/// still for a copy of them to be a point-in-time image, sampled with no engine
/// open and no lock taken.
///
/// This is the other half of reading a foreign-written database. A ReadOnly
/// open ([`CoreWalSeam::open_reader`]) gets us in the door without stealing the
/// whole-file lock, but it buys **no shared locking protocol** with upstream C
/// SQLite: turso locks whole files with `fcntl`, C SQLite uses byte-range locks
/// plus the `-shm` WAL index, and neither engine observes the other's. So a
/// foreign checkpoint landing in the middle of our copy can fold WAL frames
/// into the main file we have already half-read, and the result is a torn image
/// that still passes `PRAGMA integrity_check`.
///
/// Rather than reimplement SQLite's reader protocol (registering a read-mark in
/// the `-shm` WAL index — a research project and a permanent compatibility
/// liability against an engine we do not control), [`raw_consistent_copy_live`]
/// uses textbook **optimistic validation**: sample this before the copy, sample
/// it again after, and accept the copy only if nothing moved. That converts
/// "may silently read torn state" into "detects torn state and refuses", needs
/// no cooperation from the foreign engine, and costs two stats and a 132-byte
/// read per attempt.
///
/// ## Why these fields
///
/// - `wal` (salt + checkpoint_seq) moves on **every** WAL reset, in both fold
///   regimes — fresh randomness on a writer restart, `salt1` incrementing on an
///   in-process autocheckpoint (both measured; see [`WalSalt`]).
/// - `main_len` moves when a checkpoint grows the main database.
/// - `change_counter` (main header bytes 24..28) moves on every write to the
///   main file — i.e. on every checkpoint — even one that leaves its length
///   alone. It is meaningful here **only because the foreign writer is C
///   SQLite**: turso does not maintain this field (it stays `1` in every
///   journal mode, verified — see `snapshot.rs`'s two-gate rationale), which is
///   exactly why the WAL salt carries the weight and this one is corroboration.
/// - `wal_len` is recorded for the report but deliberately **not** part of the
///   accept/reject test — see [`Self::stable_across`], which is the comparison
///   to use. There is no `PartialEq` on this type on purpose: a bare `==` would
///   silently include `wal_len` and refuse every copy taken while the
///   application was merely writing.
#[derive(Debug, Clone, Copy)]
pub struct SourceFingerprint {
    /// Length of the main database file.
    pub main_len: u64,
    /// The main header's change counter, or `None` when the file is too short
    /// to carry a SQLite header at all (a database whose page 1 still lives
    /// only in the WAL). Unknown-and-unknown compares equal, which is safe
    /// because `main_len` participates in the same comparison.
    pub change_counter: Option<u32>,
    /// The `-wal` header, or `None` when there is no WAL sidecar.
    pub wal: Option<WalFileHeader>,
    /// Length of the `-wal` file (`0` when absent).
    pub wal_len: u64,
}

impl SourceFingerprint {
    /// Offset of the change counter in SQLite's 100-byte database header.
    const CHANGE_COUNTER_OFFSET: usize = 24;

    /// Sample the fingerprint of the database at `db_path`. Touches nothing:
    /// two metadata calls plus a 28-byte and a 32-byte read.
    pub fn read(db_path: &str) -> Result<Self> {
        let mut hdr = [0u8; Self::CHANGE_COUNTER_OFFSET + 4];
        let main_len = match std::fs::File::open(db_path) {
            Ok(mut f) => {
                let len = f
                    .metadata()
                    .with_context(|| format!("stat {db_path}"))?
                    .len();
                if len as usize >= hdr.len() {
                    std::io::Read::read_exact(&mut f, &mut hdr)
                        .with_context(|| format!("reading {db_path} header"))?;
                }
                len
            }
            Err(e) => return Err(e).with_context(|| format!("opening source db {db_path}")),
        };
        let change_counter = (main_len as usize >= hdr.len()).then(|| {
            u32::from_be_bytes(hdr[Self::CHANGE_COUNTER_OFFSET..].try_into().unwrap())
        });
        Ok(Self {
            main_len,
            change_counter,
            wal: WalFileHeader::read(db_path)?,
            wal_len: std::fs::metadata(format!("{db_path}-wal"))
                .map(|m| m.len())
                .unwrap_or(0),
        })
    }

    /// True when nothing that can **tear** a copy moved between `self` (sampled
    /// before) and `after` (sampled after). This is the accept test in
    /// [`raw_consistent_copy_live`], and it is narrower than field equality on
    /// purpose.
    ///
    /// ## What can tear the copy, and what cannot
    ///
    /// The copy reads the main file, then replays WAL frames `1..=max_frame`
    /// captured when the seam opened. Against that algorithm:
    ///
    /// - **A checkpoint tears it.** It rewrites pages of the main file *and*
    ///   resets the WAL, so our already-read main bytes and our frame reads can
    ///   straddle the fold — replaying pre-fold frames over post-fold pages
    ///   rolls pages backwards. Caught: a checkpoint bumps `change_counter`
    ///   and/or `main_len`, and a WAL restart re-rolls the salt and the sequence
    ///   (measured in both regimes, `examples/foreign_checkpoint_probe.rs`
    ///   probes A and E).
    /// - **A plain append does NOT tear it.** SQLite only ever appends frames
    ///   within a generation, and only a reset (which re-rolls `salt1`) lets it
    ///   overwrite an existing frame. So frames `1..=max_frame` are immutable
    ///   for as long as the salt holds, and a writer that commits during our
    ///   copy just means our image is a slightly earlier point in time — which
    ///   is what a point-in-time copy *is*.
    ///
    /// Which is why `wal_len` and the frame count are excluded. Including them
    /// buys no additional safety and costs a refusal on every copy taken while
    /// the application is writing at all — turning a working backup into one
    /// that only succeeds against an idle database. R858-B18 measured that
    /// difference rather than assuming it; see probe H.
    pub fn stable_across(&self, after: &Self) -> bool {
        self.main_len == after.main_len
            && self.change_counter == after.change_counter
            && self.wal == after.wal
    }

    /// One-line rendering for the refusal message, so a failure names what
    /// actually moved instead of asserting "something did".
    pub(crate) fn describe(&self) -> String {
        let wal = match self.wal {
            Some(h) => format!("seq {} salt {}", h.checkpoint_seq, h.salt),
            None => "<no WAL>".to_string(),
        };
        format!(
            "main {}B change_counter {} | wal {}B {wal}",
            self.main_len,
            self.change_counter.map_or("<unknown>".to_string(), |c| c.to_string()),
            self.wal_len,
        )
    }
}

/// Subset of `turso_core::Connection`'s `feature = "conn_raw_api"` surface
/// we use. Every WAL call into turso goes through this trait. A future
/// signature shift becomes a one-impl delta.
pub trait WalSeam {
    /// Snapshot the WAL position (checkpoint seq + max_frame).
    fn wal_state(&self) -> Result<Watermark>;

    /// Fetch frame `frame_no` (1-based) into `buf` (must be
    /// `WAL_FRAME_HEADER_SIZE + page_size` bytes). Returns the page number
    /// the frame applies to and the post-frame DB size (in pages) — non-zero
    /// `db_size` marks a commit frame.
    fn wal_get_frame(&self, frame_no: u64, buf: &mut [u8]) -> Result<FrameInfo>;

    /// Take WAL ownership: turn off both the auto-checkpoint and the
    /// auto-WAL-restart so our watermark stays meaningful across calls. The
    /// in-tree consumer (`cli/sync_server.rs`) makes the same move at startup.
    fn wal_auto_actions_disable(&self);
}

/// Restore-side counterpart to [`WalSeam`]: the three `wal_insert_*` calls
/// `cli/sync_server.rs` clients use to replay frames into a fresh DB. Same
/// rationale — one impl block to update if upstream renames.
///
/// Ordering contract: `begin` → N × `insert_frame(monotonic frame_no)` → `end`.
/// `end(false)` is the crash-safe default: the engine rolls back any suffix of
/// frames written after the last commit frame (`db_size > 0`) in the session.
pub trait WalInsertSeam {
    /// Open a write transaction with auto-checkpoint/restart suppressed so our
    /// monotonically-numbered inserts aren't reshuffled mid-session.
    fn wal_insert_begin(&self) -> Result<()>;

    /// Insert `frame` (a 24-byte header + `page_size` bytes of page data) at
    /// position `frame_no` (1-based, must be exactly `prev + 1` — gaps error).
    /// Identical content at an already-written position is a no-op (the engine
    /// compares and returns OK).
    fn wal_insert_frame(&self, frame_no: u64, frame: &[u8]) -> Result<()>;

    /// Close the session. With `force_commit = false` (the restore default) the
    /// engine drops any frames after the last commit frame — automatic
    /// crash-consistency for a tail captured mid-transaction. `force_commit = true`
    /// commits even an uncommitted suffix; not used by restore.
    fn wal_insert_end(&self, force_commit: bool) -> Result<()>;
}

/// Frame-header projection (what we need for streaming + commit detection).
/// Mirrors `turso_core::types::WalFrameInfo` without re-exporting the type.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FrameInfo {
    pub page_no: u32,
    /// Number of pages in the DB after this frame's commit, or `0` if the
    /// frame is mid-transaction.
    pub db_size: u32,
}

impl FrameInfo {
    pub fn is_commit_frame(&self) -> bool {
        self.db_size > 0
    }
}

/// Real WAL seam backed by a live `turso_core::Connection`.
pub struct CoreWalSeam {
    conn: Arc<turso_core::Connection>,
}

impl CoreWalSeam {
    /// Open a **writable** connection at `path` and disable auto-checkpoint /
    /// auto-restart so the caller owns WAL maintenance. Requires `turso_core`
    /// with `features = ["conn_raw_api"]` (set in this crate's Cargo.toml).
    ///
    /// This takes turso's **whole-file exclusive `fcntl` lock**, so it is
    /// mutually exclusive with any other process holding the database open —
    /// including upstream C SQLite (measured: `examples/foreign_checkpoint_probe.rs`
    /// probes D and G). That is correct for the restore/apply direction, which
    /// owns the destination file outright, and wrong for reading a live source:
    /// use [`Self::open_reader`] there.
    pub fn open(path: &str) -> Result<Self> {
        let io: Arc<dyn turso_core::IO> =
            Arc::new(turso_core::PlatformIO::new().context("creating turso_core PlatformIO")?);
        let db = turso_core::Database::open_file(io, path)
            .with_context(|| format!("opening turso_core db {path}"))?;
        let conn = db.connect().context("connecting to turso_core db")?;
        // Take WAL ownership — same first move sync_server.rs makes.
        conn.wal_auto_actions_disable();
        Ok(Self { conn })
    }

    /// R858-B18 — open a **read-only** connection at `path`, taking **no**
    /// whole-file lock. This is the constructor the backup direction wants: a
    /// backup is a reader, and it must not lock out the application whose
    /// database it is reading.
    ///
    /// `OpenFlags::ReadOnly` is what buys that, per handle and with no
    /// process-wide effect. `turso_core-0.7.2/io/unix.rs:67-72` takes the
    /// exclusive lock only when
    /// `env::var(ENV_DISABLE_FILE_LOCK).is_err() && !flags.intersects(ReadOnly | NoLock)`;
    /// `io_uring.rs:463` and `windows.rs:305` carry the identical condition, so
    /// this is not a unix-only accident. The `LIMBO_DISABLE_FILE_LOCK=1` escape
    /// hatch reaches the same no-lock state but does it for **every** open in
    /// the process, including the writable ones — never use it here.
    ///
    /// ## Two things this does NOT give you
    ///
    /// 1. **No shared locking protocol.** Getting in without a lock is not
    ///    coordination: a foreign checkpoint can still land mid-read. That is
    ///    what [`SourceFingerprint`] validation is for, and why
    ///    [`raw_consistent_copy_live`] pairs the two rather than shipping this
    ///    flag alone — the flag alone converts "refuses to open" into "may
    ///    return a torn image", which is strictly worse.
    /// 2. **No escape from turso's process-global registry.**
    ///    `Database::open_file_with_flags` consults `DATABASE_MANAGER`, keyed by
    ///    file id, *before* it looks at the flags (`lib.rs:940`), and hands back
    ///    an already-open `Database` with its own flags discarded. So in a
    ///    process that already holds a writable handle on this exact file, this
    ///    call returns that writable handle — same as `snapshot.rs`'s
    ///    `upload_base_snapshot` doc records. Cross-process (the headscale case)
    ///    is unaffected: the registry is per-process.
    pub fn open_reader(path: &str) -> Result<Self> {
        let io: Arc<dyn turso_core::IO> =
            Arc::new(turso_core::PlatformIO::new().context("creating turso_core PlatformIO")?);
        let db = turso_core::Database::open_file_with_flags(
            io,
            path,
            turso_core::OpenFlags::ReadOnly,
            turso_core::DatabaseOpts::new(),
            None,
        )
        .with_context(|| format!("opening turso_core db {path} read-only"))?;
        let conn = db.connect().context("connecting to turso_core db")?;
        // Belt and braces: a read-only connection cannot checkpoint anyway, but
        // the seam contract is that nothing we hold folds the WAL under us.
        conn.wal_auto_actions_disable();
        Ok(Self { conn })
    }

    /// Wrap an already-open connection. The caller is responsible for having
    /// called `wal_auto_actions_disable()` on it.
    pub fn from_conn(conn: Arc<turso_core::Connection>) -> Self {
        Self { conn }
    }
}

impl WalSeam for CoreWalSeam {
    fn wal_state(&self) -> Result<Watermark> {
        let s = self.conn.wal_state().context("turso_core wal_state")?;
        Ok(Watermark {
            checkpoint_seq: s.checkpoint_seq_no,
            last_frame: s.max_frame,
        })
    }

    fn wal_get_frame(&self, frame_no: u64, buf: &mut [u8]) -> Result<FrameInfo> {
        let info = self
            .conn
            .wal_get_frame(frame_no, buf)
            .with_context(|| format!("turso_core wal_get_frame({frame_no})"))?;
        Ok(FrameInfo {
            page_no: info.page_no,
            db_size: info.db_size,
        })
    }

    fn wal_auto_actions_disable(&self) {
        self.conn.wal_auto_actions_disable();
    }
}

impl WalInsertSeam for CoreWalSeam {
    fn wal_insert_begin(&self) -> Result<()> {
        self.conn
            .wal_insert_begin()
            .context("turso_core wal_insert_begin")
    }

    fn wal_insert_frame(&self, frame_no: u64, frame: &[u8]) -> Result<()> {
        self.conn
            .wal_insert_frame(frame_no, frame)
            .with_context(|| format!("turso_core wal_insert_frame({frame_no})"))?;
        Ok(())
    }

    fn wal_insert_end(&self, force_commit: bool) -> Result<()> {
        self.conn
            .wal_insert_end(force_commit)
            .context("turso_core wal_insert_end")
    }
}

impl CoreWalSeam {
    /// Resize this connection's page cache toward `target_kb` kilobytes via
    /// the standard `PRAGMA cache_size` surface (negative value = KB, per
    /// SQLite's own convention — see `turso_core::translate::pragma`'s
    /// `update_cache_size`) rather than reaching into `turso_core`'s
    /// `Pager::change_page_cache_size` / `CacheResizeResult` directly: the
    /// latter are technically reachable (`Connection::get_pager()` and
    /// `Pager`/`Page`/`PageRef` are all `pub use`d at the crate root) but
    /// `CacheResizeResult` itself is not re-exported, so calling it from
    /// here would mean handling a value of an unnameable type. The pragma
    /// path exercises the exact same resize logic through turso_core's own
    /// public, documented SQL surface instead.
    ///
    /// R574-F3: a warm applier trims its cache hard immediately on attach —
    /// it never serves reads, so cached pages are pure standing RSS cost
    /// (see the R574-T1 measurement this sizes against).
    pub(crate) fn trim_page_cache_kb(&self, target_kb: i64) -> Result<()> {
        anyhow::ensure!(
            target_kb > 0,
            "trim_page_cache_kb: target_kb must be positive, got {target_kb}"
        );
        self.conn
            .execute(format!("PRAGMA cache_size = -{target_kb}"))
            .with_context(|| format!("PRAGMA cache_size = -{target_kb}"))
    }
}

/// R761-T1: the tail cadence this crate recommends when a caller has no
/// reason of its own to pick a different one — **60 s**, and the number is
/// a measurement result, not a guess.
///
/// ## The measurement
///
/// `examples/tail_sweep_harness.rs`, run 2026-08-13 against a casual-app
/// workload (single-row transactions into a table with a secondary index),
/// sweeping `writes_per_tail` — how many application writes accumulate
/// between two `tail_frames` calls:
///
/// | writes_per_tail | PUTs/write | frames/write |
/// |----------------:|-----------:|-------------:|
/// |               1 |       4.03 |         2.03 |
/// |               5 |       2.43 |         2.03 |
/// |              25 |       2.11 |         2.03 |
/// |             100 |       2.05 |         2.03 |
///
/// Those four points are not four independent facts. Every `tail_frames`
/// call that uploads anything writes exactly **two fixed objects** beyond
/// the frames themselves — one generation manifest and one watermark CAS —
/// so
///
/// ```text
/// PUTs/write = frames/write + 2 / writes_per_tail
/// ```
///
/// which reproduces all four measured rows to the last digit: `2.03 + 2/1`,
/// `2.03 + 2/5`, `2.03 + 2/25`, `2.03 + 2/100`. frames/write is FLAT across
/// the sweep — it is a property of the schema and transaction shape, and
/// cadence cannot touch it. The entire lever this default pulls is the
/// `2 / writes_per_tail` term.
///
/// ## Why 60 s and not longer
///
/// That term has sharply diminishing returns. Of the total reduction
/// available (4.03 → 2.05), moving from `writes_per_tail` 1 → 5 captures
/// 81 %, 1 → 25 captures 97 %, and everything from 25 → 100 is the last
/// 3 %. So the target is the 25 region, not 100.
///
/// Converting that writes axis into a time interval needs a write RATE,
/// and the honest one is the rate DURING an active session — W313 §3.1's
/// "~10 writes/day" arrives clustered in bursts, not spread evenly, and an
/// idle tenant costs nothing at any cadence (a tail call with no new frames
/// uploads no objects at all, so a long interval only ever helps a tenant
/// that is actively writing). Across the burst rates a casual app produces,
/// ~0.1–1 write/s:
///
/// - 15 s (what a 30 s RPO bound derives today) spans 1.5–15 writes/tail
///   → 3.36–2.16 PUTs/write. The slow-burst end is the worst case the
///   ticket is named after, and it is nearly the full 4.03.
/// - **60 s spans 6–60 writes/tail → 2.36–2.06 PUTs/write.**
/// - 300 s spans 30–300 → 2.10–2.04: at most 0.26 PUTs/write better than
///   60 s, bought with a 5× wider data-loss window on exactly the tenants
///   that are actively writing. Not a trade worth making for 3 % of a
///   bill.
///
/// 60 s is where the curve has flattened but the exposure window is still
/// something an operator can say out loud.
///
/// ## What it did not fix, and what did
///
/// The 2.03 frames/write floor was untouchable from here — it was one object
/// per WAL frame, and cadence cannot amortize a per-frame cost. **R761-F2
/// removed it** by uploading a tail call's frames as one ranged object, so
/// the curve above is now
///
/// ```text
/// PUTs/write = 3 / writes_per_tail
/// ```
///
/// (one batch object + manifest + watermark), i.e. 3.00 / 0.60 / 0.12 / 0.03
/// at the same four points — a 26 % cut at `writes_per_tail = 1` and 98.5 %
/// at 100. **That does not move this default**, and the reasoning above is
/// why rather than an accident: the 60 s choice was made against the shape
/// of the curve, and what batching changes is its scale. Over the same
/// 0.1–1 write/s burst band, going 60 s → 300 s now buys 0.50 → 0.10
/// PUTs/write at the slow end and 0.05 → 0.01 at the fast one — against a
/// pre-batching worst case of 3.36, an absolute difference small enough
/// that RPO exposure is the only term still worth optimizing here. The
/// measured points above are kept as the pre-batching baseline the
/// reduction is stated against.
///
/// Re-run the harness and revisit both numbers whenever schema shape, page
/// size, or turso's WAL behaviour changes:
/// `cargo run -p turso-backup --example tail_sweep_harness`.
///
/// This crate still schedules nothing itself (R574-T4's "document the
/// contract" choice stands) — the constant is the number a caller's
/// scheduler should adopt absent a reason not to, and
/// [`DEFAULT_RPO_TARGET`] is the bound that goes with it.
pub const DEFAULT_TAIL_INTERVAL: Duration = Duration::from_secs(60);

/// R761-T1: the RPO bound that goes with [`DEFAULT_TAIL_INTERVAL`] — twice
/// it, because tailing *at* the bound makes ordinary scheduling jitter read
/// as a breach, and tailing at half of it means one missed tick still lands
/// inside the promise. See [`DEFAULT_TAIL_INTERVAL`] for why the cadence is
/// 60 s; this is that number expressed as the promise rather than the
/// mechanism, and it is what belongs in [`StreamConfig::rpo_target`].
pub const DEFAULT_RPO_TARGET: Duration = Duration::from_secs(120);

/// Configuration for a streaming session.
pub struct StreamConfig<'a> {
    /// Object-store key of the base tier-1a snapshot the frames replay onto.
    /// Recorded in every generation manifest; restore re-fetches it.
    pub base_snapshot_key: &'a str,
    /// Page size of the base snapshot — read it from the snapshot header
    /// (offset 16, `u16` big-endian; the on-disk value `1` means 65 536).
    /// Required because the seam does not return it and `sync_server.rs`'s
    /// 4 KB hardcode is the wrong default to inherit.
    pub page_size: usize,
    /// R574-F2: bounded spill buffer + overflow policy + 429/503 backoff
    /// for the R2 upload side of the drain loop. `Default` is
    /// behavior-preserving (`BackpressurePolicy::Fail` — errors bubble
    /// immediately, same as before this field existed).
    pub backpressure: BackpressureConfig,
    /// R574-T4: the stated RPO bound — the caller's own scheduler is
    /// expected to invoke `tail_frames` often enough that the watermark
    /// never goes stale past this. `None` (the default) means no target is
    /// asserted; [`RpoStatus::breached`] is always `false` in that case,
    /// but [`RpoStatus::watermark_age`] is still reported so a caller can
    /// observe the real gap before picking a number.
    ///
    /// R761-T1: [`DEFAULT_RPO_TARGET`] (120 s, tailed at
    /// [`DEFAULT_TAIL_INTERVAL`] = 60 s) is the number to put here absent a
    /// reason to pick another — its doc carries the measured write-op table
    /// that justifies it. `None` stays the default so that a caller who has
    /// not thought about RPO never gets a breach flag it did not ask for.
    pub rpo_target: Option<Duration>,
    /// R732-F2 (W245): this writer's **fencing token** — the per-tenant epoch
    /// handed out by yubaba's raft state machine
    /// (`YubabaState::tenant_fencing_token`). It is stamped into every frame
    /// key, every generation manifest, and the watermark sidecar, and it is
    /// checked before a single frame is uploaded: a writer whose epoch is
    /// *lower* than the one already recorded at the sink is a stale owner and
    /// bounces with [`StreamOutcome::Fenced`] rather than interleaving its
    /// frames with the real owner's.
    ///
    /// **`0` means unfenced**, and is the behaviour-preserving default for a
    /// single-writer deployment that has no ownership authority to ask. Note
    /// that unfenced is not exempt: a `0` writer is still fenced by any sink
    /// already stamped with a real epoch, which is exactly what should happen
    /// when a tenant has been claimed and a legacy streamer is still running.
    pub epoch: u64,
    /// R732-F2: opaque label for *who* holds `epoch` — a yubaba node id, a
    /// hostname, whatever the caller finds useful. Recorded in the manifest
    /// and never interpreted here. Purely diagnostic: when you are staring at
    /// a fenced stream at 3am, "which owner wrote generation 7" is the first
    /// question, and the epoch alone does not answer it.
    pub owner: Option<&'a str>,
    /// R736-T2 (W250): the **cross-cell** fence — this writer's belief about
    /// the global tenant→cell pointer's generation
    /// (`yah_tenant_pointer::PointerRecord::generation`). `epoch` alone
    /// fences *within* one raft group; it says nothing when ownership moves
    /// to a different cell, because the two cells run independent raft
    /// groups that share no epoch counter. Checked alongside `epoch` before
    /// any frame is uploaded: a writer whose generation is *lower* than the
    /// one already recorded at the sink is a stale cell and bounces with
    /// [`StreamOutcome::Fenced`], exactly like a stale epoch.
    ///
    /// **`0` means unfenced** — the same behaviour-preserving default as
    /// `epoch`, for a single-cell deployment that has no global pointer to
    /// ask. This mirrors `yah_tenant_pointer`'s `FIRST_GENERATION = 1`: `0`
    /// is what an unfenced writer defaults to and must never be mistakable
    /// for a live cross-cell owner. The caller (yubaba's control plane)
    /// resolves the pointer and hands the generation in as a plain `u64` —
    /// this crate does not link `yah_tenant_pointer` to read it itself.
    pub pointer_generation: u64,
}

/// What a [`tail_frames`] call did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StreamOutcome {
    /// No new frames since the last tail — sink is already current.
    Empty {
        watermark: Watermark,
        /// R574-T4: staleness of the last durably-persisted watermark —
        /// still meaningful here, since "nothing new to stream" and "the
        /// caller's scheduler stopped invoking us" look identical from the
        /// engine's side and only this field tells them apart.
        rpo: RpoStatus,
    },
    /// Uploaded a contiguous range of frames and wrote a generation manifest.
    Streamed {
        generation_key: String,
        first_frame: u64,
        last_frame: u64,
        checkpoint_seq: u32,
        frame_count: u64,
        /// R574-F2: backpressure activity during this call (policy,
        /// high-water spill-buffer occupancy, shed count, throttle retries).
        backpressure: BackpressureReport,
        /// R574-T4: see [`StreamOutcome::Empty::rpo`].
        rpo: RpoStatus,
    },
    /// The live WAL is not provably the one the sidecar watermark was taken
    /// from, so this call re-uploaded frames `1..N` from the top rather than
    /// resuming.
    ///
    /// R858-B19 widened this from "`checkpoint_seq` advanced" to "the
    /// [`WalGeneration`] did not prove itself unchanged", which is why both
    /// fields are now generations rather than bare sequence numbers: the
    /// motivating case is a writer restart where the sequence reads `0 -> 0`
    /// and only the salt moved. `previous_generation.salt == None` names the
    /// third case — a sidecar written before the salt existed, restarted
    /// because it cannot be checked, not because it was seen to change.
    Restarted {
        generation_key: String,
        previous_generation: WalGeneration,
        new_generation: WalGeneration,
        first_frame: u64,
        last_frame: u64,
        frame_count: u64,
        /// R574-F2: see [`StreamOutcome::Streamed::backpressure`].
        backpressure: BackpressureReport,
        /// R574-T4: see [`StreamOutcome::Empty::rpo`].
        rpo: RpoStatus,
    },
    /// R574-F2: `BackpressurePolicy::Shed` dropped every buffered frame in
    /// this call before any of them persisted (R2 was throttling harder
    /// than the spill buffer + backoff could absorb). No manifest/watermark
    /// was written — the next `tail_frames` call re-attempts the same
    /// range from the unchanged prior watermark. Distinct from `Empty`,
    /// which means the engine itself had nothing new.
    Shed {
        checkpoint_seq: u32,
        first_frame: u64,
        last_frame: u64,
        backpressure: BackpressureReport,
        /// R574-T4: see [`StreamOutcome::Empty::rpo`].
        rpo: RpoStatus,
    },
    /// R732-F2 (W245) / R736-T2 (W250): **this writer is a stale owner and
    /// wrote nothing.** Either the sink's watermark is stamped with an epoch
    /// higher than [`StreamConfig::epoch`] (ownership moved within the cell),
    /// or with a pointer generation higher than
    /// [`StreamConfig::pointer_generation`] (ownership moved to a different
    /// cell) — the two-level fence bounces on either. Detected before the
    /// first frame upload, so a fenced call is a pure read — no frames, no
    /// manifest, no watermark write.
    ///
    /// This is the outcome the whole fencing design exists to produce. Without
    /// it a partitioned old master and a freshly-promoted new master both
    /// stream into the same prefix and silently corrupt each other; with it
    /// the loser finds out on its very next tail and can stop.
    ///
    /// Deliberately carries no [`RpoStatus`]: a fenced writer's view of
    /// watermark staleness is not its stream's RPO any more, and reporting one
    /// here would page the wrong operator about the wrong node.
    Fenced {
        /// The epoch recorded at the sink — the real owner's token.
        current_epoch: u64,
        /// The (lower) epoch this writer tried to stream under.
        our_epoch: u64,
        /// R736-T2: the pointer generation recorded at the sink.
        current_pointer_generation: u64,
        /// R736-T2: the (lower) generation this writer tried to stream under.
        our_pointer_generation: u64,
    },
}

impl StreamOutcome {
    /// This call's RPO snapshot, or `None` for [`StreamOutcome::Fenced`] —
    /// see that variant's doc for why it deliberately carries none.
    ///
    /// R782: the accessor a caller (`tenant-streamer`'s tail loop) uses to
    /// push `watermark_age` onward without re-deriving this match on every
    /// call site that needs it.
    pub fn rpo(&self) -> Option<&RpoStatus> {
        match self {
            StreamOutcome::Empty { rpo, .. }
            | StreamOutcome::Streamed { rpo, .. }
            | StreamOutcome::Restarted { rpo, .. }
            | StreamOutcome::Shed { rpo, .. } => Some(rpo),
            StreamOutcome::Fenced { .. } => None,
        }
    }

    /// This call's backpressure activity, or `None` for the two outcomes that
    /// never reached the drain loop ([`StreamOutcome::Empty`] had nothing to
    /// send, [`StreamOutcome::Fenced`] was refused before the first frame).
    ///
    /// R760-B8: the accessor a *multi-tenant* caller needs. `Streamed` is not
    /// the same thing as "the sink took everything" — under
    /// [`BackpressurePolicy::Shed`] a call that persisted a partial prefix and
    /// dropped the rest reports `Streamed` with a nonzero
    /// [`BackpressureReport::frames_shed`], and a caller that only matches the
    /// variant cannot tell that apart from a clean tail. roadcase's shard
    /// flusher reads this on every arm so a struggling cell is nameable from
    /// its own metrics rather than by bisecting tenants.
    pub fn backpressure(&self) -> Option<&BackpressureReport> {
        match self {
            StreamOutcome::Streamed { backpressure, .. }
            | StreamOutcome::Restarted { backpressure, .. }
            | StreamOutcome::Shed { backpressure, .. } => Some(backpressure),
            StreamOutcome::Empty { .. } | StreamOutcome::Fenced { .. } => None,
        }
    }
}

/// R574-T4: watermark-staleness snapshot for RPO drift alerting, computed
/// fresh on every `tail_frames` call against [`StreamConfig::rpo_target`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct RpoStatus {
    /// The stated bound from `StreamConfig::rpo_target`, echoed back for
    /// convenience (so a caller reading only the outcome still knows what
    /// was being enforced).
    pub target: Option<Duration>,
    /// Elapsed time since the watermark last durably advanced, measured at
    /// the start of this call. `None` when the sink has never persisted a
    /// watermark — there is nothing yet to measure staleness against.
    pub watermark_age: Option<Duration>,
    /// `true` when both `target` and `watermark_age` are set and
    /// `watermark_age > target` — the RPO bound is breached and an
    /// orchestrator should alert. Always `false` when no target is
    /// configured.
    pub breached: bool,
}

impl BackupTarget {
    pub(crate) fn watermark_key(&self) -> ObjPath {
        join_key(&self.prefix, "latest.stream-watermark")
    }

    /// R732-F2: frames are namespaced by the writer's fencing epoch, so two
    /// owners at different epochs cannot land on the same key even if they
    /// somehow both get as far as uploading. The epoch check in
    /// [`tail_frames`] is the guard; this key shape is the backstop that makes
    /// a guard failure recoverable (both owners' frames survive and the
    /// manifests say who wrote what) instead of a silent overwrite.
    ///
    /// Epoch `0` keeps the original two-level layout. That is not cosmetic:
    /// backups written before fencing existed are still restorable because
    /// their manifests say `epoch 0` and land back on this branch.
    ///
    /// R761-F2: this is the *read-side legacy* key shape now — nothing writes
    /// one-object-per-frame any more. See [`Self::frame_batch_key`].
    pub(crate) fn frame_key(&self, epoch: u64, checkpoint_seq: u32, frame_no: u64) -> ObjPath {
        let suffix = if epoch == 0 {
            format!("frames/{checkpoint_seq:010}/{frame_no:020}")
        } else {
            format!("frames/{epoch:020}/{checkpoint_seq:010}/{frame_no:020}")
        };
        join_key(&self.prefix, &suffix)
    }

    /// R761-F2: key of a batch object holding frames `first..=last`
    /// concatenated. Same epoch namespacing and zero-padding as
    /// [`Self::frame_key`], so lexical order still matches frame order — one
    /// stream's batches never overlap, because each is a slice of a single
    /// monotonic drain.
    ///
    /// The range is in the key rather than only in the manifest so the sink
    /// stays self-describing: a `ls` of the prefix tells an operator exactly
    /// which frames are present, which is what made the per-frame layout easy
    /// to reason about and is worth keeping.
    ///
    /// Distinguishable from a legacy per-frame key by construction (that one
    /// has no `-`), so a stream that straddles the layout change can hold both
    /// shapes under the same directory without collision.
    pub(crate) fn frame_batch_key(
        &self,
        epoch: u64,
        checkpoint_seq: u32,
        first: u64,
        last: u64,
    ) -> ObjPath {
        let suffix = if epoch == 0 {
            format!("frames/{checkpoint_seq:010}/{first:020}-{last:020}")
        } else {
            format!("frames/{epoch:020}/{checkpoint_seq:010}/{first:020}-{last:020}")
        };
        join_key(&self.prefix, &suffix)
    }

    /// Every object holding generation `m`'s frames, as `(key, first, last)`
    /// in ascending frame order.
    ///
    /// The single place that knows how a generation's frames are laid out:
    /// R761-F2 batches when the manifest carries a `frame_batch` list, the
    /// pre-R761-F2 one-object-per-frame layout when it does not. Restore's
    /// replay and [`crate::puller::WalPuller`] both go through here, so the
    /// two can never drift apart about where a frame lives.
    pub(crate) fn frame_objects_of(
        &self,
        m: &OwnedGenerationManifest,
    ) -> Vec<(ObjPath, u64, u64)> {
        if m.frame_batches.is_empty() {
            (m.first_frame..=m.last_frame)
                .map(|n| (self.frame_key(m.epoch, m.checkpoint_seq, n), n, n))
                .collect()
        } else {
            m.frame_batches
                .iter()
                .map(|&(first, last)| {
                    (
                        self.frame_batch_key(m.epoch, m.checkpoint_seq, first, last),
                        first,
                        last,
                    )
                })
                .collect()
        }
    }

    pub(crate) fn generation_key(&self, unix_nanos: u128) -> ObjPath {
        join_key(
            &self.prefix,
            &format!("generations/gen-{unix_nanos:020}.manifest"),
        )
    }
}

/// R858-B19 — identify the WAL generation `seam` is currently reading, by
/// pulling the salt out of **frame 1's** header.
///
/// Frame 1 rather than the 32-byte WAL header because this goes through the
/// [`WalSeam`] trait, which is the crate's one boundary against `turso_core`:
/// `turso_core::WalState` reports only `checkpoint_seq_no` and `max_frame`, and
/// reading the `-wal` file directly would need a path that
/// [`CoreWalSeam::from_conn`] does not have. Every frame header carries a
/// verbatim copy of the WAL header's salt, so frame 1 answers the same question
/// through machinery every seam already implements — including the mocks, and
/// including roadcase's connection-backed seam.
///
/// `Ok(None)` means the WAL holds no frames, so there is no generation to name
/// yet. That is not a failure: it is the honest "unknown", and
/// [`WalGeneration::is_provably_same_as`] treats it as such.
///
/// Costs one page-sized read per [`tail_frames`] call — negligible next to the
/// frames that call is about to upload, and it is the only thing standing
/// between a WAL recreate and a silently spliced generation chain.
pub(crate) fn read_wal_salt<S: WalSeam>(
    seam: &S,
    page_size: usize,
    last_frame: u64,
) -> Result<Option<WalSalt>> {
    if last_frame == 0 {
        return Ok(None);
    }
    let mut buf = vec![0u8; WAL_FRAME_HEADER_SIZE + page_size];
    seam.wal_get_frame(1, &mut buf)
        .context("reading WAL frame 1 to identify the WAL generation (R858-B19)")?;
    Ok(Some(WalSalt::from_frame_header(&buf)))
}

/// Tail new WAL frames from `seam` into `target`, anchored to a base
/// snapshot. Idempotent and resumable: on the second call only frames after
/// the recorded watermark are uploaded.
///
/// Ordering within a single call:
/// 1. Read [`Watermark`] from the seam (snapshot the current `(checkpoint_seq,
///    max_frame)`) and the current [`WalGeneration`] (that sequence plus the
///    WAL salt, via [`read_wal_salt`]).
/// 2. Read the prior watermark sidecar (if any).
/// 3. Unless the sidecar's generation is *provably* the same WAL as the live
///    one, treat this as a restart: upload frames `1..=max_frame`. R858-B19 —
///    the test is [`WalGeneration::is_provably_same_as`], not a `checkpoint_seq`
///    comparison, because a writer-process restart recreates the WAL at
///    sequence `0` and an unchanged sequence therefore proves nothing.
/// 4. Otherwise upload frames `prior.last_frame+1..=max_frame`.
/// 5. Advance the watermark sidecar with a compare-and-swap on the version
///    read in step 2, then write a generation manifest pointing at the base
///    snapshot + frame range + the batch objects covering it. Frames precede
///    both, so a manifest never references a missing frame; the manifest
///    follows the CAS, so a writer that loses the sidecar race publishes
///    nothing (R732-T3).
///
/// Returns [`StreamOutcome::Empty`] if there is nothing to do (max_frame
/// hasn't advanced and checkpoint_seq is unchanged), or
/// [`StreamOutcome::Fenced`] if this writer's [`StreamConfig::epoch`] has been
/// superseded — either observed up front in step 2 or discovered by losing the
/// CAS in step 5.
pub async fn tail_frames<S: WalSeam>(
    seam: &S,
    target: &BackupTarget,
    cfg: &StreamConfig<'_>,
) -> Result<StreamOutcome> {
    let current = seam.wal_state()?;
    // R858-B19: the salt is what actually names the WAL generation. Read it
    // before anything else touches the sink, so the restart decision below is
    // made against the live WAL rather than inferred from a sequence number
    // that a writer restart silently resets.
    let current_generation = WalGeneration {
        checkpoint_seq: current.checkpoint_seq,
        salt: read_wal_salt(seam, cfg.page_size, current.last_frame)?,
    };
    let persisted = read_watermark(&target.store, &target.watermark_key()).await?;
    // R574-T4: staleness measured at call start, against the *prior*
    // sidecar write — i.e. how long the sink had gone without durable
    // progress before this call ran. On the steady cadence the caller's
    // scheduler promises, this hovers at the invocation interval; a
    // skipped/late run pushes it past `rpo_target` and flips `breached`.
    let rpo = rpo_status(cfg.rpo_target, persisted.as_ref());

    // R732-F2 (W245) / R736-T2 (W250): the two-level fence, checked before
    // anything is written. A sink stamped with a higher epoch than ours means
    // ownership moved within the cell while we were away; a sink stamped with
    // a higher pointer generation means ownership moved to a *different*
    // cell — the two raft groups don't share an epoch counter, so the epoch
    // check alone is blind to that move. Either means every byte we are about
    // to upload belongs to somebody else's stream. Bounce here and the call
    // is a pure read; bounce anywhere later and we have already interleaved
    // frames into the real owner's range.
    if let Some(p) = &persisted {
        if p.epoch > cfg.epoch || p.pointer_generation > cfg.pointer_generation {
            return Ok(StreamOutcome::Fenced {
                current_epoch: p.epoch,
                our_epoch: cfg.epoch,
                current_pointer_generation: p.pointer_generation,
                our_pointer_generation: cfg.pointer_generation,
            });
        }
    }

    // Borrowed, not consumed: the conditional watermark advance at the end of
    // this call needs the object version this read observed (R732-T3).
    let prior = persisted.as_ref().map(|p| p.watermark);

    // R858-B19: the generation the sidecar was written under. `None` for a
    // sidecar that predates the salt field — which reads as *unknown*, not as
    // "the same WAL", and so forces the restart branch exactly once. After that
    // one full re-upload the sink carries a salt and the stream self-heals.
    let prior_generation = persisted.as_ref().map(|p| p.generation);

    // Decide the range to upload. Resuming at `last_frame + 1` is only sound
    // when the live WAL is PROVABLY the one the watermark was taken from: a
    // writer-process restart deletes the `-wal` file and the next writer starts
    // a fresh WAL at checkpoint-sequence 0 with a new salt, so an unchanged
    // sequence is not evidence of anything. Anything short of proof restarts.
    let (start_frame, restarted) = match prior_generation {
        Some(g) if g.is_provably_same_as(&current_generation) => {
            (prior.map(|p| p.last_frame).unwrap_or(0) + 1, false)
        }
        Some(_) => (1, true),
        None => (1, false),
    };

    if current.last_frame < start_frame {
        return Ok(StreamOutcome::Empty { watermark: current, rpo });
    }

    // Drain frames in ascending order through the bounded spill buffer +
    // backpressure policy (R574-F2). A partial upload leaves a prefix (the
    // manifest is written last, so a prefix without a manifest is invisible
    // to restore — the next tail just overwrites the same keys).
    let drain = drain_frames_with_backpressure(
        seam,
        target,
        cfg,
        cfg.epoch,
        current.checkpoint_seq,
        start_frame,
        current.last_frame,
    )
    .await?;

    let Some(uploaded_through) = drain.uploaded_through else {
        // BackpressurePolicy::Shed dropped everything before any frame
        // persisted — no manifest/watermark write, so the next call
        // re-attempts this exact range from the unchanged prior watermark.
        return Ok(StreamOutcome::Shed {
            checkpoint_seq: current.checkpoint_seq,
            first_frame: start_frame,
            last_frame: current.last_frame,
            backpressure: drain.report,
            rpo,
        });
    };

    // R858-B19: everything above sampled the generation ONCE, before the
    // drain. A WAL recreate between calls is what this ticket is about, but
    // nothing stops one landing *during* a call — and then the frames just
    // uploaded are a mix of two WALs, which is the same corruption arriving by
    // a narrower door. Re-read the salt and refuse if it moved.
    //
    // Refusing here is cheap and complete: the watermark has not advanced and
    // no manifest has been written, so the frames that landed are orphaned
    // under keys nothing references — invisible to restore, exactly like the
    // `Shed` path — and the next tail re-derives everything from the sidecar.
    let after = seam.wal_state()?;
    let after_salt = read_wal_salt(seam, cfg.page_size, after.last_frame)?;
    if after_salt != current_generation.salt {
        anyhow::bail!(
            "the source WAL was recreated while this tail was uploading ({} -> {}) — the frames \
             this call read span two WAL generations, so nothing is published and the sink is left \
             exactly as it was; the next tail will restart cleanly against the new WAL",
            current_generation.describe(),
            WalGeneration { checkpoint_seq: after.checkpoint_seq, salt: after_salt }.describe(),
        );
    }

    // Claim the range with a conditional watermark advance, THEN publish the
    // generation manifest. Under a Shed policy that persisted a partial
    // prefix, both cover only `start_frame..=uploaded_through`, not the full
    // engine range.
    //
    // R732-T3 reordered these two. The manifest used to be written last, so
    // that a manifest never referenced a missing frame — that invariant is
    // untouched, because frames still precede both. What the old order could
    // not do is fence: a writer that lost the sidecar race had already
    // published its manifest, which would then sit in the chain at a regressed
    // epoch and make every future restore refuse. Publishing only after the
    // CAS means a fenced writer leaves no manifest at all.
    let cas = write_watermark(
        &target.store,
        &target.watermark_key(),
        Watermark { checkpoint_seq: current.checkpoint_seq, last_frame: uploaded_through },
        current_generation.salt,
        cfg.epoch,
        cfg.pointer_generation,
        persisted.as_ref().and_then(|p| p.version.as_ref()),
    )
    .await?;
    if cas == WatermarkCas::Contended {
        // Somebody replaced the sidecar under us. Re-read to learn who.
        let now = read_watermark(&target.store, &target.watermark_key()).await?;
        let current_epoch = now.as_ref().map_or(0, |p| p.epoch);
        let current_pointer_generation = now.as_ref().map_or(0, |p| p.pointer_generation);
        if current_epoch > cfg.epoch || current_pointer_generation > cfg.pointer_generation {
            // A newer owner won the race. Our frames are orphaned under our
            // own epoch prefix with no manifest naming them, so restore never
            // sees them — the sink is exactly what the winner left.
            return Ok(StreamOutcome::Fenced {
                current_epoch,
                our_epoch: cfg.epoch,
                current_pointer_generation,
                our_pointer_generation: cfg.pointer_generation,
            });
        }
        // Same or lower epoch AND generation: neither fence can tell these two
        // writers apart. That means two processes are streaming the same
        // tenant under the SAME tokens — a caller bug (a duplicate streamer,
        // or an ownership token handed out twice), and exactly the
        // condition that must not be papered over with a retry.
        anyhow::bail!(
            "watermark CAS for {} lost to a concurrent writer at epoch {current_epoch} (pointer generation {current_pointer_generation}) while we hold epoch {} (pointer generation {}) — two streamers share one fencing token",
            target.watermark_key(),
            cfg.epoch,
            cfg.pointer_generation,
        );
    }

    let nanos = unix_nanos();
    let gen_key = target.generation_key(nanos);
    let manifest = format_generation_manifest(GenerationManifest {
        base_snapshot_key: cfg.base_snapshot_key,
        page_size: cfg.page_size,
        checkpoint_seq: current.checkpoint_seq,
        // R858-B19: stamp the generation this range actually came from, so
        // restore can refuse a chain that spans two WALs instead of splicing
        // them. Always `Some` on this path — a `None` salt means an empty WAL,
        // and an empty WAL took the `Empty` return above.
        salt: current_generation.salt,
        first_frame: start_frame,
        last_frame: uploaded_through,
        epoch: cfg.epoch,
        owner: cfg.owner,
        // R761-F2: exactly the batch objects the drain persisted. Restore
        // derives its keys from this list, so it is not a summary of the range
        // — it IS the range's index.
        frame_batches: &drain.frame_batches,
    });
    target
        .store
        .put(&gen_key, manifest.into_bytes().into())
        .await
        .with_context(|| format!("writing generation manifest {gen_key}"))?;

    let frame_count = uploaded_through - start_frame + 1;
    let gen_key = gen_key.to_string();
    if restarted {
        Ok(StreamOutcome::Restarted {
            generation_key: gen_key,
            previous_generation: prior_generation.unwrap_or_default(),
            new_generation: current_generation,
            first_frame: start_frame,
            last_frame: uploaded_through,
            frame_count,
            backpressure: drain.report,
            rpo,
        })
    } else {
        Ok(StreamOutcome::Streamed {
            generation_key: gen_key,
            first_frame: start_frame,
            last_frame: uploaded_through,
            checkpoint_seq: current.checkpoint_seq,
            frame_count,
            backpressure: drain.report,
            rpo,
        })
    }
}

/// Outcome of [`drain_frames_with_backpressure`]: the last frame_no
/// successfully persisted this call (`None` if every buffered frame was
/// shed before any of them landed) plus the backpressure activity report.
struct DrainOutcome {
    uploaded_through: Option<u64>,
    /// R761-F2: the `(first, last)` range of every batch object that actually
    /// landed, ascending and gap-free from the call's `start_frame` through
    /// `uploaded_through` (a batch is popped only on a successful upload, so a
    /// partial drain truncates this list rather than holing it). Copied into
    /// the generation manifest, which is what tells restore where to look.
    frame_batches: Vec<(u64, u64)>,
    report: BackpressureReport,
}

/// Read WAL frames `start_frame..=last_frame` into a bounded spill buffer
/// (`cfg.backpressure.spill_buffer_frames`) and drain them to `target`'s
/// object store through [`put_with_backoff`] (R574-F2), one **batch object**
/// per buffer-full (R761-F2).
///
/// The buffer only refills up to its bound, so once full the loop must
/// resolve the buffered batch — either by a successful upload or by the
/// configured [`BackpressurePolicy`] deciding what "stuck" means:
/// `Block` retries the batch forever on a throttling error (no data loss,
/// but the call can run long); `Fail` bubbles the error once
/// `backoff.max_retries` is exhausted (nothing persists past what already
/// landed); `Shed` drops the whole buffered backlog (loudly, via the
/// returned report) and returns whatever prefix already persisted. Any
/// non-throttling error bubbles immediately regardless of policy — this
/// mechanism is specifically for R2 throttling, not general fault
/// tolerance.
///
/// R761-F2 made the upload unit the buffer's contents rather than its head
/// frame, which is why `spill_buffer_frames` also sizes the largest object
/// this sink will write: `spill_buffer_frames * (24 + page_size)` bytes, ~1 MB
/// at the 256-frame default and a 4 KB page. That coupling is deliberate — the
/// bound already promises a memory ceiling, and a second knob for batch size
/// would only ever be set to some fraction of it.
async fn drain_frames_with_backpressure<S: WalSeam>(
    seam: &S,
    target: &BackupTarget,
    cfg: &StreamConfig<'_>,
    epoch: u64,
    checkpoint_seq: u32,
    start_frame: u64,
    last_frame: u64,
) -> Result<DrainOutcome> {
    let bp = &cfg.backpressure;
    let frame_size = WAL_FRAME_HEADER_SIZE + cfg.page_size;
    let bound = bp.spill_buffer_frames.max(1);

    let mut report = BackpressureReport { policy: bp.policy, ..Default::default() };
    let mut pending: VecDeque<(u64, Vec<u8>)> = VecDeque::new();
    let mut uploaded_through: Option<u64> = None;
    let mut frame_batches: Vec<(u64, u64)> = Vec::new();
    let mut next_to_read = start_frame;
    let mut buf = vec![0u8; frame_size];

    loop {
        // Refill up to the bound while there's more WAL to read. Once full,
        // the buffered batch must be resolved before we accept more.
        while pending.len() < bound && next_to_read <= last_frame {
            seam.wal_get_frame(next_to_read, &mut buf)
                .with_context(|| format!("reading wal frame {next_to_read}"))?;
            pending.push_back((next_to_read, buf.clone()));
            next_to_read += 1;
            report.high_water_frames = report.high_water_frames.max(pending.len());
        }
        let Some(&(first, _)) = pending.front() else {
            break; // Fully drained: nothing buffered, nothing left to read.
        };
        // Non-empty (we just matched `front`), and the buffer is filled in
        // ascending order without gaps, so the back frame closes the range.
        let last = pending.back().map_or(first, |&(n, _)| n);
        let mut body = Vec::with_capacity(pending.len() * frame_size);
        for (_, bytes) in &pending {
            body.extend_from_slice(bytes);
        }
        let key = target.frame_batch_key(epoch, checkpoint_seq, first, last);
        match put_with_backoff(&target.store, &key, body.into(), bp, &mut report.throttle_retries)
            .await
        {
            Ok(_) => {
                pending.clear();
                uploaded_through = Some(last);
                frame_batches.push((first, last));
            }
            Err(e) => match bp.policy {
                BackpressurePolicy::Shed => {
                    report.frames_shed += pending.len() as u64;
                    return Ok(DrainOutcome { uploaded_through, frame_batches, report });
                }
                BackpressurePolicy::Block | BackpressurePolicy::Fail => {
                    return Err(e).with_context(|| {
                        format!("uploading wal frames {first}-{last} to {key}")
                    });
                }
            },
        }
    }
    Ok(DrainOutcome { uploaded_through, frame_batches, report })
}

/// R858-B18 — how many times [`raw_consistent_copy_live`] will re-take a copy
/// whose [`SourceFingerprint`] moved underneath it before giving up loudly.
///
/// Bounded on purpose. An unbounded retry against a database under sustained
/// write pressure is an infinite loop that looks like a hang; a caller that
/// wants to keep trying should be the one deciding how long to keep trying, on
/// its own schedule. Four attempts is enough to ride out an isolated
/// checkpoint (headscale's database measures 94 KB, so an attempt is
/// sub-millisecond) and few enough that a genuinely hot database is reported as
/// hot within a few milliseconds instead of being ground at.
pub const COPY_VALIDATION_ATTEMPTS: u32 = 4;

/// Take a raw, point-in-time-consistent byte image of the database at `db_path`
/// WITHOUT folding the WAL via a `TRUNCATE` checkpoint, and WITHOUT locking out
/// the process that owns it. The live-writer pair of
/// [`crate::dedup::raw_consistent_copy`], which a concurrent writer can make
/// return `busy`.
///
/// Algorithm — the read-only "copy main + replay WAL frames ourselves" path
/// flagged in the working doc and the R005-T1 spike, wrapped in R858-B18's
/// optimistic validation:
///
/// 1. Sample the source's [`SourceFingerprint`] — WAL salt + checkpoint
///    sequence, WAL length, main length, main change counter — straight off
///    disk, with no engine open.
/// 2. Open a fresh `turso_core::Connection` via [`CoreWalSeam::open_reader`],
///    which takes **no** whole-file lock (so the application keeps working) and
///    calls `wal_auto_actions_disable` so our seam can't auto-checkpoint or
///    restart the WAL header mid-read on our connection.
/// 3. Read the main DB file bytes from disk. With auto-actions disabled on our
///    connection the file cannot be folded by us; a concurrent writer in a
///    separate connection only ever extends the WAL (the main file is only
///    written by a checkpoint).
/// 4. Walk WAL frames `1..=max_frame`, replaying each page into the in-memory
///    image at offset `(page_no - 1) * page_size`. Track the last
///    `is_commit_frame` and the corresponding `db_size`. Any frames past the
///    last commit are uncommitted mid-transaction garbage — drop them
///    (crash-consistency, matching restore's `wal_insert_end(false)`).
/// 5. Truncate / grow the image to `db_size * page_size`.
/// 6. Re-sample the fingerprint. **Accept the image only if
///    [`SourceFingerprint::stable_across`] holds** — i.e. only if nothing that
///    can tear the copy moved. A foreign checkpoint inside our read window would
///    splice pre- and post-fold state; discard that image and retry from step 1,
///    up to [`COPY_VALIDATION_ATTEMPTS`] times, then fail. (A plain WAL append
///    is *not* movement for this purpose, and that distinction is what keeps
///    this usable against a database that is actually in use — see
///    `stable_across`.)
///
/// Step 6 is the entire correctness story against a foreign engine, and it is
/// why this function never returns an unvalidated image: a torn WAL replay
/// still produces a structurally valid SQLite file that passes
/// `PRAGMA integrity_check`, so "it parsed" proves nothing. See
/// [`SourceFingerprint`] for why validation rather than SQLite's real
/// `-shm` reader protocol.
///
/// Returned bytes are a self-contained vanilla-SQLite image (page-offset
/// stable, no `-wal` sidecar required), ready to feed
/// [`crate::dedup::snapshot_dedup`]'s content-addressed chunking under a
/// concurrent writer.
pub async fn raw_consistent_copy_live(db_path: &str, page_size: usize) -> Result<Vec<u8>> {
    let image = validated_against_source(db_path, "live copy", || async {
        let seam = CoreWalSeam::open_reader(db_path)
            .with_context(|| format!("opening read-only WAL seam on {db_path}"))?;
        let main_bytes = std::fs::read(db_path)
            .with_context(|| format!("reading main db file {db_path}"))?;
        replay_wal_onto_main(&seam, main_bytes, page_size)
        // Seam dropped at the end of this block, so nothing of ours holds the
        // file while the post-copy fingerprint is sampled.
    })
    .await?;
    anyhow::ensure!(
        image.starts_with(b"SQLite format 3\0"),
        "live consistent copy of {db_path} is not a SQLite database"
    );
    Ok(image)
}

/// R858-B18 — run `take` against the live database at `db_path` and return its
/// result **only** if the source provably held still for the duration.
///
/// This is the optimistic-validation protocol both source-side tiers share:
/// tier 2's WAL-replay copy ([`raw_consistent_copy_live`]) and tier 1a's
/// `VACUUM INTO` (`crate::snapshot`). Both read a database a foreign engine may
/// be checkpointing underneath them, neither can take a lock that engine
/// respects, and both are therefore only correct if a fold inside their read
/// window is *detected*. `what` names the operation in the refusal message.
///
/// Each attempt samples a [`SourceFingerprint`] before and after, then
/// classifies the outcome on both axes — did it succeed, and did the source
/// move:
///
/// | | source held still | source moved |
/// |---|---|---|
/// | **`Ok`** | accept | discard, retry (it may splice pre-/post-fold state) |
/// | **`Err`** | return the error — nothing raced us, so it is real | retry (a symptom of the race) |
///
/// That bottom-right cell is why the fingerprint is load-bearing beyond
/// accept/reject: R858-B18's probe H measured the *common* shape of a caught
/// race not as a clean torn image but as `short read on WAL frame` — a foreign
/// checkpoint truncating the WAL while turso walked it. Without the fingerprint
/// there is no way to tell that transient apart from a genuinely corrupt
/// database, and the two want opposite handling.
///
/// After [`COPY_VALIDATION_ATTEMPTS`] attempts all of which saw movement, this
/// fails loudly and returns nothing.
pub(crate) async fn validated_against_source<T, F, Fut>(
    db_path: &str,
    what: &str,
    mut take: F,
) -> Result<T>
where
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = Result<T>>,
{
    let mut moved: Option<(SourceFingerprint, SourceFingerprint, Option<anyhow::Error>)> = None;
    for attempt in 1..=COPY_VALIDATION_ATTEMPTS {
        if attempt > 1 {
            // Back off between attempts: retrying instantly against a writer
            // mid-checkpoint just spends the whole budget inside one fold.
            tokio::time::sleep(std::time::Duration::from_millis(20 * u64::from(attempt - 1)))
                .await;
        }
        let before = SourceFingerprint::read(db_path)?;
        let attempted = take().await;
        let after = SourceFingerprint::read(db_path)?;
        let stable = before.stable_across(&after);
        match attempted {
            Ok(value) if stable => return Ok(value),
            Ok(_) => moved = Some((before, after, None)),
            Err(e) if !stable => moved = Some((before, after, Some(e))),
            Err(e) => {
                return Err(e).with_context(|| {
                    format!(
                        "{what} of {db_path} failed against a source that did NOT move during \
                         the attempt ({}), so this is not a concurrent-writer race",
                        before.describe()
                    )
                })
            }
        }
    }
    let (before, after, last_err) = moved.expect("COPY_VALIDATION_ATTEMPTS is non-zero");
    let because = match last_err {
        Some(e) => format!("the last attempt also failed mid-read ({e:#})"),
        None => "each attempt produced a result that could not be validated".to_string(),
    };
    anyhow::bail!(
        "refusing a {what} of {db_path}: the source moved under every one of \
         {COPY_VALIDATION_ATTEMPTS} attempts, so nothing could be validated as \
         point-in-time — {because}. The last attempt saw [{}] before and [{}] after, so a \
         foreign writer or checkpointer is active. Retry on your own schedule; NO result \
         is returned, because a torn read of a SQLite database still parses as valid SQLite.",
        before.describe(),
        after.describe(),
    )
}

/// Pure replay of every committed WAL frame visible through `seam` onto
/// `main_bytes`. Split out from [`raw_consistent_copy_live`] so it can be
/// driven by a mock seam in unit tests; the live entry point layers disk I/O
/// and magic-byte validation on top.
///
/// Contract: the highest `is_commit_frame` in `1..=wal_state.last_frame`
/// defines both the post-replay page count and the cutoff for which frames
/// are applied. If no frame in that window is a commit, `main_bytes` is
/// returned unchanged.
pub(crate) fn replay_wal_onto_main<S: WalSeam>(
    seam: &S,
    mut main_bytes: Vec<u8>,
    page_size: usize,
) -> Result<Vec<u8>> {
    anyhow::ensure!(page_size > 0, "page_size must be non-zero");
    let watermark = seam.wal_state()?;

    let frame_size = WAL_FRAME_HEADER_SIZE + page_size;
    let mut buf = vec![0u8; frame_size];

    struct PendingFrame {
        page_no: u32,
        db_size: u32,
        page_bytes: Vec<u8>,
    }
    let mut frames: Vec<PendingFrame> = Vec::new();
    let mut last_commit_idx: Option<usize> = None;
    for frame_no in 1..=watermark.last_frame {
        let info = seam
            .wal_get_frame(frame_no, &mut buf)
            .with_context(|| format!("reading WAL frame {frame_no}"))?;
        anyhow::ensure!(
            info.page_no >= 1,
            "WAL frame {frame_no}: page_no must be >= 1"
        );
        frames.push(PendingFrame {
            page_no: info.page_no,
            db_size: info.db_size,
            page_bytes: buf[WAL_FRAME_HEADER_SIZE..].to_vec(),
        });
        if info.is_commit_frame() {
            last_commit_idx = Some(frames.len() - 1);
        }
    }

    let Some(last_commit) = last_commit_idx else {
        // No committed frames in our view — main file alone is the image.
        // Any uncommitted suffix in the WAL is dropped by construction.
        return Ok(main_bytes);
    };
    let final_db_size = frames[last_commit].db_size as usize;
    let target_size = final_db_size
        .checked_mul(page_size)
        .context("db_size * page_size overflow")?;

    // Grow image to hold the final committed image AND any page slot we'll
    // touch in the commit prefix (a frame may write a page above db_size
    // mid-grow; the final truncate cuts that back to db_size).
    let max_off_needed: usize = frames[..=last_commit]
        .iter()
        .map(|f| f.page_no as usize * page_size)
        .max()
        .unwrap_or(0);
    let need = target_size.max(max_off_needed);
    if main_bytes.len() < need {
        main_bytes.resize(need, 0);
    }
    for f in &frames[..=last_commit] {
        let off = (f.page_no as usize - 1) * page_size;
        main_bytes[off..off + page_size].copy_from_slice(&f.page_bytes);
    }
    main_bytes.truncate(target_size);

    Ok(main_bytes)
}

/// Summary of a [`restore_latest_stream`] call.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RestoreOutcome {
    /// The tier-1a snapshot key every generation manifest referenced (must agree).
    pub base_snapshot_key: String,
    /// Sole `checkpoint_seq_no` across the replayed manifests (v1 refuses to
    /// span a WAL restart — see [`validate_generation_chain`]).
    pub checkpoint_seq: u32,
    /// Number of generation manifests replayed (≥ 1).
    pub generation_count: usize,
    /// Total frames inserted across all generations (`last_frame - 0`, since
    /// the chain is required to start at frame 1 and be gap-free).
    pub frames_replayed: u64,
    /// The last frame position written into the destination WAL.
    pub last_frame: u64,
    /// R732-F2: the highest fencing epoch contributing to this restore (`0`
    /// for a chain written before fencing existed). Reported so an operator
    /// restoring after an ownership transfer can see which owner's data they
    /// actually got.
    pub epoch: u64,
}

/// A consistency-checked sequence of generation manifests, ready to drive a
/// replay. The fields are the single base / page size / checkpoint sequence
/// shared by every manifest in the chain.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ValidatedChain {
    pub base_snapshot_key: String,
    pub page_size: usize,
    /// R858-B19: the one WAL generation every manifest in the chain agrees on.
    /// `salt` is `Some` for any chain of two or more manifests — a chain that
    /// could not prove a single generation never gets this far.
    pub generation: WalGeneration,
    pub total_frames: u64,
    /// R732-F2: the highest fencing epoch in the chain — i.e. the most recent
    /// owner that contributed frames. Unlike the other fields this is a
    /// *maximum*, not a shared constant: ownership legitimately moves
    /// mid-chain, so a chain may span epochs as long as they never go
    /// backwards.
    pub epoch: u64,
}

/// Validate that a sorted list of generation manifests forms a single,
/// replayable chain: same base snapshot, same page size, single checkpoint
/// sequence, frames starting at 1 and contiguous across manifests.
///
/// V1 refuses to span a WAL restart (multiple `checkpoint_seq` values). A
/// restart implies the source engine folded the WAL into main between
/// generations — replaying the post-restart frames onto our pre-restart base
/// would skip that fold and corrupt the result. The remediation is a fresh
/// tier-1a snapshot, not heroics in restore.
///
/// # R858-B19 — refuse what cannot be proven
///
/// The `checkpoint_seq` test above was the *only* generation check here, and it
/// is blind to the fold that matters: a writer-process restart deletes the
/// `-wal` file and the next writer starts a fresh WAL back at sequence `0`, so
/// two unrelated WALs both report `0` and a spliced chain sailed through. That
/// produced a restore that reported SUCCESS while writing a stale-but-plausible
/// image (probe B: 9 rows against a 12-row source, `integrity_check ok`) or one
/// upstream sqlite3 calls malformed (probe F). **A silent wrong image is the
/// specific failure this function now exists to make impossible.**
///
/// So the salt is checked too, and — the part that matters — a chain that
/// cannot be *shown* to come from one WAL is refused rather than replayed:
///
/// - Two or more manifests, any of them lacking a salt (written before `v4`):
///   REFUSE. The splice is exactly what a pre-R858-B19 writer produced, and
///   nothing in those manifests records which WAL each range came from.
/// - Two or more manifests with disagreeing salts: REFUSE, naming both.
/// - A single manifest: accepted with whatever salt it has, including none.
///   One generation is not a splice; there is nothing to prove.
pub(crate) fn validate_generation_chain(
    manifests: &[OwnedGenerationManifest],
) -> Result<ValidatedChain> {
    let first = manifests
        .first()
        .context("validate_generation_chain: empty manifest list")?;
    let mut expected_next_frame: u64 = 1;
    let mut chain_epoch: u64 = 0;
    let multi = manifests.len() > 1;
    for (i, m) in manifests.iter().enumerate() {
        // R732-F2 (W245): generations are ordered by write time, so a chain
        // whose epoch goes BACKWARDS says a stale owner wrote after the sink
        // had already moved on — precisely the split-brain the fence exists to
        // stop, caught here on the read side too. Non-decreasing is fine and
        // expected: ownership transfers mid-stream and the new owner keeps
        // appending frames to the same contiguous range.
        if m.epoch < chain_epoch {
            anyhow::bail!(
                "generation #{i} was written at epoch {} but an earlier generation in the chain is at epoch {} — a fenced (stale) owner wrote to this sink, restore refuses rather than replay interleaved frames",
                m.epoch,
                chain_epoch,
            );
        }
        chain_epoch = m.epoch;
        if m.base_snapshot_key != first.base_snapshot_key {
            anyhow::bail!(
                "generation #{i} references base {:?}, expected {:?} — chain spans bases, restore needs a fresh tier-1a snapshot",
                m.base_snapshot_key,
                first.base_snapshot_key,
            );
        }
        if m.page_size != first.page_size {
            anyhow::bail!(
                "generation #{i} page_size {} differs from chain page_size {} — corrupt manifest or mixed sinks",
                m.page_size,
                first.page_size,
            );
        }
        if m.checkpoint_seq != first.checkpoint_seq {
            anyhow::bail!(
                "generation #{i} checkpoint_seq {} differs from chain checkpoint_seq {} — WAL restart between generations, restore needs a fresh tier-1a snapshot",
                m.checkpoint_seq,
                first.checkpoint_seq,
            );
        }
        // R858-B19: the check `checkpoint_seq` cannot make. Only enforced on a
        // multi-manifest chain — a lone generation has nothing to be spliced
        // to, so refusing it would strand every pre-v4 single-generation backup
        // for no safety gain.
        if multi {
            let Some(s) = m.salt else {
                anyhow::bail!(
                    "generation #{i} carries no WAL salt (written by a pre-R858-B19 writer) and this chain spans {} generations — which WAL each range came from was never recorded, so a chain spliced across a WAL recreate is indistinguishable from a good one. Restore refuses rather than replay a plausible wrong image; take a fresh tier-1a snapshot",
                    manifests.len(),
                );
            };
            if let Some(fs) = first.salt {
                if s != fs {
                    anyhow::bail!(
                        "generation #{i} was written under WAL salt {s} but the chain starts at salt {fs} (both at checkpoint_seq {}) — the source WAL was RECREATED mid-stream, so these frames belong to two different WALs and replaying them as one would corrupt the image. Restore needs a fresh tier-1a snapshot",
                        first.checkpoint_seq,
                    );
                }
            }
        }
        if m.first_frame != expected_next_frame {
            anyhow::bail!(
                "generation #{i} starts at frame {} but the previous generation ended at frame {} — gap in stream",
                m.first_frame,
                expected_next_frame.saturating_sub(1),
            );
        }
        if m.last_frame < m.first_frame {
            anyhow::bail!(
                "generation #{i} has last_frame {} < first_frame {} — corrupt manifest",
                m.last_frame,
                m.first_frame,
            );
        }
        expected_next_frame = m.last_frame + 1;
    }
    Ok(ValidatedChain {
        base_snapshot_key: first.base_snapshot_key.clone(),
        page_size: first.page_size,
        generation: WalGeneration {
            checkpoint_seq: first.checkpoint_seq,
            salt: first.salt,
        },
        total_frames: expected_next_frame - 1,
        epoch: chain_epoch,
    })
}

/// Fetch one generation's frames and hand each to `on_frame` in ascending
/// frame order, skipping anything before `from_frame`. Returns how many frames
/// were delivered.
///
/// R761-F2: the one read path that resolves a manifest to frame bytes,
/// batched or not — `BackupTarget::frame_objects_of` decides which layout
/// the generation used, and a batch object is split here at
/// `24 + page_size` boundaries. Both restore's replay and
/// [`crate::puller::WalPuller`] call it, so a layout the writer can produce
/// can never be readable by one and not the other.
pub(crate) async fn for_each_frame_in_generation<F>(
    target: &BackupTarget,
    m: &OwnedGenerationManifest,
    from_frame: u64,
    mut on_frame: F,
) -> Result<u64>
where
    F: FnMut(u64, &[u8]) -> Result<()>,
{
    let frame_size = WAL_FRAME_HEADER_SIZE + m.page_size;
    let mut delivered: u64 = 0;
    for (key, first, last) in target.frame_objects_of(m) {
        if last < from_frame {
            continue; // wholly behind the caller's cursor — don't pay for the GET
        }
        let bytes = target
            .store
            .get(&key)
            .await
            .with_context(|| format!("fetching frame object {key}"))?
            .bytes()
            .await
            .with_context(|| format!("reading frame object body {key}"))?;
        let frames_in_object = (last - first + 1) as usize;
        let want = frames_in_object * frame_size;
        if bytes.len() != want {
            anyhow::bail!(
                "frame object {key} is {} bytes, expected {want} ({frames_in_object} frame(s) x [{WAL_FRAME_HEADER_SIZE} header + {} page])",
                bytes.len(),
                m.page_size,
            );
        }
        for (i, frame_no) in (first..=last).enumerate() {
            if frame_no < from_frame {
                continue;
            }
            let at = i * frame_size;
            on_frame(frame_no, &bytes[at..at + frame_size])?;
            delivered += 1;
        }
    }
    Ok(delivered)
}

/// Replay every frame named by `manifests` into `seam`, in (checkpoint_seq,
/// frame_no) order. Caller must have already called `wal_insert_begin` on the
/// seam; `wal_insert_end` is also the caller's responsibility (so a test or a
/// future fault-injection path can choose `force_commit`).
///
/// Each manifest's own `page_size` sizes its frames —
/// [`validate_generation_chain`] has already established they all agree, so
/// there is no separate chain-level page size to thread through.
async fn replay_frames_into<S: WalInsertSeam>(
    target: &BackupTarget,
    seam: &S,
    manifests: &[OwnedGenerationManifest],
) -> Result<u64> {
    let mut total: u64 = 0;
    for m in manifests {
        total += for_each_frame_in_generation(target, m, 0, |frame_no, bytes| {
            seam.wal_insert_frame(frame_no, bytes)
                .with_context(|| format!("inserting frame {frame_no}"))
        })
        .await?;
    }
    Ok(total)
}

/// Restore a database from a streamed backup: download the base tier-1a
/// snapshot referenced by the generation manifests, then replay every uploaded
/// WAL frame onto it via the [`WalInsertSeam`] of a fresh `turso_core`
/// connection. The session ends with `force_commit = false` so any frames
/// captured after the last commit frame are dropped (crash-consistency).
///
/// Errors loudly when:
/// - There are no generation manifests under the prefix (caller should restore
///   via the tier-1a path instead).
/// - The chain spans more than one `checkpoint_seq` or more than one base
///   snapshot (restart / mixed bases — needs a fresh tier-1a snapshot).
/// - Frame ranges are not gap-free starting at 1.
/// - A referenced frame object is missing or the wrong byte length.
///
/// `dest_path`'s `-wal` and `-shm` sidecars are removed before the base is
/// written; any pre-existing turso connection on `dest_path` must be closed
/// by the caller.
///
/// A caller that has already listed the chain for its own reasons should call
/// [`restore_stream_from_manifests`] instead and skip the re-listing this one
/// does — see its docs for what that costs (R760-T18).
pub async fn restore_latest_stream(
    target: &BackupTarget,
    dest_path: &str,
) -> Result<RestoreOutcome> {
    let manifests = list_and_parse_generation_manifests(target).await?;
    restore_stream_from_manifests(target, dest_path, &manifests).await
}

/// [`restore_latest_stream`] for a caller that has already listed and parsed
/// the chain — same replay, minus the listing.
///
/// R760-T18: a caller that must inspect the chain *before* deciding how to
/// restore otherwise pays for every manifest twice. roadcase's `SinkHydrator`
/// is the live example: it calls [`list_and_parse_generation_manifests`] to
/// ask whether the era has any frames at all (an era whose base is still the
/// whole story restores via `snapshot::restore_latest` instead), and then
/// `restore_latest_stream` re-listed and re-fetched the identical objects. A
/// cold start with `n` generations was measured at `3n+1` Class B operations
/// against a `2n+1` floor — one GET per manifest, one per generation's frame
/// batch, one for the base. Handing the parse straight in removes the `n`
/// duplicates.
///
/// `manifests` must be ascending by generation, which is the order
/// [`list_and_parse_generation_manifests`] returns them in; the chain is
/// validated here exactly as it is on the listing path, so a hand-assembled
/// list cannot smuggle past a check.
pub async fn restore_stream_from_manifests(
    target: &BackupTarget,
    dest_path: &str,
    manifests: &[OwnedGenerationManifest],
) -> Result<RestoreOutcome> {
    if manifests.is_empty() {
        anyhow::bail!(
            "no generation manifests under {} — restore tier-1a directly via snapshot::restore_latest",
            join_key(&target.prefix, "generations"),
        );
    }
    let chain = validate_generation_chain(manifests)?;

    // Download the base snapshot and lay it down at dest_path. Strip any stale
    // WAL/-shm sidecars first — the base alone is the entire pre-replay image
    // (VACUUM INTO output has no WAL).
    let base_bytes = target
        .store
        .get(&ObjPath::from(chain.base_snapshot_key.clone()))
        .await
        .with_context(|| format!("downloading base snapshot {}", chain.base_snapshot_key))?
        .bytes()
        .await
        .with_context(|| format!("reading base snapshot body {}", chain.base_snapshot_key))?;
    for sfx in ["-wal", "-shm"] {
        let _ = std::fs::remove_file(format!("{dest_path}{sfx}"));
    }
    std::fs::write(dest_path, &base_bytes)
        .with_context(|| format!("writing restored base snapshot to {dest_path}"))?;

    // Open a fresh seam on the laid-down base and replay. `CoreWalSeam::open`
    // disables auto-actions; `wal_insert_begin` further locks the session to
    // empty auto-actions for the txn, so restart can't race the replay.
    let seam = CoreWalSeam::open(dest_path)?;
    seam.wal_insert_begin()
        .context("starting WAL insert session on restore destination")?;
    let frames = match replay_frames_into(target, &seam, manifests).await {
        Ok(n) => n,
        Err(e) => {
            // Best-effort: roll the partial replay back so we never leave the
            // dest's WAL with an uncommitted suffix.
            let _ = seam.wal_insert_end(false);
            return Err(e);
        }
    };
    // force_commit=false: the engine drops any tail past the last commit
    // frame, which is exactly the crash-consistency story we want — a
    // mid-transaction tail captured by tail_frames gets truncated cleanly.
    seam.wal_insert_end(false)
        .context("closing WAL insert session on restore destination")?;

    Ok(RestoreOutcome {
        base_snapshot_key: chain.base_snapshot_key,
        checkpoint_seq: chain.generation.checkpoint_seq,
        generation_count: manifests.len(),
        frames_replayed: frames,
        last_frame: chain.total_frames,
        epoch: chain.epoch,
    })
}

/// Every generation manifest at a sink, in key (i.e. chronological) order.
///
/// **The GETs are serialized**, one manifest at a time, and so is the frame
/// replay in [`for_each_frame_in_generation`] — so a restore's wall clock is
/// `(2n+1) x RTT` plus transfer for `n` generations, not `max(RTT)`. That is
/// fine at the tens of generations a dedicated streamer accumulates between
/// checkpoints and is not fine at thousands; R760-T18 bounds it on the
/// *writer* side (roadcase caps generations per era) rather than by making
/// this concurrent, because concurrency here would mean a real
/// `futures`/`tokio::spawn` dependency in a crate that deliberately keeps
/// `futures_util` to `[dev-dependencies]`.
///
/// Public since R732-T5: a split-brain reconciliation oracle needs to ask what
/// a sink would actually replay — which owner wrote which frame range, under
/// which epoch — without running a full restore. Pair it with
/// [`validate_generation_chain`]'s public counterpart if you need the chain
/// checked rather than merely listed.
pub async fn list_and_parse_generation_manifests(
    target: &BackupTarget,
) -> Result<Vec<OwnedGenerationManifest>> {
    let prefix = join_key(&target.prefix, "generations");
    let listing = target
        .store
        .list_with_delimiter(Some(&prefix))
        .await
        .with_context(|| format!("listing generation manifests under {prefix}"))?;
    let mut keys: Vec<ObjPath> = listing.objects.into_iter().map(|o| o.location).collect();
    keys.sort();
    let mut manifests = Vec::with_capacity(keys.len());
    for key in &keys {
        let bytes = target
            .store
            .get(key)
            .await
            .with_context(|| format!("fetching generation manifest {key}"))?
            .bytes()
            .await
            .with_context(|| format!("reading generation manifest body {key}"))?;
        let m = parse_generation_manifest(&String::from_utf8_lossy(&bytes))
            .with_context(|| format!("parsing generation manifest {key}"))?;
        manifests.push(m);
    }
    Ok(manifests)
}

/// What the sink's fence currently stands at — the two comparands
/// [`tail_frames`] checks a writer against, and nothing else.
///
/// R869: this is the off-fleet copy of a fencing token. It matters because
/// `epoch` is minted by a raft group, and a raft group can be *destroyed* —
/// wipe the raft dir, re-form, and the fresh cluster's first
/// `ClaimTenant` grants epoch 1 while this sidecar still says 5, so the
/// rebuilt cluster is fenced out of its own sink. Reading the fence back is how
/// a rebuild starts above the number its dead predecessor left here.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct FenceState {
    /// The fencing epoch of the writer that last advanced the sidecar. `0`
    /// means "unfenced" — either nothing has claimed this sink or it predates
    /// R732-F2 — and fences nobody.
    pub epoch: u64,
    /// The pointer generation of that same writer. `0` means unfenced, exactly
    /// as for `epoch`.
    pub pointer_generation: u64,
}

/// Read the fence [`tail_frames`] would check a writer against, without writing
/// anything or needing a WAL.
///
/// `Ok(None)` means the sidecar does not exist: nothing has ever streamed to
/// this prefix, so **there is no fence here at all** and any writer is
/// accepted. Do not collapse that into `FenceState::default()` at a call site
/// that is deciding whether to fence — "unfenced because nobody has written"
/// and "unfenced because an old writer stamped 0" are the same *value* but a
/// caller that cares about the difference (a rebuild deciding whether a tenant
/// was ever live) needs the `Option`.
///
/// **A floor derived from `epoch` is a lower bound, not an upper one.** It is
/// the highest epoch anyone has *written under*, which is not the highest a
/// dead raft group *granted* — a tenant claimed twice while idle leaves a node
/// holding an epoch strictly above anything this sidecar ever saw. So a rebuild
/// that seeds from this number closes availability (its own writes are
/// accepted) and does **not** on its own fence a resurrected node; that takes
/// `pointer_generation`, which is minted off-fleet where a dead cluster cannot
/// reach it. `oss/yubaba/crates/yubaba/tests/raft_rebuild_fencing.rs` drives
/// both halves against this code.
pub async fn read_fence_state(target: &BackupTarget) -> Result<Option<FenceState>> {
    Ok(read_watermark(&target.store, &target.watermark_key())
        .await?
        .map(|p| FenceState {
            epoch: p.epoch,
            pointer_generation: p.pointer_generation,
        }))
}

/// In-memory shape of a generation manifest. Format on disk:
///
/// ```text
/// TURSO-BACKUP STREAM v4
/// base_snapshot <key>
/// page_size <n>
/// checkpoint_seq <n>
/// wal_salt <salt1>-<salt2>
/// first_frame <n>
/// last_frame <n>
/// epoch <n>
/// owner <label>        (optional)
/// frame_batch <first>-<last>   (one per batch object, ascending, gap-free)
/// ```
///
/// Text, dependency-free (no serde), one field per line. Mirrors the
/// `dedup::Manifest` convention so the crate stays consistent.
///
/// R732-F2 added `epoch`/`owner` and moved the header to `v2`. R761-F2 added
/// the `frame_batch` list and moved it to `v3`. R858-B19 added `wal_salt` and
/// moved it to `v4`. Older manifests still parse — `v1` with `epoch = 0`,
/// `owner = None`; `v1`/`v2` with no batch list, which is what marks their
/// frames as living one-per-object; `v1`/`v2`/`v3` with `salt = None` — so
/// backups written before any of those changes stay *parseable*. No older
/// version is ever written any more.
///
/// R858-B19, and this is the one place a legacy manifest is not merely
/// second-class: a chain of two or more manifests where any of them lacks a
/// salt is **refused** by [`validate_generation_chain`], because a pre-R858-B19
/// writer could splice two WAL generations into a contiguous-looking chain and
/// nothing in the manifest records which WAL each range came from. A
/// single-manifest chain is still restored — one generation cannot be a splice.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GenerationManifest<'a> {
    pub base_snapshot_key: &'a str,
    pub page_size: usize,
    pub checkpoint_seq: u32,
    /// R858-B19: the WAL header salt these frames were read under — the field
    /// that actually names the generation, since `checkpoint_seq` resets to 0
    /// across a writer restart. `None` only when re-formatting a pre-R858-B19
    /// manifest; [`tail_frames`] always has one.
    pub salt: Option<WalSalt>,
    pub first_frame: u64,
    pub last_frame: u64,
    /// R732-F2: the fencing epoch this generation was written under. `0` means
    /// it predates fencing (a `v1` manifest), which is also the key shape its
    /// frames live under — see [`BackupTarget::frame_key`].
    pub epoch: u64,
    /// R732-F2: opaque owner label, diagnostic only. See [`StreamConfig::owner`].
    pub owner: Option<&'a str>,
    /// R761-F2: the `(first, last)` frame range of each batch object holding
    /// this generation's frames, ascending and exactly tiling
    /// `first_frame..=last_frame`. Empty means the pre-batching layout (one
    /// object per frame); a writer emits it empty only when re-formatting a
    /// legacy manifest.
    pub frame_batches: &'a [(u64, u64)],
}

/// Header of a manifest written by a fencing-aware writer (R732-F2). The
/// version is bumped rather than the `epoch` key just being added, because
/// [`parse_generation_manifest`] rejects unknown keys: a pre-R732 binary
/// reading a fenced manifest would otherwise fail with `unknown manifest key:
/// epoch`, which reads like corruption. Failing on the *header* says the real
/// thing — this backup was written by a newer writer.
const MANIFEST_HEADER_V2: &str = "TURSO-BACKUP STREAM v2";
/// Pre-fencing header. Still accepted on read (those backups must stay
/// restorable) and parses with `epoch = 0`, `owner = None`; never written.
const MANIFEST_HEADER_V1: &str = "TURSO-BACKUP STREAM v1";
/// R761-F2 — carries the `frame_batch` list. Bumped for the same reason
/// `v2` was: [`parse_generation_manifest`] rejects unknown keys, so a
/// pre-R761-F2 binary reading a batched manifest would fail with `unknown
/// manifest key: frame_batch`, which reads like corruption. Failing on the
/// header says the true thing — a newer writer wrote this sink. The refusal
/// matters more here than it did for fencing: an older reader that somehow
/// skipped the batch list would look for per-frame keys that do not exist.
const MANIFEST_HEADER_V3: &str = "TURSO-BACKUP STREAM v3";
/// R858-B19 — carries `wal_salt`, the field that makes a generation chain
/// checkable across a WAL recreate. Bumped for the same reason `v2` and `v3`
/// were: [`parse_generation_manifest`] rejects unknown keys, so an older binary
/// reading one of these would fail with `unknown manifest key: wal_salt`, which
/// reads like corruption. Failing on the header says the true thing.
const MANIFEST_HEADER_V4: &str = "TURSO-BACKUP STREAM v4";

pub(crate) fn format_generation_manifest(m: GenerationManifest<'_>) -> String {
    // Each header promises the fields that version introduced, so stamp the
    // newest version whose promises this manifest can actually keep. The only
    // way to reach anything but v4 is re-formatting a legacy manifest (a test,
    // or a repair tool); tail_frames always has both a salt and batches.
    // A v4 manifest promises BOTH the salt and the batch list, so a legacy
    // shape missing either falls back rather than stamping a version whose
    // promises it cannot keep. `salt` without batches has no representation and
    // cannot occur: only tail_frames produces a salt, and it always batches.
    let salt = if m.frame_batches.is_empty() { None } else { m.salt };
    let header = if salt.is_some() {
        MANIFEST_HEADER_V4
    } else if m.frame_batches.is_empty() {
        MANIFEST_HEADER_V2
    } else {
        MANIFEST_HEADER_V3
    };
    let mut out = format!(
        "{}\nbase_snapshot {}\npage_size {}\ncheckpoint_seq {}\n",
        header, m.base_snapshot_key, m.page_size, m.checkpoint_seq,
    );
    if let Some(s) = salt {
        out.push_str(&format!("wal_salt {}-{}\n", s.salt1, s.salt2));
    }
    out.push_str(&format!(
        "first_frame {}\nlast_frame {}\nepoch {}\n",
        m.first_frame, m.last_frame, m.epoch,
    ));
    // Omitted rather than written empty: the parser splits on the first space,
    // so `owner ` with no value would round-trip to `Some("")`.
    if let Some(owner) = m.owner {
        out.push_str(&format!("owner {owner}\n"));
    }
    for (first, last) in m.frame_batches {
        out.push_str(&format!("frame_batch {first}-{last}\n"));
    }
    out
}

/// Parse a generation manifest. Tolerant to trailing whitespace; rejects any
/// other shape (so a corrupt manifest fails loudly during restore, doesn't
/// silently degrade).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OwnedGenerationManifest {
    pub base_snapshot_key: String,
    pub page_size: usize,
    pub checkpoint_seq: u32,
    /// R858-B19: `None` for a pre-`v4` manifest, which means the WAL
    /// generation behind these frames was never recorded and cannot be
    /// recovered. Unknown, not "the same as its neighbour" — see
    /// [`validate_generation_chain`].
    pub salt: Option<WalSalt>,
    pub first_frame: u64,
    pub last_frame: u64,
    /// R732-F2: `0` for a `v1` (pre-fencing) manifest.
    pub epoch: u64,
    /// R732-F2: diagnostic only; `None` when the writer did not label itself.
    pub owner: Option<String>,
    /// R761-F2: the batch objects covering `first_frame..=last_frame`,
    /// ascending and gap-free (enforced by [`parse_generation_manifest`]).
    /// **Empty means the pre-batching layout** — one object per frame — which
    /// is how a `v1`/`v2` manifest keeps restoring. Resolve it to keys with
    /// `BackupTarget::frame_objects_of` (crate-private) rather than branching
    /// at each call site.
    pub frame_batches: Vec<(u64, u64)>,
}

pub fn parse_generation_manifest(text: &str) -> Result<OwnedGenerationManifest> {
    let mut lines = text.lines();
    let header = lines.next().context("empty generation manifest")?.trim();
    let (v2, v3, v4) = match header {
        MANIFEST_HEADER_V4 => (true, true, true),
        MANIFEST_HEADER_V3 => (true, true, false),
        MANIFEST_HEADER_V2 => (true, false, false),
        MANIFEST_HEADER_V1 => (false, false, false),
        other => anyhow::bail!("unexpected manifest header: {other:?}"),
    };
    let mut base_snapshot_key: Option<String> = None;
    let mut page_size: Option<usize> = None;
    let mut checkpoint_seq: Option<u32> = None;
    let mut salt: Option<WalSalt> = None;
    let mut first_frame: Option<u64> = None;
    let mut last_frame: Option<u64> = None;
    let mut epoch: Option<u64> = None;
    let mut owner: Option<String> = None;
    let mut frame_batches: Vec<(u64, u64)> = Vec::new();
    for line in lines {
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        let (k, v) = line
            .split_once(' ')
            .with_context(|| format!("malformed manifest line: {line:?}"))?;
        match k {
            "base_snapshot" => base_snapshot_key = Some(v.to_string()),
            "page_size" => page_size = Some(v.parse().context("page_size")?),
            "checkpoint_seq" => checkpoint_seq = Some(v.parse().context("checkpoint_seq")?),
            "wal_salt" => {
                let (s1, s2) = v
                    .split_once('-')
                    .with_context(|| format!("malformed wal_salt pair: {v:?}"))?;
                salt = Some(WalSalt {
                    salt1: s1.trim().parse().context("wal_salt salt1")?,
                    salt2: s2.trim().parse().context("wal_salt salt2")?,
                });
            }
            "first_frame" => first_frame = Some(v.parse().context("first_frame")?),
            "last_frame" => last_frame = Some(v.parse().context("last_frame")?),
            "epoch" => epoch = Some(v.parse().context("epoch")?),
            "owner" => owner = Some(v.to_string()),
            "frame_batch" => {
                let (first, last) = v
                    .split_once('-')
                    .with_context(|| format!("malformed frame_batch range: {v:?}"))?;
                frame_batches.push((
                    first.trim().parse().context("frame_batch first")?,
                    last.trim().parse().context("frame_batch last")?,
                ));
            }
            other => anyhow::bail!("unknown manifest key: {other}"),
        }
    }
    // A v2 manifest without an epoch is corrupt, not legacy — the writer that
    // stamped the v2 header always writes one. Defaulting it to 0 would
    // silently demote a fenced generation to unfenced, which is the one
    // direction this whole mechanism must never fail in.
    if v2 && epoch.is_none() {
        anyhow::bail!("{MANIFEST_HEADER_V2} manifest is missing `epoch`");
    }
    // R858-B19: same reasoning as the `epoch` check above. A v4 writer always
    // stamps the salt, so a v4 manifest without one is corrupt, not legacy —
    // and defaulting it to `None` would silently demote a checkable generation
    // to an unknown one, which is the direction this mechanism must never fail
    // in. The converse guard matters just as much: a pre-v4 header carrying a
    // `wal_salt` line is hand-edited or truncated, and trusting that salt would
    // let a forged line wave a spliced chain through.
    if v4 && salt.is_none() {
        anyhow::bail!("{MANIFEST_HEADER_V4} manifest is missing `wal_salt`");
    }
    if !v4 && salt.is_some() {
        anyhow::bail!(
            "manifest header {header:?} carries a `wal_salt` line — the WAL salt is a {MANIFEST_HEADER_V4} field, so this manifest is corrupt or hand-edited"
        );
    }
    let first_frame = first_frame.context("missing first_frame")?;
    let last_frame = last_frame.context("missing last_frame")?;
    // R761-F2: the batch list IS the frame index, so a v3 manifest that does
    // not tile its own range exactly would send restore looking for objects
    // that were never written — caught here, at parse, rather than as a 404
    // halfway through a replay.
    if v3 {
        anyhow::ensure!(
            !frame_batches.is_empty(),
            "{MANIFEST_HEADER_V3} manifest has no `frame_batch` lines — it cannot say where its frames are"
        );
        let mut expected = first_frame;
        for &(first, last) in &frame_batches {
            anyhow::ensure!(
                first == expected && last >= first,
                "frame_batch {first}-{last} does not continue the range at frame {expected}"
            );
            expected = last + 1;
        }
        anyhow::ensure!(
            expected == last_frame + 1,
            "frame_batch list covers frames {first_frame}..={} but the manifest claims {first_frame}..={last_frame}",
            expected - 1,
        );
    } else if !frame_batches.is_empty() {
        anyhow::bail!(
            "manifest header {header:?} carries `frame_batch` lines — batching is a {MANIFEST_HEADER_V3} feature, so this manifest is corrupt or hand-edited"
        );
    }
    Ok(OwnedGenerationManifest {
        base_snapshot_key: base_snapshot_key.context("missing base_snapshot")?,
        page_size: page_size.context("missing page_size")?,
        checkpoint_seq: checkpoint_seq.context("missing checkpoint_seq")?,
        salt,
        first_frame,
        last_frame,
        epoch: epoch.unwrap_or(0),
        owner,
        frame_batches,
    })
}

/// A watermark as read back from the sidecar: the engine position plus the
/// wall-clock instant the sidecar was last written (R574-T4 — this is what
/// [`RpoStatus::watermark_age`] measures against). `written_at_nanos` is
/// `None` for sidecars written before the timestamp field existed.
#[derive(Debug, Clone, PartialEq, Eq)]
struct PersistedWatermark {
    watermark: Watermark,
    /// R858-B19: the WAL generation `watermark.last_frame` is a position
    /// *within*. `generation.salt == None` marks a sidecar written before the
    /// salt fields existed — unknown, therefore never provably equal to the
    /// live WAL, therefore a forced restart on the next tail. That is the
    /// deliberate choice: one redundant re-upload beats resuming into a WAL
    /// nobody can show is the same one.
    generation: WalGeneration,
    written_at_nanos: Option<u128>,
    /// R732-F2: the fencing epoch of the writer that last advanced this
    /// sidecar. `0` for a sidecar written before the field existed, which
    /// reads as "unfenced" and therefore fences nobody — the same
    /// behaviour-preserving default as [`StreamConfig::epoch`].
    ///
    /// This is the value [`tail_frames`] compares against, so it is the single
    /// piece of state the whole fence rests on. R732-T3 makes advancing it a
    /// compare-and-swap (R732-T3), so the *concurrent* hole is closed too:
    /// two writers racing the read-modify-write cannot both win, because the
    /// loser's conditional put fails against the version the winner replaced.
    epoch: u64,
    /// R736-T2: the pointer generation of the writer that last advanced this
    /// sidecar. `0` for a sidecar written before the field existed, which
    /// reads as "unfenced" and therefore fences nobody — the same
    /// behaviour-preserving default as [`StreamConfig::pointer_generation`].
    /// Checked alongside `epoch` in [`tail_frames`]; either being stale
    /// bounces the writer.
    pointer_generation: u64,
    /// R732-T3: the object version this record was read at, carried so the
    /// next advance can be conditional on it. `None` only when the sidecar
    /// does not exist yet, which selects [`PutMode::Create`] instead of
    /// [`PutMode::Update`] — "I believe nobody owns this sink" is as much a
    /// precondition as "I believe it is still at version V".
    version: Option<UpdateVersion>,
}

/// Sidecar format:
/// `<checkpoint_seq> <last_frame> [<written_at_unix_nanos> [<epoch>
/// [<pointer_generation> [<salt1> <salt2>]]]]`.
///
/// The third field was added by R574-T4, the fourth by R732-F2, the fifth by
/// R736-T2, and the sixth and seventh by R858-B19; all are positional appends,
/// which is what keeps this readable in both directions. A shorter sidecar (an
/// older writer) still parses, with the missing tail defaulting to `None` /
/// `0`, and an older reader stops after its last known field and never sees the
/// newer ones. Fields are only ever appended for exactly that reason — do not
/// reorder them.
///
/// R858-B19: a missing salt pair parses to `WalGeneration { salt: None }`,
/// which is *unknown*, not *unchanged*. `tail_frames` restarts against it
/// rather than resuming — see [`PersistedWatermark::generation`]. The two
/// fields are read as a pair: a sidecar carrying only one of them is corrupt
/// (this writer emits both or neither) and is rejected rather than
/// half-trusted.
async fn read_watermark(
    store: &Arc<dyn ObjectStore>,
    key: &ObjPath,
) -> Result<Option<PersistedWatermark>> {
    match store.get(key).await {
        Ok(res) => {
            // Capture the version BEFORE consuming the body — `bytes()` takes
            // `res` by value. Both fields are kept because stores differ in
            // which one they honour for a conditional put (object_store's own
            // `UpdateVersion` docs say to preserve both).
            let version = UpdateVersion {
                e_tag: res.meta.e_tag.clone(),
                version: res.meta.version.clone(),
            };
            let bytes = res.bytes().await.context("reading watermark sidecar")?;
            let s = String::from_utf8_lossy(&bytes);
            let mut fields = s.split_whitespace();
            let seq = fields
                .next()
                .context("watermark sidecar must be '<checkpoint_seq> <last_frame> [<nanos>]'")?;
            let frame = fields
                .next()
                .context("watermark sidecar must be '<checkpoint_seq> <last_frame> [<nanos>]'")?;
            let written_at_nanos = fields
                .next()
                .map(|n| n.parse::<u128>().context("watermark written_at nanos"))
                .transpose()?;
            let epoch = fields
                .next()
                .map(|e| e.parse::<u64>().context("watermark epoch"))
                .transpose()?
                .unwrap_or(0);
            let pointer_generation = fields
                .next()
                .map(|g| g.parse::<u64>().context("watermark pointer_generation"))
                .transpose()?
                .unwrap_or(0);
            // R858-B19: both salt words or neither. A lone `salt1` is not a
            // half-known generation, it is a torn write — say so instead of
            // silently downgrading it to "unknown" and papering over it.
            let salt = match (fields.next(), fields.next()) {
                (Some(s1), Some(s2)) => Some(WalSalt {
                    salt1: s1.parse().context("watermark salt1")?,
                    salt2: s2.parse().context("watermark salt2")?,
                }),
                (None, _) => None,
                (Some(_), None) => anyhow::bail!(
                    "watermark sidecar carries salt1 but no salt2 — truncated or hand-edited; \
                     refusing rather than guessing which WAL generation it names"
                ),
            };
            let checkpoint_seq: u32 = seq.parse().context("watermark checkpoint_seq")?;
            Ok(Some(PersistedWatermark {
                watermark: Watermark {
                    checkpoint_seq,
                    last_frame: frame.parse().context("watermark last_frame")?,
                },
                generation: WalGeneration { checkpoint_seq, salt },
                written_at_nanos,
                epoch,
                pointer_generation,
                version: Some(version),
            }))
        }
        Err(object_store::Error::NotFound { .. }) => Ok(None),
        Err(e) => Err(e).context("fetching watermark sidecar"),
    }
}

/// R732-T3: result of a conditional watermark advance.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WatermarkCas {
    /// We held the version we read, and the sidecar now names us.
    Advanced,
    /// Somebody else replaced the sidecar between our read and our write.
    /// Says nothing about *who* — the caller re-reads to find out whether it
    /// was a newer owner (we are fenced) or a same-epoch racer (a bug).
    Contended,
}

/// Advance the watermark sidecar **conditionally** on the version it was read
/// at (R732-T3 / W245).
///
/// This is the atomic half of the fence. The epoch check in [`tail_frames`]
/// stops a *sequential* stale owner — one that returns after a transfer and
/// reads a sidecar already stamped higher. It cannot stop a *concurrent* one:
/// two writers that both read the old sidecar in the same instant would both
/// pass that check and then both blindly overwrite, last write winning, which
/// is exactly the corruption W245 describes. Making the advance a
/// compare-and-swap removes the window — at most one of them holds the version
/// the other replaced.
///
/// No new CAS primitive was needed: `object_store` already models this as
/// [`PutMode::Update`] (→ [`object_store::Error::Precondition`]) and
/// [`PutMode::Create`] (→ `AlreadyExists`) for the first write, and R2 honours
/// the underlying `If-Match` / `If-None-Match`.
///
/// **Deployment caveat, not a code path:** `AmazonS3Builder` must be
/// configured for conditional puts against a store that supports them. If the
/// backend silently degrades to unconditional writes, this returns `Advanced`
/// unconditionally and the concurrent window reopens — the sequential fence in
/// `tail_frames` still holds, but the race does not.
async fn write_watermark(
    store: &Arc<dyn ObjectStore>,
    key: &ObjPath,
    w: Watermark,
    salt: Option<WalSalt>,
    epoch: u64,
    pointer_generation: u64,
    expected: Option<&UpdateVersion>,
) -> Result<WatermarkCas> {
    let mut body = format!(
        "{} {} {} {} {}",
        w.checkpoint_seq,
        w.last_frame,
        unix_nanos(),
        epoch,
        pointer_generation
    );
    // R858-B19: omitted entirely rather than written as a sentinel, so an
    // unknown generation is one shape (absent) on both the read and write
    // sides, and no magic value can ever be mistaken for a real salt.
    if let Some(s) = salt {
        body.push_str(&format!(" {} {}", s.salt1, s.salt2));
    }
    body.push('\n');
    let opts = PutOptions {
        mode: match expected {
            Some(v) => PutMode::Update(v.clone()),
            // No sidecar when we read: assert that is *still* true, so two
            // writers bootstrapping the same fresh sink cannot both proceed.
            None => PutMode::Create,
        },
        ..Default::default()
    };
    match store.put_opts(key, body.into_bytes().into(), opts).await {
        Ok(_) => Ok(WatermarkCas::Advanced),
        Err(object_store::Error::Precondition { .. })
        | Err(object_store::Error::AlreadyExists { .. }) => Ok(WatermarkCas::Contended),
        Err(e) => Err(e).with_context(|| format!("writing watermark sidecar {key}")),
    }
}

/// Which conditional-put mode a preflight probe found unenforced.
///
/// Both matter and they fail independently — a store can honour `If-None-Match`
/// (guarding the bootstrap put) while ignoring `If-Match` (guarding every
/// steady-state advance), or the reverse. Naming which one degraded is the
/// difference between an operator fixing a bucket setting in a minute and
/// bisecting a corruption in a week.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PreflightStage {
    /// [`PutMode::Create`] / `If-None-Match` — the guard on two writers
    /// bootstrapping the same fresh sink.
    Create,
    /// [`PutMode::Update`] / `If-Match` — the guard on every subsequent
    /// watermark advance, and therefore the one the steady state rests on.
    Update,
}

impl PreflightStage {
    pub fn as_str(&self) -> &'static str {
        match self {
            PreflightStage::Create => "PutMode::Create (If-None-Match)",
            PreflightStage::Update => "PutMode::Update (If-Match)",
        }
    }
}

/// What [`probe_conditional_puts`] observed at a real sink.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PreconditionSupport {
    /// Both conditional modes were enforced — a put that should have been
    /// rejected was rejected. The watermark CAS is real at this sink.
    Honoured,
    /// A put that *must* have failed its precondition succeeded instead, so
    /// this backend has silently degraded to unconditional writes.
    Degraded { stage: PreflightStage },
}

impl PreconditionSupport {
    pub fn is_honoured(&self) -> bool {
        matches!(self, PreconditionSupport::Honoured)
    }
}

/// R732-T4: prove, against the *real* configured sink, that conditional puts
/// are actually enforced — and hand the caller a hard answer so it can refuse
/// to start when they are not.
///
/// [`write_watermark`] carries a deployment caveat that no test can close:
/// if `AmazonS3Builder` is pointed at a store that does not honour
/// `If-Match` / `If-None-Match`, every conditional put silently succeeds, the
/// CAS degrades to last-write-wins, and the concurrent split-brain window
/// W245 exists to shut reopens — while every unit test stays green, because
/// the in-memory store used in tests does honour them. Support is a property
/// of configuration, not of code, so a runtime probe is the only guard that
/// can exist.
///
/// The probe writes a uniquely-keyed canary under `<prefix>/preflight/`,
/// exercises both modes with puts that MUST be rejected, and deletes it. It
/// touches no watermark, no manifest and no frame, so it is safe to run at
/// startup against a live sink another node owns — and it is deliberately
/// keyed per-process-per-nanosecond so two nodes probing at once cannot fail
/// each other.
///
/// `Ok(Degraded)` is the interesting return and is NOT an error: the probe
/// worked perfectly, and reported that the store is unsafe. An `Err` means the
/// probe could not reach a verdict at all (sink unreachable, credentials
/// wrong), which is also a refuse-to-start condition but a different one for
/// an operator to read.
pub async fn probe_conditional_puts(target: &BackupTarget) -> Result<PreconditionSupport> {
    let key = join_key(
        &target.prefix,
        &format!("preflight/conditional-put-{:020}-{}.canary", unix_nanos(), std::process::id()),
    );
    let result = probe_at_key(&target.store, &key).await;
    // Best-effort cleanup: a leaked canary is inert (nothing reads
    // `preflight/`), so a delete failure must not mask the verdict — which is
    // the whole reason the caller ran this.
    if let Err(e) = target.store.delete(&key).await {
        tracing_delete_failure(&key, &e);
    }
    result
}

/// The probe body, split out so the canary is deleted on every path.
async fn probe_at_key(
    store: &Arc<dyn ObjectStore>,
    key: &ObjPath,
) -> Result<PreconditionSupport> {
    let create = PutOptions { mode: PutMode::Create, ..Default::default() };

    // 1. Claim the key. This one is *expected* to succeed; if it doesn't, the
    //    probe cannot reach a verdict (and `AlreadyExists` here means the
    //    per-process-per-nanosecond key collided, which is a bug, not a store
    //    property — so it stays an error rather than a `Degraded` verdict).
    let first = store
        .put_opts(key, b"preflight-1".as_slice().into(), create.clone())
        .await
        .with_context(|| format!("preflight canary could not be created at {key}"))?;
    let v1 = UpdateVersion { e_tag: first.e_tag.clone(), version: first.version.clone() };

    // 2. Create again over a key that now exists. A store honouring
    //    `If-None-Match` rejects this; one that succeeds has degraded.
    match store.put_opts(key, b"preflight-2".as_slice().into(), create).await {
        Err(object_store::Error::AlreadyExists { .. })
        | Err(object_store::Error::Precondition { .. }) => {}
        Ok(_) => return Ok(PreconditionSupport::Degraded { stage: PreflightStage::Create }),
        Err(e) => {
            return Err(e).with_context(|| format!("preflight Create probe failed at {key}"))
        }
    }

    // 3. Advance the canary conditionally on the version we hold, to obtain a
    //    *superseded* version. Expected to succeed — it is the ordinary
    //    steady-state write `write_watermark` makes.
    store
        .put_opts(
            key,
            b"preflight-3".as_slice().into(),
            PutOptions { mode: PutMode::Update(v1.clone()), ..Default::default() },
        )
        .await
        .with_context(|| format!("preflight Update probe could not advance {key}"))?;

    // 4. Update again on the now-stale version — exactly the shape of a fenced
    //    writer losing the watermark race. A store honouring `If-Match`
    //    rejects it.
    match store
        .put_opts(
            key,
            b"preflight-4".as_slice().into(),
            PutOptions { mode: PutMode::Update(v1), ..Default::default() },
        )
        .await
    {
        Err(object_store::Error::Precondition { .. })
        | Err(object_store::Error::AlreadyExists { .. }) => Ok(PreconditionSupport::Honoured),
        Ok(_) => Ok(PreconditionSupport::Degraded { stage: PreflightStage::Update }),
        Err(e) => Err(e).with_context(|| format!("preflight Update probe failed at {key}")),
    }
}

/// turso-backup takes no logging dependency (it is a library consumed by
/// binaries that pick their own), so a failed canary cleanup goes to stderr
/// rather than through `tracing`.
fn tracing_delete_failure(key: &ObjPath, e: &object_store::Error) {
    eprintln!("turso-backup: preflight canary {key} could not be deleted: {e}");
}

/// R574-T4: compute the watermark-staleness snapshot for this call. Pure so
/// the breach edge cases are unit-testable without staging a sidecar.
fn rpo_status(target: Option<Duration>, prior: Option<&PersistedWatermark>) -> RpoStatus {
    let watermark_age = prior
        .and_then(|p| p.written_at_nanos)
        .and_then(|written_at| unix_nanos().checked_sub(written_at))
        .map(|nanos| Duration::from_nanos(u64::try_from(nanos).unwrap_or(u64::MAX)));
    let breached = matches!((target, watermark_age), (Some(t), Some(age)) if age > t);
    RpoStatus { target, watermark_age, breached }
}

fn join_key(prefix: &str, leaf: &str) -> ObjPath {
    let prefix = prefix.trim_matches('/');
    if prefix.is_empty() {
        ObjPath::from(leaf)
    } else {
        ObjPath::from(format!("{prefix}/{leaf}"))
    }
}

fn unix_nanos() -> u128 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0)
}

/// Default [`StreamGcConfig::grace`] — 24 hours.
///
/// Conservative on purpose. The window has to cover the longest gap between an
/// object becoming unreachable-looking and a live reader or writer finishing
/// with it, and there are three such gaps, the largest of which is not bounded
/// by anything this crate controls:
///
/// 1. **A restore in flight.** [`restore_stream_from_manifests`] lists the
///    chain, then fetches the base, then walks every frame object one at a
///    time with the GETs serialized (see
///    [`list_and_parse_generation_manifests`] on why). A cold restore of a
///    large database over a slow link is minutes to hours, and it is reading
///    objects it named *before* the GC listed anything.
/// 2. **A tail mid-upload.** [`tail_frames`] uploads frame batches and only
///    then writes the generation manifest naming them, so between those two
///    points a perfectly live batch is referenced by no manifest and looks
///    exactly like an orphan.
/// 3. **A rebase mid-flight.** `crate::tail::rebase` publishes the new base
///    *before* deleting the old generation manifests, so in that window the
///    surviving manifests name the OLD base and the new one is reachable from
///    nothing. The grace window is the only thing that keeps a concurrent GC
///    from deleting the base a recovery just published.
///
/// 24 hours also swallows clock skew between the object store's
/// `last_modified` and this process's wall clock, which is what the age
/// comparison is made against. A day of retained garbage costs storage; an
/// hour too few costs a restore.
pub const DEFAULT_STREAM_GC_GRACE: Duration = Duration::from_secs(24 * 60 * 60);

/// Knobs for [`gc_stream`].
///
/// [`Default`] is `{ grace: DEFAULT_STREAM_GC_GRACE, dry_run: true }` — the
/// posture for something a human points at a live bucket.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StreamGcConfig {
    /// Never collect an object younger than this. See
    /// [`DEFAULT_STREAM_GC_GRACE`].
    pub grace: Duration,
    /// Report what would be collected and delete nothing. **Defaults to
    /// `true`.**
    pub dry_run: bool,
}

impl Default for StreamGcConfig {
    fn default() -> Self {
        Self {
            grace: DEFAULT_STREAM_GC_GRACE,
            dry_run: true,
        }
    }
}

/// What a [`gc_stream`] pass found — and, when `dry_run` was false, deleted.
///
/// The collected keys are listed rather than counted because the primary
/// consumer is a human reading a dry run before authorizing the real one.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct StreamGcOutcome {
    /// Echo of [`StreamGcConfig::dry_run`]. `true` means nothing was deleted
    /// and `collected_*` is a proposal.
    pub dry_run: bool,
    /// Every snapshot key a restore could still reach, sorted. See
    /// [`gc_stream`]'s "Liveness" section for how this is derived.
    pub live_base_snapshot_keys: Vec<String>,
    /// Superseded base snapshots, sorted.
    pub collected_snapshots: Vec<String>,
    /// Orphaned frame objects (batch or legacy per-frame), sorted.
    pub collected_frame_objects: Vec<String>,
    /// Total bytes across `collected_snapshots` + `collected_frame_objects`.
    pub collected_bytes: u64,
    /// Snapshot objects kept — because they are live, or because the whole
    /// snapshot half was skipped (see `base_snapshots_skipped`).
    pub retained_snapshots: usize,
    /// `true` when the prefix held **no generation manifests**, so this is not
    /// provably a tier-2 sink and no base snapshot was collected.
    ///
    /// A `snapshots/` prefix with no chain over it is indistinguishable from a
    /// plain tier-1a sink, whose older snapshots are *history* rather than
    /// garbage — [`crate::snapshot::restore_latest`] takes the newest, but an
    /// operator restoring a point in time names an older key by hand. Pruning
    /// those is [`crate::dedup::gc_dedup`]'s `keep_n` decision to make, not
    /// this sweep's. A live tier-2 sink always has a chain, so in the steady
    /// state this is `false` and the superseded bases are collected; the one
    /// way to see it `true` on a real stream is a GC that lands inside
    /// `crate::tail::rebase`'s manifests-deleted-frames-not-yet-written window,
    /// where skipping one pass costs nothing.
    pub base_snapshots_skipped: bool,
    /// Frame objects kept because a generation manifest still names them.
    pub retained_frame_objects: usize,
    /// Objects that were unreachable but too young to touch — the grace window
    /// did its job. A number that never falls to zero across consecutive runs
    /// means the grace is longer than the churn interval, not that the sweep
    /// is broken.
    pub spared_by_grace: usize,
}

/// Tier-2 GC: reclaim the objects a `crate::tail::rebase` orphans.
///
/// The tier-1b counterpart is [`crate::dedup::gc_dedup`], and this deliberately
/// mirrors its shape: an explicitly-invoked sweep that deletes only what
/// nothing can reach, never a step on the write or recovery path. `rebase`
/// leaves its garbage behind on purpose — deleting data as part of a recovery
/// path is how a recovery path becomes the outage — and this is where that
/// debt is settled.
///
/// ## Two orphan classes, not one
///
/// R850-T3 was filed against the frames alone. The frames are the **smaller**
/// term:
///
/// - **Orphaned frame objects.** Each `rebase` abandons
///   `frames/{old_checkpoint_seq}/…` (or `frames/{epoch}/{seq}/…`) once
///   `crate::tail::delete_generation_manifests` removes the manifests that
///   named them. Bounded per rebase by SQLite's autocheckpoint threshold —
///   ~1000 pages, i.e. a handful of batch objects.
/// - **Superseded base snapshots.** `rebase` calls
///   [`crate::snapshot::upload_base_snapshot`], which is the explicitly
///   *non*-deduplicating one-shot variant: it `put`s a new
///   `snapshots/snapshot-{nanos}.db` and deletes nothing. Every rebase
///   therefore leaves a complete permanent copy of the database behind, and
///   for any database bigger than a few megabytes that dwarfs the frames.
///   Found while implementing this ticket; covered here rather than filed,
///   because it is the same leak in the same function.
///
/// A sweep that reclaimed only the frames would leave the larger leak running.
///
/// ## Liveness
///
/// Nothing is deleted unless *no* restore path can reach it. The rule is the
/// exact complement of what a restore selects, and the two are pinned together
/// by `gc_liveness_is_the_complement_of_restore_selection` in this module's
/// tests so they cannot drift:
///
/// - **Frames.** Live iff some generation manifest names the object, resolved
///   through `BackupTarget::frame_objects_of` — the same function restore's
///   replay and [`crate::puller::WalPuller`] go through, so batch and legacy
///   per-frame layouts and both epoch key shapes are handled by construction
///   rather than by a second copy of the layout rules here.
/// - **Base snapshots.** Live iff *either* some generation manifest names it,
///   *or* it is the lexically-greatest key under `snapshots/`. Those are the
///   two selections a restore makes:
///   [`restore_stream_from_manifests`] takes the chain's `base_snapshot_key`,
///   and a prefix with no generations falls back to
///   [`crate::snapshot::restore_latest`], which takes the newest key (see
///   [`crate::hydrate`]'s `restore_subject`). Their union is kept.
///
/// The base rule uses *every* manifest's key rather than
/// `validate_generation_chain`'s single answer, and that is the safe
/// direction: a chain that spans a WAL restart does not validate at all, and a
/// GC that refuses to guess keeps both bases instead of deleting the one the
/// next rebase is about to adopt.
///
/// And when there is no chain at all, the snapshot half is skipped entirely
/// rather than falling back to "keep the newest, collect the rest" — see
/// [`StreamGcOutcome::base_snapshots_skipped`] for why that distinction is not
/// paranoia. The frame half still runs: a `frames/` prefix under a sink with no
/// generation manifests is unreachable by construction.
///
/// ## Grace window
///
/// An object younger than [`StreamGcConfig::grace`] is never collected, no
/// matter how unreachable it looks. [`DEFAULT_STREAM_GC_GRACE`] documents the
/// three live windows that depend on it.
///
/// ## Order
///
/// Frames first, then snapshots. Unlike [`crate::dedup::gc_dedup`] the order is
/// not load-bearing — this sweep deletes no manifests, so nothing retained ever
/// points at anything collected, at any point during the pass or after a crash
/// in the middle of one.
pub async fn gc_stream(target: &BackupTarget, cfg: &StreamGcConfig) -> Result<StreamGcOutcome> {
    let manifests = list_and_parse_generation_manifests(target).await?;

    // Normalize through ObjPath so a manifest's textual key compares equal to
    // the same object's listed location regardless of slash padding.
    let mut live_bases: HashSet<String> = manifests
        .iter()
        .map(|m| ObjPath::from(m.base_snapshot_key.as_str()).to_string())
        .collect();
    if let Some(newest) = crate::snapshot::latest_snapshot_key(target).await? {
        live_bases.insert(ObjPath::from(newest).to_string());
    }

    let mut live_frames: HashSet<String> = HashSet::new();
    for m in &manifests {
        for (key, _first, _last) in target.frame_objects_of(m) {
            live_frames.insert(key.to_string());
        }
    }

    let mut out = StreamGcOutcome {
        dry_run: cfg.dry_run,
        base_snapshots_skipped: manifests.is_empty(),
        ..Default::default()
    };
    out.live_base_snapshot_keys = live_bases.iter().cloned().collect();
    out.live_base_snapshot_keys.sort();

    let now_secs = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0) as i64;
    let grace_secs = i64::try_from(cfg.grace.as_secs()).unwrap_or(i64::MAX);
    let too_young = |meta: &object_store::ObjectMeta| {
        now_secs.saturating_sub(meta.last_modified.timestamp()) < grace_secs
    };

    for meta in list_recursive(&target.store, &join_key(&target.prefix, "frames")).await? {
        if live_frames.contains(&meta.location.to_string()) {
            out.retained_frame_objects += 1;
            continue;
        }
        if too_young(&meta) {
            out.spared_by_grace += 1;
            continue;
        }
        if !cfg.dry_run {
            delete_collected(target, &meta.location).await?;
        }
        out.collected_bytes += meta.size;
        out.collected_frame_objects.push(meta.location.to_string());
    }

    let snapshots_prefix = join_key(&target.prefix, "snapshots");
    let listing = target
        .store
        .list_with_delimiter(Some(&snapshots_prefix))
        .await
        .with_context(|| format!("listing base snapshots under {snapshots_prefix}"))?;
    for meta in listing.objects {
        // No chain over this prefix means it is not provably a tier-2 sink, and
        // a tier-1a sink's older snapshots are history rather than garbage —
        // see `StreamGcOutcome::base_snapshots_skipped`.
        if out.base_snapshots_skipped || live_bases.contains(&meta.location.to_string()) {
            out.retained_snapshots += 1;
            continue;
        }
        if too_young(&meta) {
            out.spared_by_grace += 1;
            continue;
        }
        if !cfg.dry_run {
            delete_collected(target, &meta.location).await?;
        }
        out.collected_bytes += meta.size;
        out.collected_snapshots.push(meta.location.to_string());
    }

    out.collected_frame_objects.sort();
    out.collected_snapshots.sort();
    Ok(out)
}

/// Delete one collected object, treating "already gone" as success.
///
/// Two GC passes can legitimately overlap, and a retry after a partial one
/// lands here too — in both cases the object being absent is the outcome we
/// asked for, exactly as it is for `crate::tail::rebase`'s deletes.
async fn delete_collected(target: &BackupTarget, key: &ObjPath) -> Result<()> {
    match target.store.delete(key).await {
        Ok(()) => Ok(()),
        Err(object_store::Error::NotFound { .. }) => Ok(()),
        Err(e) => Err(anyhow::Error::new(e)).with_context(|| format!("collecting {key}")),
    }
}

/// Every object at or below `root`, walked with `list_with_delimiter` rather
/// than the streaming `list`.
///
/// `ObjectStore::list` returns a `BoxStream`, which needs `futures_util` to
/// drain — and this crate deliberately keeps `futures_util` in
/// `[dev-dependencies]` (see `Cargo.toml`), so a streaming drain here would add
/// a real dependency to the published crate for one listing. A worklist over
/// `list_with_delimiter` costs one request per directory instead of one per
/// page, and the tier-2 frame tree is three levels deep at most
/// (`frames/{epoch}/{seq}/{object}`), so that is a handful of requests.
async fn list_recursive(
    store: &Arc<dyn ObjectStore>,
    root: &ObjPath,
) -> Result<Vec<object_store::ObjectMeta>> {
    let mut found = Vec::new();
    let mut pending = vec![root.clone()];
    while let Some(dir) = pending.pop() {
        let listing = store
            .list_with_delimiter(Some(&dir))
            .await
            .with_context(|| format!("listing {dir}"))?;
        found.extend(listing.objects);
        pending.extend(listing.common_prefixes);
    }
    Ok(found)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backpressure::fault_injection::{Fault, FaultyStore};
    use crate::backpressure::BackoffConfig;
    use object_store::memory::InMemory;
    use std::cell::RefCell;
    use std::time::Duration;

    /// In-memory WAL seam: a fixed page_size, a vector of frames the test
    /// appends to, and an advancing checkpoint_seq the test can bump.
    struct MockWal {
        page_size: usize,
        state: RefCell<MockState>,
    }
    struct MockState {
        checkpoint_seq: u32,
        /// R858-B19: the WAL header salt this mock's frames are stamped with.
        /// Modelled explicitly because the two fold regimes move it
        /// differently, and the whole bug was reading only the sequence:
        /// [`MockWal::restart`] moves both, [`MockWal::recreate`] moves only
        /// this one.
        salt: WalSalt,
        frames: Vec<MockFrame>,
        auto_actions_disabled: bool,
        /// R858-B19: re-roll `salt` once this many frame reads have happened,
        /// so a test can land a WAL recreate *inside* a single `tail_frames`
        /// call rather than only between two. See
        /// [`MockWal::recreate_after_reads`].
        swap_salt_after_reads: Option<u32>,
        reads: u32,
    }
    #[derive(Clone)]
    struct MockFrame {
        info: FrameInfo,
        page_bytes: Vec<u8>,
    }

    impl MockWal {
        fn new(page_size: usize) -> Self {
            Self {
                page_size,
                state: RefCell::new(MockState {
                    checkpoint_seq: 0,
                    salt: WalSalt { salt1: 0xd492_ea8a, salt2: 0x7acf_42a3 },
                    frames: Vec::new(),
                    auto_actions_disabled: false,
                    swap_salt_after_reads: None,
                    reads: 0,
                }),
            }
        }
        fn append(&self, page_no: u32, db_size: u32, fill: u8) {
            let mut s = self.state.borrow_mut();
            s.frames.push(MockFrame {
                info: FrameInfo { page_no, db_size },
                page_bytes: vec![fill; self.page_size],
            });
        }
        /// Simulate an **in-process** WAL restart — a long-lived connection
        /// folding its own WAL at the autocheckpoint threshold. The WAL file is
        /// reused, so `checkpoint_seq` advances and `salt1` advances with it
        /// (measured `d492ea8a -> d492ea8b` alongside seq `0 -> 1`). This is
        /// the regime the pre-R858-B19 `checkpoint_seq` test already caught.
        fn restart(&self) {
            let mut s = self.state.borrow_mut();
            s.checkpoint_seq += 1;
            s.salt.salt1 = s.salt.salt1.wrapping_add(1);
            s.frames.clear();
        }
        /// R858-B19 — simulate a **writer-process restart**: the last
        /// connection closed, SQLite checkpointed and DELETED the `-wal` file,
        /// and the next writer created a fresh WAL. `checkpoint_seq` goes back
        /// to 0 (it never moves, from a watcher's point of view: `0 -> 0`) and
        /// the salt is fresh randomness, unrelated to the old one.
        ///
        /// This is the regime `checkpoint_seq` is blind to, and the reason the
        /// mock models a salt at all.
        fn recreate(&self, salt: WalSalt) {
            let mut s = self.state.borrow_mut();
            s.checkpoint_seq = 0;
            s.salt = salt;
            s.frames.clear();
        }
        /// R858-B19: arm a mid-call WAL recreate — the salt flips once `reads`
        /// frame reads have been served, which lands it inside the drain loop
        /// rather than between two `tail_frames` calls.
        fn recreate_after_reads(&self, reads: u32) {
            self.state.borrow_mut().swap_salt_after_reads = Some(reads);
        }
        fn frame_count(&self) -> u64 {
            self.state.borrow().frames.len() as u64
        }
    }

    impl WalSeam for MockWal {
        fn wal_state(&self) -> Result<Watermark> {
            let s = self.state.borrow();
            Ok(Watermark {
                checkpoint_seq: s.checkpoint_seq,
                last_frame: s.frames.len() as u64,
            })
        }
        fn wal_get_frame(&self, frame_no: u64, buf: &mut [u8]) -> Result<FrameInfo> {
            let mut s = self.state.borrow_mut();
            s.reads += 1;
            if s.swap_salt_after_reads == Some(s.reads) {
                s.salt.salt1 = !s.salt.salt1;
            }
            let idx = frame_no
                .checked_sub(1)
                .context("frame_no must be >= 1")? as usize;
            let f = s
                .frames
                .get(idx)
                .with_context(|| format!("frame {frame_no} out of range"))?;
            // Synthesize a 24-byte header: big-endian page_no, db_size, the
            // WAL's salt pair (R858-B19 reads the generation out of exactly
            // these bytes), then zeros for the checksums (never validated).
            buf[0..4].copy_from_slice(&f.info.page_no.to_be_bytes());
            buf[4..8].copy_from_slice(&f.info.db_size.to_be_bytes());
            buf[8..12].copy_from_slice(&s.salt.salt1.to_be_bytes());
            buf[12..16].copy_from_slice(&s.salt.salt2.to_be_bytes());
            buf[16..WAL_FRAME_HEADER_SIZE].fill(0);
            buf[WAL_FRAME_HEADER_SIZE..].copy_from_slice(&f.page_bytes);
            Ok(f.info)
        }
        fn wal_auto_actions_disable(&self) {
            self.state.borrow_mut().auto_actions_disabled = true;
        }
    }

    /// A store that accepts every put unconditionally — the exact failure the
    /// preflight probe exists to catch. It is not a contrived shape: it is
    /// what an S3-compatible backend without conditional-write support looks
    /// like from `object_store`'s side, and what `AmazonS3Builder` degrades to
    /// when pointed at one. Everything but `put_opts` delegates.
    #[derive(Debug)]
    struct UnconditionalStore {
        inner: Arc<dyn ObjectStore>,
    }

    impl std::fmt::Display for UnconditionalStore {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "UnconditionalStore({})", self.inner)
        }
    }

    #[async_trait::async_trait]
    impl ObjectStore for UnconditionalStore {
        async fn put_opts(
            &self,
            location: &ObjPath,
            payload: object_store::PutPayload,
            mut opts: PutOptions,
        ) -> object_store::Result<object_store::PutResult> {
            opts.mode = PutMode::Overwrite;
            self.inner.put_opts(location, payload, opts).await
        }
        async fn put_multipart_opts(
            &self,
            location: &ObjPath,
            opts: object_store::PutMultipartOptions,
        ) -> object_store::Result<Box<dyn object_store::MultipartUpload>> {
            self.inner.put_multipart_opts(location, opts).await
        }
        async fn get_opts(
            &self,
            location: &ObjPath,
            options: object_store::GetOptions,
        ) -> object_store::Result<object_store::GetResult> {
            self.inner.get_opts(location, options).await
        }
        fn delete_stream(
            &self,
            locations: futures_util::stream::BoxStream<'static, object_store::Result<ObjPath>>,
        ) -> futures_util::stream::BoxStream<'static, object_store::Result<ObjPath>> {
            self.inner.delete_stream(locations)
        }
        fn list(
            &self,
            prefix: Option<&ObjPath>,
        ) -> futures_util::stream::BoxStream<'static, object_store::Result<object_store::ObjectMeta>>
        {
            self.inner.list(prefix)
        }
        async fn list_with_delimiter(
            &self,
            prefix: Option<&ObjPath>,
        ) -> object_store::Result<object_store::ListResult> {
            self.inner.list_with_delimiter(prefix).await
        }
        async fn copy_opts(
            &self,
            from: &ObjPath,
            to: &ObjPath,
            options: object_store::CopyOptions,
        ) -> object_store::Result<()> {
            self.inner.copy_opts(from, to, options).await
        }
    }

    /// A store that honours `If-None-Match` but not `If-Match`. The nastiest
    /// real-world shape, because the bootstrap put looks fine and only the
    /// steady-state advance — the one every tail after the first depends on —
    /// is unguarded.
    #[derive(Debug)]
    struct CreateOnlyStore {
        inner: Arc<dyn ObjectStore>,
    }

    impl std::fmt::Display for CreateOnlyStore {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "CreateOnlyStore({})", self.inner)
        }
    }

    #[async_trait::async_trait]
    impl ObjectStore for CreateOnlyStore {
        async fn put_opts(
            &self,
            location: &ObjPath,
            payload: object_store::PutPayload,
            mut opts: PutOptions,
        ) -> object_store::Result<object_store::PutResult> {
            if matches!(opts.mode, PutMode::Update(_)) {
                opts.mode = PutMode::Overwrite;
            }
            self.inner.put_opts(location, payload, opts).await
        }
        async fn put_multipart_opts(
            &self,
            location: &ObjPath,
            opts: object_store::PutMultipartOptions,
        ) -> object_store::Result<Box<dyn object_store::MultipartUpload>> {
            self.inner.put_multipart_opts(location, opts).await
        }
        async fn get_opts(
            &self,
            location: &ObjPath,
            options: object_store::GetOptions,
        ) -> object_store::Result<object_store::GetResult> {
            self.inner.get_opts(location, options).await
        }
        fn delete_stream(
            &self,
            locations: futures_util::stream::BoxStream<'static, object_store::Result<ObjPath>>,
        ) -> futures_util::stream::BoxStream<'static, object_store::Result<ObjPath>> {
            self.inner.delete_stream(locations)
        }
        fn list(
            &self,
            prefix: Option<&ObjPath>,
        ) -> futures_util::stream::BoxStream<'static, object_store::Result<object_store::ObjectMeta>>
        {
            self.inner.list(prefix)
        }
        async fn list_with_delimiter(
            &self,
            prefix: Option<&ObjPath>,
        ) -> object_store::Result<object_store::ListResult> {
            self.inner.list_with_delimiter(prefix).await
        }
        async fn copy_opts(
            &self,
            from: &ObjPath,
            to: &ObjPath,
            options: object_store::CopyOptions,
        ) -> object_store::Result<()> {
            self.inner.copy_opts(from, to, options).await
        }
    }

    /// The happy path: a store that honours both modes passes, and — the part
    /// that matters for running this at startup against a live sink — leaves
    /// nothing behind.
    #[tokio::test]
    async fn the_preflight_probe_passes_on_a_conditional_store_and_leaves_no_trace() {
        let target = fresh_target();
        assert_eq!(
            probe_conditional_puts(&target).await.unwrap(),
            PreconditionSupport::Honoured
        );

        let leftovers = objects_under(&target).await;
        assert!(
            leftovers.is_empty(),
            "the probe must clean up its canary — found {leftovers:?}"
        );
    }

    /// The whole point: a backend that silently ignores preconditions is
    /// *reported*, not tolerated. Without this the watermark CAS degrades to
    /// last-write-wins and every other test in this file still passes.
    #[tokio::test]
    async fn the_preflight_probe_catches_a_store_that_ignores_preconditions() {
        let target = BackupTarget {
            store: Arc::new(UnconditionalStore { inner: Arc::new(InMemory::new()) }),
            prefix: "backups".into(),
        };
        assert_eq!(
            probe_conditional_puts(&target).await.unwrap(),
            PreconditionSupport::Degraded { stage: PreflightStage::Create },
            "an unconditional store fails at the first guard it meets"
        );
    }

    /// Half-degraded stores are the ones that actually ship. `If-None-Match`
    /// works, so bootstrapping looks healthy; `If-Match` does not, so every
    /// steady-state advance is unguarded. The probe must name `Update`
    /// specifically — "conditional puts are broken" would send an operator to
    /// the wrong setting.
    #[tokio::test]
    async fn the_preflight_probe_names_update_when_only_if_match_is_ignored() {
        let target = BackupTarget {
            store: Arc::new(CreateOnlyStore { inner: Arc::new(InMemory::new()) }),
            prefix: "backups".into(),
        };
        assert_eq!(
            probe_conditional_puts(&target).await.unwrap(),
            PreconditionSupport::Degraded { stage: PreflightStage::Update }
        );
    }

    /// The canary is keyed per process per nanosecond, so two nodes probing
    /// the same sink at once each get a real verdict instead of failing each
    /// other. A startup probe that flaked under concurrency would be turned
    /// off within a week.
    #[tokio::test]
    async fn concurrent_preflight_probes_do_not_collide() {
        let target = fresh_target();
        let (a, b) = tokio::join!(
            probe_conditional_puts(&target),
            probe_conditional_puts(&target)
        );
        assert_eq!(a.unwrap(), PreconditionSupport::Honoured);
        assert_eq!(b.unwrap(), PreconditionSupport::Honoured);
        assert!(objects_under(&target).await.is_empty());
    }

    async fn objects_under(target: &BackupTarget) -> Vec<String> {
        use futures_util::StreamExt;
        target
            .store
            .list(None)
            .map(|m| m.unwrap().location.to_string())
            .collect()
            .await
    }

    fn fresh_target() -> BackupTarget {
        BackupTarget {
            store: Arc::new(InMemory::new()),
            prefix: "backups".into(),
        }
    }

    fn cfg() -> StreamConfig<'static> {
        StreamConfig {
            base_snapshot_key: "backups/snapshots/snapshot-00000000000000000001.db",
            page_size: 4096,
            backpressure: BackpressureConfig::default(),
            rpo_target: None,
            epoch: 0,
            owner: None,
            pointer_generation: 0,
        }
    }

    /// Fast-ticking backoff for backpressure tests — a few ms, never the
    /// production defaults, so retry-heavy tests stay fast.
    fn fast_backoff() -> BackoffConfig {
        BackoffConfig {
            initial_delay: Duration::from_millis(1),
            max_delay: Duration::from_millis(4),
            multiplier: 2.0,
            max_retries: 3,
        }
    }

    fn bp_cfg(policy: BackpressurePolicy, spill_buffer_frames: usize) -> StreamConfig<'static> {
        StreamConfig {
            base_snapshot_key: "backups/snapshots/snapshot-00000000000000000001.db",
            page_size: 4096,
            backpressure: BackpressureConfig {
                spill_buffer_frames,
                policy,
                backoff: fast_backoff(),
            },
            rpo_target: None,
            epoch: 0,
            owner: None,
            pointer_generation: 0,
        }
    }

    fn faulty_target(faults: impl IntoIterator<Item = Fault>) -> BackupTarget {
        BackupTarget {
            store: Arc::new(FaultyStore::new(Arc::new(InMemory::new()), faults)),
            prefix: "backups".into(),
        }
    }

    /// Five WAL frames (1..=5), the last a commit — a small, deterministic
    /// range for the backpressure tests below.
    fn five_frame_seam() -> MockWal {
        let seam = MockWal::new(4096);
        for i in 1..=4u32 {
            seam.append(i, 0, i as u8);
        }
        seam.append(5, 5, 5); // commit
        seam
    }

    // --- R574-F2: explicit R2 backpressure ---------------------------------

    /// A single 429 triggers exactly one retry, then the upload succeeds —
    /// the whole range still lands and the report shows the retry.
    #[tokio::test]
    async fn throttled_429_retries_then_succeeds() {
        let seam = five_frame_seam();
        let target = faulty_target([Fault::TooManyRequests]);
        let cfg = bp_cfg(BackpressurePolicy::Fail, 8);
        let out = tail_frames(&seam, &target, &cfg).await.unwrap();
        match out {
            StreamOutcome::Streamed { frame_count, backpressure, .. } => {
                assert_eq!(frame_count, 5);
                assert_eq!(backpressure.throttle_retries, 1);
                assert_eq!(backpressure.frames_shed, 0);
            }
            other => panic!("expected Streamed, got {other:?}"),
        }
    }

    /// A 503 is classified the same as a 429 and also triggers backoff then
    /// retry — two 503s in a row cost exactly two retries.
    #[tokio::test]
    async fn throttled_503_retries_then_succeeds() {
        let seam = five_frame_seam();
        let target = faulty_target([Fault::ServiceUnavailable, Fault::ServiceUnavailable]);
        let cfg = bp_cfg(BackpressurePolicy::Fail, 8);
        let out = tail_frames(&seam, &target, &cfg).await.unwrap();
        match out {
            StreamOutcome::Streamed { frame_count, backpressure, .. } => {
                assert_eq!(frame_count, 5);
                assert_eq!(backpressure.throttle_retries, 2);
            }
            other => panic!("expected Streamed, got {other:?}"),
        }
    }

    /// Buffer at bound + `Fail`: sustained throttling exhausts backoff and
    /// the whole call errors — nothing is persisted.
    #[tokio::test]
    async fn buffer_at_bound_fail_policy_errors_without_persisting() {
        let seam = five_frame_seam();
        let faults = std::iter::repeat_n(Fault::TooManyRequests, 50);
        let target = faulty_target(faults);
        let cfg = bp_cfg(BackpressurePolicy::Fail, 2);
        let err = tail_frames(&seam, &target, &cfg).await.unwrap_err();
        assert!(format!("{err}").contains("uploading wal frame"), "err was {err}");
        assert!(
            read_watermark(&target.store, &target.watermark_key())
                .await
                .unwrap()
                .is_none(),
            "Fail must not persist a watermark when it gives up"
        );
    }

    /// Buffer at bound + `Shed`: the backlog is dropped with a loud report
    /// instead of erroring; nothing persists, so the next call would
    /// re-attempt the same range.
    #[tokio::test]
    async fn buffer_at_bound_shed_policy_drops_backlog_without_error() {
        let seam = five_frame_seam();
        let faults = std::iter::repeat_n(Fault::TooManyRequests, 50);
        let target = faulty_target(faults);
        let cfg = bp_cfg(BackpressurePolicy::Shed, 2);
        let out = tail_frames(&seam, &target, &cfg).await.unwrap();
        match out {
            StreamOutcome::Shed { first_frame, last_frame, backpressure, .. } => {
                assert_eq!(first_frame, 1);
                assert_eq!(last_frame, 5);
                assert_eq!(backpressure.high_water_frames, 2, "capped at the bound");
                assert!(backpressure.frames_shed >= 1, "must report the dropped backlog");
            }
            other => panic!("expected Shed, got {other:?}"),
        }
        assert!(
            read_watermark(&target.store, &target.watermark_key())
                .await
                .unwrap()
                .is_none(),
            "Shed must not persist a watermark for a fully-dropped batch"
        );
    }

    /// Buffer at bound + `Block`: retries never give up on a throttling
    /// error; once the store recovers, the full range still lands.
    #[tokio::test]
    async fn buffer_at_bound_block_policy_eventually_drains() {
        let seam = five_frame_seam();
        // More failures than fast_backoff's max_retries would tolerate under
        // Fail/Shed — Block must push through them anyway.
        let faults = std::iter::repeat_n(Fault::TooManyRequests, 4);
        let target = faulty_target(faults);
        let cfg = bp_cfg(BackpressurePolicy::Block, 2);
        let out = tail_frames(&seam, &target, &cfg).await.unwrap();
        match out {
            StreamOutcome::Streamed { frame_count, backpressure, .. } => {
                assert_eq!(frame_count, 5);
                assert_eq!(backpressure.high_water_frames, 2);
                assert_eq!(backpressure.throttle_retries, 4);
                assert_eq!(backpressure.frames_shed, 0);
            }
            other => panic!("expected Streamed, got {other:?}"),
        }
    }

    /// R761-F2: a call with more frames than the spill buffer holds splits
    /// into one batch object per buffer-full, the manifest indexes every one
    /// of them, and replay puts the frames back in order. The bound is the
    /// only thing sizing an object, so this is also the assertion that the
    /// largest object this sink writes stays bounded.
    #[tokio::test]
    async fn a_call_larger_than_the_spill_buffer_splits_into_several_batch_objects() {
        let seam = MockWal::new(4096);
        for i in 1..=5u32 {
            seam.append(i, i, i as u8);
        }
        let target = fresh_target();
        let cfg = bp_cfg(BackpressurePolicy::Fail, 2);
        let out = tail_frames(&seam, &target, &cfg).await.unwrap();
        assert!(matches!(out, StreamOutcome::Streamed { frame_count: 5, .. }));

        let manifests = list_and_parse_generation_manifests(&target).await.unwrap();
        assert_eq!(
            manifests[0].frame_batches,
            vec![(1, 2), (3, 4), (5, 5)],
            "bound 2 over 5 frames is two full batches and a remainder"
        );
        let frame_size = WAL_FRAME_HEADER_SIZE + 4096;
        for (first, last) in [(1u64, 2u64), (3, 4), (5, 5)] {
            let bytes = target
                .store
                .get(&target.frame_batch_key(0, 0, first, last))
                .await
                .unwrap()
                .bytes()
                .await
                .unwrap();
            assert_eq!(bytes.len() as u64, (last - first + 1) * frame_size as u64);
        }

        // And it all comes back, in order, through the normal read path.
        let insert = MockInsertSeam::new();
        assert_eq!(replay_frames_into(&target, &insert, &manifests).await.unwrap(), 5);
        let frames: Vec<u64> = insert
            .events()
            .into_iter()
            .filter_map(|e| match e {
                MockInsertEvent::Frame { frame_no, .. } => Some(frame_no),
                _ => None,
            })
            .collect();
        assert_eq!(frames, vec![1, 2, 3, 4, 5]);
    }

    /// R761-F2 + R574-F2: a drain that lands one batch and then gets stuck
    /// under `Shed` publishes ONLY the batch that landed. The manifest's index
    /// is what restore follows, so a batch list claiming a shed object would
    /// be a 404 mid-replay — worse than the frames simply not being there.
    #[tokio::test]
    async fn a_partially_shed_drain_indexes_only_the_batches_that_landed() {
        let seam = five_frame_seam();
        // First batch through; the next one throttled until Shed gives up.
        // Exactly enough faults to exhaust one put's retry budget and no more,
        // so the watermark + manifest writes that follow the shed still land
        // (they go through the same store).
        let faults = std::iter::once(Fault::Pass).chain(std::iter::repeat_n(
            Fault::TooManyRequests,
            fast_backoff().max_retries as usize + 1,
        ));
        let target = faulty_target(faults);
        let cfg = bp_cfg(BackpressurePolicy::Shed, 2);
        let out = tail_frames(&seam, &target, &cfg).await.unwrap();
        match out {
            StreamOutcome::Streamed { first_frame, last_frame, frame_count, backpressure, .. } => {
                assert_eq!((first_frame, last_frame, frame_count), (1, 2, 2));
                assert_eq!(backpressure.frames_shed, 2, "frames 3-4 were buffered and dropped");
            }
            other => panic!("expected a Streamed prefix, got {other:?}"),
        }
        let manifests = list_and_parse_generation_manifests(&target).await.unwrap();
        assert_eq!(manifests.len(), 1);
        assert_eq!(manifests[0].last_frame, 2);
        assert_eq!(manifests[0].frame_batches, vec![(1, 2)]);
        // Every object the manifest names actually exists — the property that
        // makes the published prefix restorable.
        let insert = MockInsertSeam::new();
        assert_eq!(replay_frames_into(&target, &insert, &manifests).await.unwrap(), 2);
    }

    /// The high-water mark reports the peak spill-buffer occupancy for the
    /// call, capped at the configured bound even when more frames remain.
    #[tokio::test]
    async fn high_water_reports_peak_buffered_frames() {
        let seam = five_frame_seam();
        let target = faulty_target([]); // no faults — pure high-water measurement
        let cfg = bp_cfg(BackpressurePolicy::Fail, 3);
        let out = tail_frames(&seam, &target, &cfg).await.unwrap();
        match out {
            StreamOutcome::Streamed { backpressure, .. } => {
                assert_eq!(backpressure.high_water_frames, 3, "capped at the configured bound");
            }
            other => panic!("expected Streamed, got {other:?}"),
        }
    }

    // --- R574-T4: explicit RPO knob --------------------------------------

    /// First tail ever: no sidecar, so age is unknown and a configured
    /// target cannot be breached (there is nothing to measure against).
    #[tokio::test]
    async fn rpo_first_tail_has_unknown_age_and_no_breach() {
        let seam = five_frame_seam();
        let target = fresh_target();
        let mut cfg = cfg();
        cfg.rpo_target = Some(Duration::from_secs(1));
        let out = tail_frames(&seam, &target, &cfg).await.unwrap();
        match out {
            StreamOutcome::Streamed { rpo, .. } => {
                assert_eq!(rpo.target, Some(Duration::from_secs(1)));
                assert_eq!(rpo.watermark_age, None);
                assert!(!rpo.breached);
            }
            other => panic!("expected Streamed, got {other:?}"),
        }
    }

    /// Second tail past the target: the sidecar's stamped write instant is
    /// older than `rpo_target`, so the outcome flags a breach — the
    /// staleness emission the orchestrator alerts on.
    #[tokio::test]
    async fn rpo_stale_watermark_past_target_reports_breach() {
        let seam = MockWal::new(4096);
        seam.append(1, 1, 0xAA);
        let target = fresh_target();
        let mut cfg = cfg();
        // Zero target: any measurable gap between the two tails is a breach.
        cfg.rpo_target = Some(Duration::ZERO);
        let _ = tail_frames(&seam, &target, &cfg).await.unwrap();
        tokio::time::sleep(Duration::from_millis(5)).await;
        seam.append(2, 2, 0xBB);
        let out = tail_frames(&seam, &target, &cfg).await.unwrap();
        match out {
            StreamOutcome::Streamed { rpo, .. } => {
                let age = rpo.watermark_age.expect("age known after first sidecar write");
                assert!(age >= Duration::from_millis(5), "age was {age:?}");
                assert!(rpo.breached, "zero target must flag any nonzero age");
            }
            other => panic!("expected Streamed, got {other:?}"),
        }
    }

    /// A generous target with a prompt second tail: age is reported but the
    /// bound holds — and with no target at all, `breached` is always false.
    #[tokio::test]
    async fn rpo_within_target_and_no_target_do_not_breach() {
        let seam = MockWal::new(4096);
        seam.append(1, 1, 0xAA);
        let target = fresh_target();
        let mut with_target = cfg();
        with_target.rpo_target = Some(Duration::from_secs(3600));
        let _ = tail_frames(&seam, &target, &with_target).await.unwrap();

        seam.append(2, 2, 0xBB);
        let out = tail_frames(&seam, &target, &with_target).await.unwrap();
        match out {
            StreamOutcome::Streamed { rpo, .. } => {
                assert!(rpo.watermark_age.is_some());
                assert!(!rpo.breached, "an hour budget can't be blown in-process");
            }
            other => panic!("expected Streamed, got {other:?}"),
        }

        // Same staged sidecar, target removed: age still reported, never
        // breached (observe-before-you-pick-a-number mode).
        seam.append(3, 3, 0xCC);
        let out = tail_frames(&seam, &target, &cfg()).await.unwrap();
        match out {
            StreamOutcome::Streamed { rpo, .. } => {
                assert_eq!(rpo.target, None);
                assert!(rpo.watermark_age.is_some());
                assert!(!rpo.breached);
            }
            other => panic!("expected Streamed, got {other:?}"),
        }
    }

    /// A pre-T4 two-field sidecar still parses (written_at unknown), and the
    /// next write upgrades it to the stamped three-field format.
    #[tokio::test]
    async fn rpo_legacy_two_field_sidecar_parses_and_upgrades() {
        let target = fresh_target();
        target
            .store
            .put(&target.watermark_key(), b"0 1\n".to_vec().into())
            .await
            .unwrap();
        let legacy = read_watermark(&target.store, &target.watermark_key())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(legacy.watermark, Watermark { checkpoint_seq: 0, last_frame: 1 });
        assert_eq!(legacy.written_at_nanos, None);

        // A tail against the legacy sidecar reports unknown age (not a
        // breach), uploads the frames, and re-stamps the sidecar.
        //
        // R858-B19 CHANGED THE OUTCOME HERE, deliberately: this used to assert
        // `Streamed` with `first_frame == 2` ("resumes after the legacy
        // watermark"). A two-field sidecar records no salt, so nothing says the
        // WAL it names is the WAL in front of us — and resuming at frame 2 on
        // that basis is the exact inference that spliced two generations. The
        // sidecar's RPO semantics (an *unknown* age is not a breach) are what
        // this test is about and they are untouched.
        let seam = MockWal::new(4096);
        seam.append(1, 0, 0xAA);
        seam.append(2, 2, 0xBB);
        let mut cfg = cfg();
        cfg.rpo_target = Some(Duration::ZERO);
        let out = tail_frames(&seam, &target, &cfg).await.unwrap();
        match out {
            StreamOutcome::Restarted { rpo, first_frame, previous_generation, .. } => {
                assert_eq!(first_frame, 1, "an unverifiable watermark is re-uploaded, not resumed");
                assert_eq!(previous_generation.salt, None);
                assert_eq!(rpo.watermark_age, None);
                assert!(!rpo.breached, "unknown age is not a breach even at zero target");
            }
            other => panic!("expected Restarted, got {other:?}"),
        }
        let upgraded = read_watermark(&target.store, &target.watermark_key())
            .await
            .unwrap()
            .unwrap();
        assert!(upgraded.written_at_nanos.is_some(), "rewrite stamps the timestamp");
    }

    /// Pure rpo_status edge cases that don't need a staged store.
    #[test]
    fn rpo_status_truth_table() {
        let stamped = |nanos_ago: u128| PersistedWatermark {
            watermark: Watermark { checkpoint_seq: 0, last_frame: 1 },
            generation: WalGeneration::default(),
            written_at_nanos: Some(unix_nanos().saturating_sub(nanos_ago)),
            epoch: 0,
            pointer_generation: 0,
            version: None,
        };
        // No prior at all.
        let s = rpo_status(Some(Duration::from_secs(1)), None);
        assert_eq!((s.watermark_age, s.breached), (None, false));
        // Prior without a stamp (legacy sidecar).
        let legacy = PersistedWatermark {
            watermark: Watermark::default(),
            generation: WalGeneration::default(),
            written_at_nanos: None,
            epoch: 0,
            pointer_generation: 0,
            version: None,
        };
        let s = rpo_status(Some(Duration::ZERO), Some(&legacy));
        assert_eq!((s.watermark_age, s.breached), (None, false));
        // Old stamp vs tight target: breached.
        let s = rpo_status(Some(Duration::from_millis(1)), Some(&stamped(5_000_000_000)));
        assert!(s.watermark_age.unwrap() >= Duration::from_secs(4));
        assert!(s.breached);
        // Old stamp, no target: age known, never breached.
        let s = rpo_status(None, Some(&stamped(5_000_000_000)));
        assert!(s.watermark_age.is_some());
        assert!(!s.breached);
    }

    /// R761-T1: the shipped cadence default is an arithmetic consequence of
    /// the 2026-08-13 `tail_sweep_harness` table, so the arithmetic is a
    /// test rather than a claim in a comment nobody re-checks. If a future
    /// harness run moves the measured points, this test is where the
    /// mismatch surfaces — re-derive the default, do not relax the test.
    #[test]
    fn default_tail_cadence_matches_the_measured_write_op_curve() {
        // Every uploading tail_frames call writes exactly two fixed objects
        // beyond the frames: one generation manifest, one watermark CAS.
        const FIXED_OBJECTS_PER_TAIL: f64 = 2.0;
        // Measured, flat across the whole sweep — a property of the schema
        // and transaction shape, not of the cadence.
        const MEASURED_FRAMES_PER_WRITE: f64 = 2.03;
        let puts_per_write =
            |w: f64| MEASURED_FRAMES_PER_WRITE + FIXED_OBJECTS_PER_TAIL / w;

        // The model reproduces all four measured rows.
        for (writes_per_tail, measured) in [(1.0, 4.03), (5.0, 2.43), (25.0, 2.11), (100.0, 2.05)] {
            let modelled = puts_per_write(writes_per_tail);
            assert!(
                (modelled - measured).abs() < 0.005,
                "writes_per_tail={writes_per_tail}: model {modelled:.3} vs measured {measured:.3}"
            );
        }

        // The default is tailed at half the stated bound, so one missed tick
        // still lands inside the promise.
        assert_eq!(DEFAULT_RPO_TARGET, DEFAULT_TAIL_INTERVAL * 2);

        // At the burst rates the default was chosen against (~0.1-1 write/s
        // during an active session), 60s lands in the flat part of the curve:
        // the fixed-object term is under a fifth of the frame floor even at
        // the slow end, where 15s would still be paying 1.33.
        let slow_burst_writes_per_tail = 0.1 * DEFAULT_TAIL_INTERVAL.as_secs_f64();
        let fixed_term = FIXED_OBJECTS_PER_TAIL / slow_burst_writes_per_tail;
        assert!(
            fixed_term < MEASURED_FRAMES_PER_WRITE / 5.0,
            "fixed-object term {fixed_term:.3} at the slow-burst end is no longer small \
             relative to the {MEASURED_FRAMES_PER_WRITE} frame floor"
        );
        assert!(
            puts_per_write(slow_burst_writes_per_tail) < 2.4,
            "the default's worst modelled case should sit below the measured \
             writes_per_tail=5 point (2.43)"
        );

        // R761-F2 removed the frames/write term: a tail call whose frames fit
        // one batch writes the batch, the manifest and the watermark, full
        // stop. Same curve shape, no floor.
        let batched_puts_per_write = |w: f64| (1.0 + FIXED_OBJECTS_PER_TAIL) / w;
        for (writes_per_tail, pre_batching) in [(1.0, 4.03), (5.0, 2.43), (25.0, 2.11), (100.0, 2.05)]
        {
            let cut = 1.0 - batched_puts_per_write(writes_per_tail) / pre_batching;
            let want = match writes_per_tail as u32 {
                1 => 0.256,
                5 => 0.753,
                25 => 0.943,
                _ => 0.985,
            };
            assert!(
                (cut - want).abs() < 0.005,
                "writes_per_tail={writes_per_tail}: batching cuts {:.1}%, expected {:.1}%",
                cut * 100.0,
                want * 100.0,
            );
        }
        // Which is why the cadence default did not move: what is left to win
        // past 60s is now a hundredth of a PUT per write at the fast-burst end
        // of the same band, against an RPO window that would grow 5x.
        assert!(
            batched_puts_per_write(1.0 * DEFAULT_TAIL_INTERVAL.as_secs_f64())
                - batched_puts_per_write(5.0 * DEFAULT_TAIL_INTERVAL.as_secs_f64())
                < 0.05,
            "batching should have flattened the cadence lever at the fast-burst end"
        );
    }

    /// R782: `StreamOutcome::rpo()` is the accessor `tenant-streamer` pushes
    /// through — every variant that carries an `RpoStatus` gives it back, and
    /// `Fenced` (which deliberately carries none, per its own doc) gives
    /// `None` rather than a default/synthesized one.
    #[test]
    fn stream_outcome_rpo_accessor_covers_every_variant() {
        let rpo = RpoStatus { target: None, watermark_age: Some(Duration::from_secs(1)), breached: false };
        assert_eq!(StreamOutcome::Empty { watermark: Watermark::default(), rpo }.rpo(), Some(&rpo));
        assert_eq!(
            StreamOutcome::Streamed {
                generation_key: String::new(),
                first_frame: 1,
                last_frame: 1,
                checkpoint_seq: 0,
                frame_count: 1,
                backpressure: Default::default(),
                rpo,
            }
            .rpo(),
            Some(&rpo)
        );
        assert_eq!(
            StreamOutcome::Restarted {
                generation_key: String::new(),
                previous_generation: WalGeneration::default(),
                new_generation: WalGeneration {
                    checkpoint_seq: 1,
                    salt: Some(WalSalt { salt1: 1, salt2: 2 }),
                },
                first_frame: 1,
                last_frame: 1,
                frame_count: 1,
                backpressure: Default::default(),
                rpo,
            }
            .rpo(),
            Some(&rpo)
        );
        assert_eq!(
            StreamOutcome::Shed {
                checkpoint_seq: 0,
                first_frame: 1,
                last_frame: 1,
                backpressure: Default::default(),
                rpo,
            }
            .rpo(),
            Some(&rpo)
        );
        assert_eq!(
            StreamOutcome::Fenced {
                current_epoch: 2,
                our_epoch: 1,
                current_pointer_generation: 0,
                our_pointer_generation: 0,
            }
            .rpo(),
            None
        );
    }

    /// First tail with no frames yet — Empty, no manifest, no watermark.
    #[tokio::test]
    async fn empty_when_no_frames() {
        let seam = MockWal::new(4096);
        let target = fresh_target();
        let out = tail_frames(&seam, &target, &cfg()).await.unwrap();
        match out {
            StreamOutcome::Empty { watermark, rpo } => {
                assert_eq!(watermark, Watermark::default());
                // No sidecar has ever been written — age is unknowable.
                assert_eq!(rpo, RpoStatus::default());
            }
            other => panic!("expected Empty, got {other:?}"),
        }
        // No manifest, no watermark sidecar.
        assert!(
            read_watermark(&target.store, &target.watermark_key())
                .await
                .unwrap()
                .is_none()
        );
    }

    /// First tail with 3 frames — Streamed, manifest written, watermark
    /// recorded, every frame object retrievable at the right key.
    #[tokio::test]
    async fn streams_initial_frames_and_records_watermark() {
        let seam = MockWal::new(4096);
        seam.append(1, 0, 0xAA);
        seam.append(2, 0, 0xBB);
        seam.append(3, 3, 0xCC); // commit frame: db_size = 3 pages

        let target = fresh_target();
        let out = tail_frames(&seam, &target, &cfg()).await.unwrap();
        let gen_key = match out {
            StreamOutcome::Streamed {
                generation_key,
                first_frame,
                last_frame,
                checkpoint_seq,
                frame_count,
                ..
            } => {
                assert_eq!(first_frame, 1);
                assert_eq!(last_frame, 3);
                assert_eq!(checkpoint_seq, 0);
                assert_eq!(frame_count, 3);
                assert!(generation_key.starts_with("backups/generations/gen-"));
                generation_key
            }
            other => panic!("expected Streamed, got {other:?}"),
        };

        // R761-F2: ONE batch object holds all three frames, at the ranged key,
        // each frame at its own offset inside it.
        let frame_size = WAL_FRAME_HEADER_SIZE + 4096;
        let bytes = target
            .store
            .get(&target.frame_batch_key(0, 0, 1, 3))
            .await
            .unwrap()
            .bytes()
            .await
            .unwrap();
        assert_eq!(bytes.len(), 3 * frame_size);
        for frame_no in 1..=3u64 {
            let at = (frame_no as usize - 1) * frame_size;
            let page_no = u32::from_be_bytes(bytes[at..at + 4].try_into().unwrap());
            assert_eq!(page_no, frame_no as u32);
        }
        // And nothing was written per-frame.
        assert!(
            target.store.get(&target.frame_key(0, 0, 1)).await.is_err(),
            "the pre-R761-F2 per-frame key must not be written any more"
        );

        // Manifest is parseable and round-trips.
        let m_bytes = target
            .store
            .get(&ObjPath::from(gen_key.clone()))
            .await
            .unwrap()
            .bytes()
            .await
            .unwrap();
        let parsed = parse_generation_manifest(&String::from_utf8_lossy(&m_bytes)).unwrap();
        assert_eq!(parsed.first_frame, 1);
        assert_eq!(parsed.last_frame, 3);
        assert_eq!(parsed.checkpoint_seq, 0);
        assert_eq!(parsed.page_size, 4096);
        assert_eq!(parsed.base_snapshot_key, cfg().base_snapshot_key);
        assert_eq!(parsed.frame_batches, vec![(1, 3)], "R761-F2: one batch, indexed");

        // Watermark sidecar matches.
        let wm = read_watermark(&target.store, &target.watermark_key())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(wm.watermark, Watermark { checkpoint_seq: 0, last_frame: 3 });
        assert!(wm.written_at_nanos.is_some(), "R574-T4: writes stamp the sidecar");

        // The seam was told to take WAL ownership? Not directly via
        // tail_frames — callers do that at construction (CoreWalSeam::open).
        // Verify the mock invariant separately.
        assert!(!seam.state.borrow().auto_actions_disabled);
    }

    /// Second tail with no new frames returns Empty without uploading.
    #[tokio::test]
    async fn second_tail_with_no_new_frames_is_empty() {
        let seam = MockWal::new(4096);
        seam.append(1, 1, 0xAA);
        let target = fresh_target();
        let _ = tail_frames(&seam, &target, &cfg()).await.unwrap();

        let out = tail_frames(&seam, &target, &cfg()).await.unwrap();
        match out {
            StreamOutcome::Empty { watermark, rpo } => {
                assert_eq!(watermark, Watermark { checkpoint_seq: 0, last_frame: 1 });
                // A sidecar exists from the first tail, so age is known even
                // with no rpo_target configured — and no target means no breach.
                assert!(rpo.watermark_age.is_some());
                assert!(!rpo.breached);
            }
            other => panic!("expected Empty, got {other:?}"),
        }
    }

    /// Second tail with new frames uploads only the new ones and the
    /// generation manifest covers the incremental range. Previous frames
    /// remain at their prior keys (idempotent put on same key).
    #[tokio::test]
    async fn second_tail_uploads_only_new_frames() {
        let seam = MockWal::new(4096);
        seam.append(1, 0, 0x11);
        seam.append(2, 2, 0x22); // commit
        let target = fresh_target();
        let _ = tail_frames(&seam, &target, &cfg()).await.unwrap();

        // Append more.
        seam.append(3, 0, 0x33);
        seam.append(4, 4, 0x44); // commit

        let out = tail_frames(&seam, &target, &cfg()).await.unwrap();
        match out {
            StreamOutcome::Streamed {
                first_frame,
                last_frame,
                frame_count,
                ..
            } => {
                assert_eq!(first_frame, 3);
                assert_eq!(last_frame, 4);
                assert_eq!(frame_count, 2);
            }
            other => panic!("expected Streamed, got {other:?}"),
        }
        assert_eq!(seam.frame_count(), 4);
        // Frames 1..=4 all retrievable under checkpoint_seq=0, as one batch
        // object per tail call — the first call's object is untouched by the
        // second (disjoint ranges, so no overwrite and no re-upload).
        for (first, last) in [(1u64, 2u64), (3, 4)] {
            target
                .store
                .get(&target.frame_batch_key(0, 0, first, last))
                .await
                .unwrap();
        }
    }

    /// WAL restart bumps checkpoint_seq; the next tail uploads under the new
    /// sequence and reports Restarted.
    #[tokio::test]
    async fn wal_restart_emits_restarted_under_new_sequence() {
        let seam = MockWal::new(4096);
        seam.append(1, 1, 0xAA);
        seam.append(2, 2, 0xBB);
        let target = fresh_target();
        let _ = tail_frames(&seam, &target, &cfg()).await.unwrap();

        // Simulate the engine restarting the WAL (e.g. after a checkpoint).
        seam.restart();
        seam.append(1, 1, 0xCC);

        let out = tail_frames(&seam, &target, &cfg()).await.unwrap();
        match out {
            StreamOutcome::Restarted {
                previous_generation,
                new_generation,
                first_frame,
                last_frame,
                frame_count,
                ..
            } => {
                assert_eq!(previous_generation.checkpoint_seq, 0);
                assert_eq!(new_generation.checkpoint_seq, 1);
                // R858-B19: an in-process restart moves the salt too, so this
                // regime is now caught twice over.
                assert_ne!(previous_generation.salt, new_generation.salt);
                assert_eq!(first_frame, 1);
                assert_eq!(last_frame, 1);
                assert_eq!(frame_count, 1);
            }
            other => panic!("expected Restarted, got {other:?}"),
        }

        // Old seq=0 frames still in place; new seq=1 frame under its own
        // sequence — same key shape, different directory.
        target
            .store
            .get(&target.frame_batch_key(0, 0, 1, 2))
            .await
            .unwrap();
        target
            .store
            .get(&target.frame_batch_key(0, 1, 1, 1))
            .await
            .unwrap();
    }

    /// R858-B19, the property this whole ticket turns on: **a WAL whose salt
    /// changed is a different WAL, even when `checkpoint_seq` did not move.**
    ///
    /// This is the writer-process-restart regime. The last connection closed,
    /// SQLite checkpointed and deleted the `-wal`, and the next writer built a
    /// fresh WAL back at checkpoint-sequence 0. Both tails therefore see
    /// `checkpoint_seq == 0`, which is precisely why the old
    /// `p.checkpoint_seq == current.checkpoint_seq` test resumed at
    /// `last_frame + 1` and spliced frames from two unrelated WALs into one
    /// generation chain — a restore that reported success and produced a stale
    /// or malformed image, with nothing raised anywhere.
    ///
    /// The probe (`examples/foreign_checkpoint_probe.rs -- b f`) drives the
    /// same fold end-to-end against the real system `sqlite3`. This test exists
    /// so the property is pinned here too: the probe needs an upstream sqlite3
    /// binary and half a minute, and a property this load-bearing should fail
    /// in `cargo test` when someone re-derives "the sequence is enough".
    #[tokio::test]
    async fn a_salt_change_is_a_restart_even_when_checkpoint_seq_is_unchanged() {
        let seam = MockWal::new(4096);
        seam.append(1, 1, 0xAA);
        seam.append(2, 2, 0xBB);
        seam.append(3, 3, 0xCC);
        let target = fresh_target();
        let first = tail_frames(&seam, &target, &cfg()).await.unwrap();
        assert!(matches!(first, StreamOutcome::Streamed { .. }), "got {first:?}");

        // The foreign writer restarted: brand-new WAL, unrelated salt, and a
        // checkpoint-sequence that reads 0 both before and after.
        seam.recreate(WalSalt { salt1: 0x9ca8_9e29, salt2: 0x2be5_66fc });
        seam.append(1, 1, 0xDD);
        seam.append(2, 2, 0xEE);
        assert_eq!(
            seam.wal_state().unwrap().checkpoint_seq,
            0,
            "the premise: the sequence did NOT move across the recreate"
        );

        let out = tail_frames(&seam, &target, &cfg()).await.unwrap();
        let StreamOutcome::Restarted {
            previous_generation,
            new_generation,
            first_frame,
            last_frame,
            ..
        } = &out
        else {
            panic!(
                "a recreated WAL must report Restarted — got {out:?}. If this is Streamed with \
                 first_frame 4, the salt check has been removed and two WAL generations are being \
                 spliced into one chain again (R858-B19)"
            );
        };
        assert_eq!(previous_generation.checkpoint_seq, new_generation.checkpoint_seq);
        assert_ne!(
            previous_generation.salt, new_generation.salt,
            "the salt is the only thing that moved, and it is what must be noticed"
        );
        assert_eq!((*first_frame, *last_frame), (1, 2), "re-upload from the top of the new WAL");

        // And the chain that results is refused rather than restored: two
        // generations both starting at frame 1 is not a stream, and restore
        // must say so instead of producing a plausible-looking wrong image.
        let manifests = list_and_parse_generation_manifests(&target).await.unwrap();
        assert_eq!(manifests.len(), 2);
        let err = validate_generation_chain(&manifests).unwrap_err().to_string();
        assert!(
            err.contains("WAL") || err.contains("gap in stream"),
            "restore must refuse the post-recreate chain loudly, got: {err}"
        );
    }

    /// R858-B19 — the narrower door: a WAL recreate that lands **inside** one
    /// `tail_frames` call rather than between two. The pre-drain sample says
    /// generation A, the frames that got read are a mix of A and B, and no
    /// comparison of two watermarks can see it. The call must publish nothing.
    #[tokio::test]
    async fn a_wal_recreated_mid_drain_publishes_nothing() {
        let seam = MockWal::new(4096);
        for n in 1..=3u32 {
            seam.append(n, n, n as u8);
        }
        let target = fresh_target();
        // Read #1 is the pre-drain salt probe; reads #2..#4 are the drain. Flip
        // the WAL underneath us on the second frame of the drain.
        seam.recreate_after_reads(3);

        let err = tail_frames(&seam, &target, &cfg()).await.unwrap_err().to_string();
        assert!(err.contains("recreated while this tail was uploading"), "got: {err}");

        // The sink is untouched: no watermark to resume from, no manifest to
        // restore. Whatever frame objects landed are orphaned and unreferenced.
        assert!(read_watermark(&target.store, &target.watermark_key()).await.unwrap().is_none());
        assert!(list_and_parse_generation_manifests(&target).await.unwrap().is_empty());
    }

    /// R858-B19 — the sidecar half of the same property. A watermark persisted
    /// by a pre-salt writer says nothing about which WAL it names, so the next
    /// tail must restart rather than resume from `last_frame + 1`. Unknown is
    /// not "unchanged".
    #[tokio::test]
    async fn a_saltless_legacy_watermark_forces_a_restart_rather_than_a_resume() {
        let target = fresh_target();
        // Exactly what a pre-R858-B19 writer left behind: five positional
        // fields, no salt pair.
        target
            .store
            .put(&target.watermark_key(), format!("0 2 {} 0 0\n", unix_nanos()).into_bytes().into())
            .await
            .unwrap();

        let seam = MockWal::new(4096);
        seam.append(1, 1, 0xAA);
        seam.append(2, 2, 0xBB);
        seam.append(3, 3, 0xCC);

        let out = tail_frames(&seam, &target, &cfg()).await.unwrap();
        let StreamOutcome::Restarted { previous_generation, first_frame, .. } = &out else {
            panic!(
                "an unverifiable watermark must restart, not resume — got {out:?}. Resuming here \
                 trusts a position in a WAL nobody can show is the same one (R858-B19)"
            );
        };
        assert_eq!(previous_generation.salt, None, "the prior generation is unknown, not equal");
        assert_eq!(*first_frame, 1, "re-upload everything rather than trust the old position");

        // Self-healing: the sidecar now carries a salt, so the very next tail
        // resumes normally. One redundant re-upload, not a permanent restart
        // loop.
        seam.append(4, 4, 0xDD);
        let next = tail_frames(&seam, &target, &cfg()).await.unwrap();
        let StreamOutcome::Streamed { first_frame, .. } = &next else {
            panic!("the salt is recorded now, so this must resume — got {next:?}");
        };
        assert_eq!(*first_frame, 4);
    }

    /// R858-B19 — the restore-side refusal, on the shape a pre-`v4` writer
    /// could actually leave in a bucket: a chain that looks perfectly
    /// contiguous but whose manifests never recorded which WAL they came from.
    /// `checkpoint_seq` agrees, the frame ranges tile, and it is still
    /// unrestorable, because that is exactly what a splice looks like.
    #[test]
    fn validate_chain_refuses_a_multi_generation_chain_with_no_recorded_salt() {
        let err = validate_generation_chain(&[
            mk_manifest_salted("base.db", 4096, 0, None, 1, 5),
            mk_manifest_salted("base.db", 4096, 0, None, 6, 9),
        ])
        .unwrap_err()
        .to_string();
        assert!(err.contains("no WAL salt"), "got: {err}");

        // One generation is not a splice — a legacy single-manifest backup
        // stays restorable, because there is nothing here to prove.
        let chain =
            validate_generation_chain(&[mk_manifest_salted("base.db", 4096, 0, None, 1, 5)])
                .unwrap();
        assert_eq!(chain.generation.salt, None);
        assert_eq!(chain.total_frames, 5);
    }

    /// R858-B19 — and the same refusal when the salts are recorded and
    /// *disagree*: a contiguous frame range across two different WALs. This is
    /// the case `checkpoint_seq` can never catch, since a recreate resets it to
    /// the value it already had.
    #[test]
    fn validate_chain_refuses_a_chain_whose_salt_changes_mid_stream() {
        let other = WalSalt { salt1: 0xea10_2175, salt2: 0xb4ff_221f };
        let err = validate_generation_chain(&[
            mk_manifest_salted("base.db", 4096, 0, Some(CHAIN_SALT), 1, 30),
            mk_manifest_salted("base.db", 4096, 0, Some(other), 31, 57),
        ])
        .unwrap_err()
        .to_string();
        assert!(err.contains("RECREATED mid-stream"), "got: {err}");
    }

    /// R858-B19 — manifest `v4` carries the salt, and the version guards move
    /// with it: a `v4` header without a `wal_salt` is corrupt (not legacy), and
    /// a pre-`v4` header *with* one is forged (not a newer writer).
    #[test]
    fn manifest_v4_round_trips_the_wal_salt_and_guards_both_directions() {
        let batches = [(12u64, 34u64)];
        let salt = WalSalt { salt1: 0xb83c_03f5, salt2: 0x0000_0001 };
        let text = format_generation_manifest(GenerationManifest {
            base_snapshot_key: "backups/snapshots/snapshot-1.db",
            page_size: 4096,
            checkpoint_seq: 7,
            salt: Some(salt),
            first_frame: 12,
            last_frame: 34,
            epoch: 9,
            owner: Some("node-3"),
            frame_batches: &batches,
        });
        assert!(text.starts_with(MANIFEST_HEADER_V4), "got: {text}");
        let parsed = parse_generation_manifest(&text).unwrap();
        assert_eq!(parsed.salt, Some(salt));
        assert_eq!(parsed.checkpoint_seq, 7);
        assert_eq!(parsed.frame_batches, batches);

        let no_salt = text.replace(&format!("wal_salt {}-{}\n", salt.salt1, salt.salt2), "");
        let err = parse_generation_manifest(&no_salt).unwrap_err().to_string();
        assert!(err.contains("missing `wal_salt`"), "got: {err}");

        let forged = text.replace(MANIFEST_HEADER_V4, MANIFEST_HEADER_V3);
        let err = parse_generation_manifest(&forged).unwrap_err().to_string();
        assert!(err.contains("corrupt or hand-edited"), "got: {err}");
    }

    /// R858-B19 — the sidecar's salt pair is read as a pair. Half a pair is a
    /// torn write, and downgrading it to "unknown" would hide that.
    #[tokio::test]
    async fn watermark_sidecar_round_trips_the_salt_and_rejects_half_a_pair() {
        let target = fresh_target();
        let key = target.watermark_key();
        let salt = WalSalt { salt1: 0xd492_ea8a, salt2: 0x7acf_42a3 };
        write_watermark(
            &target.store,
            &key,
            Watermark { checkpoint_seq: 2, last_frame: 11 },
            Some(salt),
            6,
            3,
            None,
        )
        .await
        .unwrap();
        let read = read_watermark(&target.store, &key).await.unwrap().unwrap();
        assert_eq!(read.generation, WalGeneration { checkpoint_seq: 2, salt: Some(salt) });

        target
            .store
            .put(&key, format!("2 11 {} 6 3 {}\n", unix_nanos(), salt.salt1).into_bytes().into())
            .await
            .unwrap();
        let err = read_watermark(&target.store, &key).await.unwrap_err().to_string();
        assert!(err.contains("salt1 but no salt2"), "got: {err}");
    }

    /// Captured frame insert: a mock [`WalInsertSeam`] records every call so
    /// tests can assert ordering and content without a real turso connection.
    struct MockInsertSeam {
        log: RefCell<Vec<MockInsertEvent>>,
    }
    #[derive(Debug, Clone, PartialEq, Eq)]
    enum MockInsertEvent {
        Begin,
        Frame { frame_no: u64, page_no: u32, db_size: u32 },
        End { force_commit: bool },
    }
    impl MockInsertSeam {
        fn new() -> Self {
            Self { log: RefCell::new(Vec::new()) }
        }
        fn events(&self) -> Vec<MockInsertEvent> {
            self.log.borrow().clone()
        }
    }
    impl WalInsertSeam for MockInsertSeam {
        fn wal_insert_begin(&self) -> Result<()> {
            self.log.borrow_mut().push(MockInsertEvent::Begin);
            Ok(())
        }
        fn wal_insert_frame(&self, frame_no: u64, frame: &[u8]) -> Result<()> {
            let page_no = u32::from_be_bytes(frame[0..4].try_into().unwrap());
            let db_size = u32::from_be_bytes(frame[4..8].try_into().unwrap());
            self.log.borrow_mut().push(MockInsertEvent::Frame { frame_no, page_no, db_size });
            Ok(())
        }
        fn wal_insert_end(&self, force_commit: bool) -> Result<()> {
            self.log.borrow_mut().push(MockInsertEvent::End { force_commit });
            Ok(())
        }
    }

    fn mk_manifest(
        base: &str,
        page_size: usize,
        checkpoint_seq: u32,
        first_frame: u64,
        last_frame: u64,
    ) -> OwnedGenerationManifest {
        mk_manifest_salted(
            base,
            page_size,
            checkpoint_seq,
            Some(CHAIN_SALT),
            first_frame,
            last_frame,
        )
    }

    /// R858-B19: the salt every `mk_manifest` fixture shares, so a chain built
    /// from them is one generation unless a test deliberately says otherwise.
    const CHAIN_SALT: WalSalt = WalSalt { salt1: 0x1109_ca5e, salt2: 0x7acf_42a3 };

    /// R858-B19: `mk_manifest` with the generation salt spelled out — `None`
    /// reproduces a manifest written by a pre-`v4` writer.
    fn mk_manifest_salted(
        base: &str,
        page_size: usize,
        checkpoint_seq: u32,
        salt: Option<WalSalt>,
        first_frame: u64,
        last_frame: u64,
    ) -> OwnedGenerationManifest {
        OwnedGenerationManifest {
            base_snapshot_key: base.to_string(),
            page_size,
            checkpoint_seq,
            salt,
            first_frame,
            last_frame,
            epoch: 0,
            owner: None,
            // Chain validation is about frame ranges, not object layout, so
            // these fixtures stay on the pre-R761-F2 shape — which also keeps
            // them exercising the legacy read path.
            frame_batches: Vec::new(),
        }
    }

    /// validate_generation_chain accepts a single well-formed manifest.
    #[test]
    fn validate_chain_accepts_single_generation() {
        let chain = validate_generation_chain(&[mk_manifest("base.db", 4096, 0, 1, 5)]).unwrap();
        assert_eq!(chain.base_snapshot_key, "base.db");
        assert_eq!(chain.page_size, 4096);
        assert_eq!(chain.generation.checkpoint_seq, 0);
        assert_eq!(chain.total_frames, 5);
    }

    /// validate_generation_chain accepts a contiguous multi-generation chain
    /// and sums the frame count across generations.
    #[test]
    fn validate_chain_accepts_contiguous_multi_generation() {
        let chain = validate_generation_chain(&[
            mk_manifest("base.db", 4096, 3, 1, 5),
            mk_manifest("base.db", 4096, 3, 6, 9),
            mk_manifest("base.db", 4096, 3, 10, 12),
        ])
        .unwrap();
        assert_eq!(chain.generation.checkpoint_seq, 3);
        assert_eq!(chain.total_frames, 12);
    }

    /// Empty chain is the "no generations under prefix" signal.
    #[test]
    fn validate_chain_rejects_empty() {
        assert!(validate_generation_chain(&[]).is_err());
    }

    /// A gap between manifest ranges is corruption — fail loudly.
    #[test]
    fn validate_chain_rejects_gap() {
        let err = validate_generation_chain(&[
            mk_manifest("base.db", 4096, 0, 1, 5),
            mk_manifest("base.db", 4096, 0, 7, 9), // missing frame 6
        ])
        .unwrap_err();
        assert!(format!("{err}").contains("gap in stream"), "err was {err}");
    }

    /// A chain that does not start at frame 1 means the base+chain don't match
    /// (some early frames were never uploaded, or the chain was truncated by a
    /// retention sweep). Refuse.
    #[test]
    fn validate_chain_rejects_non_one_start() {
        let err = validate_generation_chain(&[mk_manifest("base.db", 4096, 0, 5, 9)]).unwrap_err();
        assert!(format!("{err}").contains("gap in stream"), "err was {err}");
    }

    /// A WAL restart between generations means the source folded the WAL into
    /// main; the post-restart frames do not replay onto a pre-restart base.
    /// Refuse and direct the caller at a fresh tier-1a snapshot.
    #[test]
    fn validate_chain_rejects_restart() {
        let err = validate_generation_chain(&[
            mk_manifest("base.db", 4096, 0, 1, 5),
            mk_manifest("base.db", 4096, 1, 1, 3), // restart, new seq
        ])
        .unwrap_err();
        let s = format!("{err}");
        assert!(s.contains("WAL restart"), "err was {s}");
        assert!(s.contains("fresh tier-1a snapshot"), "err was {s}");
    }

    /// Generations referencing different bases mean the chain mixed sinks /
    /// the base was rotated mid-stream. Refuse.
    #[test]
    fn validate_chain_rejects_base_mismatch() {
        let err = validate_generation_chain(&[
            mk_manifest("base-A.db", 4096, 0, 1, 5),
            mk_manifest("base-B.db", 4096, 0, 6, 9),
        ])
        .unwrap_err();
        assert!(format!("{err}").contains("chain spans bases"), "err was {err}");
    }

    /// Page-size mismatch across manifests is a corrupt-manifest signal.
    #[test]
    fn validate_chain_rejects_page_size_mismatch() {
        let err = validate_generation_chain(&[
            mk_manifest("base.db", 4096, 0, 1, 5),
            mk_manifest("base.db", 8192, 0, 6, 9),
        ])
        .unwrap_err();
        assert!(format!("{err}").contains("page_size"), "err was {err}");
    }

    /// replay_frames_into walks every manifest's range in (seq, frame_no) order
    /// and pushes frames into the insert seam. The mock seam records the call
    /// log so we can verify ordering and frame content.
    #[tokio::test]
    async fn replay_walks_manifests_in_frame_order() {
        // Stage two generations' worth of frames into the object store via
        // tail_frames, then point replay at the resulting manifests.
        let seam = MockWal::new(4096);
        seam.append(1, 0, 0xAA);
        seam.append(2, 2, 0xBB); // commit
        let target = fresh_target();
        let _ = tail_frames(&seam, &target, &cfg()).await.unwrap();
        seam.append(3, 0, 0xCC);
        seam.append(4, 4, 0xDD); // commit
        let _ = tail_frames(&seam, &target, &cfg()).await.unwrap();

        let manifests = list_and_parse_generation_manifests(&target).await.unwrap();
        let insert = MockInsertSeam::new();
        let total = replay_frames_into(&target, &insert, &manifests)
            .await
            .unwrap();
        assert_eq!(total, 4);

        let events = insert.events();
        assert_eq!(events.len(), 4, "begin/end are the caller's responsibility");
        for (i, ev) in events.iter().enumerate() {
            let want_frame_no = (i + 1) as u64;
            let want_page = want_frame_no as u32;
            let want_db_size = if want_frame_no.is_multiple_of(2) { want_frame_no as u32 } else { 0 };
            assert_eq!(
                ev,
                &MockInsertEvent::Frame {
                    frame_no: want_frame_no,
                    page_no: want_page,
                    db_size: want_db_size,
                }
            );
        }
    }

    /// R761-F2 read compatibility, and the reason `frame_batches` being empty
    /// has to MEAN something rather than merely be absent: a sink written by a
    /// pre-batching writer — per-frame objects, a v2 manifest — still replays,
    /// and a chain that straddles the change replays as one stream. Written
    /// here by hand at the object level, because the writer that produced this
    /// shape no longer exists to produce it.
    #[tokio::test]
    async fn a_chain_straddling_the_batching_change_replays_as_one_stream() {
        let target = fresh_target();
        let frame_size = WAL_FRAME_HEADER_SIZE + 4096;
        let frame = |frame_no: u64, page_no: u32, db_size: u32| {
            let mut b = vec![0u8; frame_size];
            b[0..4].copy_from_slice(&page_no.to_be_bytes());
            b[4..8].copy_from_slice(&db_size.to_be_bytes());
            b[WAL_FRAME_HEADER_SIZE..].fill(frame_no as u8);
            b
        };

        // Generation 1: the old layout — one object per frame, v2 manifest
        // with no batch list.
        for frame_no in 1..=2u64 {
            target
                .store
                .put(
                    &target.frame_key(0, 0, frame_no),
                    frame(frame_no, frame_no as u32, frame_no as u32).into(),
                )
                .await
                .unwrap();
        }
        let legacy = format_generation_manifest(GenerationManifest {
            base_snapshot_key: "b.db",
            page_size: 4096,
            checkpoint_seq: 0,
            salt: None,
            first_frame: 1,
            last_frame: 2,
            epoch: 0,
            owner: None,
            frame_batches: &[],
        });
        assert!(legacy.starts_with("TURSO-BACKUP STREAM v2\n"), "{legacy}");
        target
            .store
            .put(&target.generation_key(1), legacy.into_bytes().into())
            .await
            .unwrap();

        // Generation 2: the new layout, batched, appended to the same chain.
        let mut batch = frame(3, 3, 0);
        batch.extend_from_slice(&frame(4, 4, 4));
        target
            .store
            .put(&target.frame_batch_key(0, 0, 3, 4), batch.into())
            .await
            .unwrap();
        let batched = format_generation_manifest(GenerationManifest {
            base_snapshot_key: "b.db",
            page_size: 4096,
            checkpoint_seq: 0,
            salt: None,
            first_frame: 3,
            last_frame: 4,
            epoch: 0,
            owner: None,
            frame_batches: &[(3, 4)],
        });
        target
            .store
            .put(&target.generation_key(2), batched.into_bytes().into())
            .await
            .unwrap();

        let manifests = list_and_parse_generation_manifests(&target).await.unwrap();
        assert!(manifests[0].frame_batches.is_empty(), "gen 1 is the legacy layout");
        assert_eq!(manifests[1].frame_batches, vec![(3, 4)]);

        // R858-B19 CHANGED THIS ASSERTION, deliberately. It used to read
        // `validate_generation_chain(&manifests).expect("layout is not a chain
        // property")`. Layout still is not a chain property — that claim is
        // what the replay assertion below tests, and it still holds. What
        // changed is PROVENANCE: this fixture is a two-generation chain in
        // which neither manifest records which WAL its frames came from,
        // because both predate the `wal_salt` field. That is byte-for-byte the
        // shape a WAL recreate produced (probe F: contiguous ranges, one
        // checkpoint_seq, two different WALs), so restore can no longer tell
        // this chain from a spliced one and must refuse rather than hand back a
        // plausible wrong image. A chain this old needs a fresh tier-1a
        // snapshot; one generation of it would still restore.
        let err = validate_generation_chain(&manifests).unwrap_err().to_string();
        assert!(err.contains("no WAL salt"), "got: {err}");

        let insert = MockInsertSeam::new();
        assert_eq!(replay_frames_into(&target, &insert, &manifests).await.unwrap(), 4);
        assert_eq!(
            insert.events(),
            (1..=4u64)
                .map(|n| MockInsertEvent::Frame {
                    frame_no: n,
                    page_no: n as u32,
                    db_size: if n == 3 { 0 } else { n as u32 },
                })
                .collect::<Vec<_>>(),
            "both layouts deliver the same frames in the same order"
        );
    }

    /// restore_latest_stream errors loudly when there are no manifests under
    /// the prefix — the caller should be using tier-1a restore instead.
    #[tokio::test]
    async fn restore_errors_when_no_generations() {
        let target = fresh_target();
        let dest = std::env::temp_dir().join(format!(
            "turso-backup-restore-empty-{}-{}.db",
            std::process::id(),
            unix_nanos()
        ));
        let err = restore_latest_stream(&target, dest.to_str().unwrap())
            .await
            .unwrap_err();
        assert!(format!("{err}").contains("no generation manifests"), "err was {err}");
        // The dest file is never written when validation fails up front.
        assert!(!dest.exists());
    }

    /// A frame uploaded with a wrong byte length corrupts the chain; restore's
    /// replay path catches it before touching the destination's WAL.
    #[tokio::test]
    async fn replay_rejects_wrong_size_frame() {
        let seam = MockWal::new(4096);
        seam.append(1, 1, 0xAA);
        let target = fresh_target();
        let _ = tail_frames(&seam, &target, &cfg()).await.unwrap();

        // Overwrite the one uploaded frame object with garbage of the wrong
        // length.
        target
            .store
            .put(&target.frame_batch_key(0, 0, 1, 1), b"too short".to_vec().into())
            .await
            .unwrap();

        let manifests = list_and_parse_generation_manifests(&target).await.unwrap();
        let insert = MockInsertSeam::new();
        let err = replay_frames_into(&target, &insert, &manifests)
            .await
            .unwrap_err();
        assert!(format!("{err}").contains("expected"), "err was {err}");
    }

    /// Live-DB end-to-end fixture: a temp db path that cleans itself up
    /// (including `-wal` / `-shm` sidecars).
    struct TempDb(std::path::PathBuf);
    impl TempDb {
        fn new(tag: &str) -> Self {
            TempDb(std::env::temp_dir().join(format!(
                "turso-backup-stream-{tag}-{}-{}.db",
                std::process::id(),
                unix_nanos(),
            )))
        }
        fn path(&self) -> &str {
            self.0.to_str().unwrap()
        }
    }
    impl Drop for TempDb {
        fn drop(&mut self) {
            for sfx in ["", "-wal", "-shm"] {
                let _ = std::fs::remove_file(format!("{}{sfx}", self.0.display()));
            }
        }
    }

    async fn seed_rows(path: &str, start: i64, count: i64) {
        let db = turso::Builder::new_local(path).build().await.unwrap();
        let conn = db.connect().unwrap();
        conn.execute("CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, v TEXT)", ())
            .await
            .unwrap();
        conn.execute("BEGIN", ()).await.unwrap();
        for i in start..start + count {
            conn.execute("INSERT INTO t (id, v) VALUES (?, ?)", (i, format!("v{i}")))
                .await
                .unwrap();
        }
        conn.execute("COMMIT", ()).await.unwrap();
    }

    async fn checkpoint_truncate(path: &str) {
        let db = turso::Builder::new_local(path).build().await.unwrap();
        let conn = db.connect().unwrap();
        let mut rows = conn.query("PRAGMA wal_checkpoint(TRUNCATE)", ()).await.unwrap();
        while rows.next().await.unwrap().is_some() {}
    }

    async fn count_rows(path: &str) -> i64 {
        let db = turso::Builder::new_local(path).build().await.unwrap();
        let conn = db.connect().unwrap();
        let mut r = conn.query("SELECT COUNT(*) FROM t", ()).await.unwrap();
        let row = r.next().await.unwrap().unwrap();
        row.get::<i64>(0).unwrap()
    }

    /// End-to-end against a real turso DB and the real `CoreWalSeam`:
    /// seed → snapshot → checkpoint (resets WAL, bumps checkpoint_seq) →
    /// append more rows (creates frames under the new seq) → tail → restore →
    /// row count on the dest matches the post-append source.
    ///
    /// This is the live-DB ping-pong R005-F2's handoff named: it exercises
    /// CoreWalSeam's `wal_state` / `wal_get_frame` AND the new
    /// `wal_insert_begin` / `wal_insert_frame` / `wal_insert_end` impls, plus
    /// the full manifest discovery + replay pipeline.
    #[tokio::test]
    async fn live_db_seed_snapshot_tail_restore_round_trips() {
        let src = TempDb::new("src");
        let dest = TempDb::new("dest");

        // Seed 50 rows. These end up in the base snapshot via VACUUM INTO.
        seed_rows(src.path(), 0, 50).await;

        let target = BackupTarget {
            store: Arc::new(InMemory::new()),
            prefix: "backups".into(),
        };
        let base_key = match crate::snapshot::snapshot_and_upload(src.path(), &target)
            .await
            .unwrap()
        {
            crate::snapshot::SnapshotOutcome::Uploaded { key, .. } => key,
            other => panic!("expected Uploaded base snapshot, got {other:?}"),
        };

        // Checkpoint to fold prior WAL into main and reset the WAL header —
        // any subsequent writes land in fresh frames under a new
        // checkpoint_seq. This mirrors how a real orchestrator would hand off
        // from tier-1a (full snapshot) into tier-2 streaming.
        checkpoint_truncate(src.path()).await;

        // Append 25 more rows — these become the WAL frames we stream.
        seed_rows(src.path(), 1000, 25).await;

        // Tail the new frames into the sink via the real CoreWalSeam.
        {
            let seam = CoreWalSeam::open(src.path()).unwrap();
            let cfg = StreamConfig {
                base_snapshot_key: &base_key,
                page_size: 4096,
                backpressure: BackpressureConfig::default(),
                rpo_target: None,
                epoch: 0,
                owner: None,
                pointer_generation: 0,
            };
            let outcome = tail_frames(&seam, &target, &cfg).await.unwrap();
            match outcome {
                StreamOutcome::Streamed { frame_count, .. } => {
                    assert!(frame_count > 0, "expected at least one frame uploaded");
                }
                StreamOutcome::Restarted { frame_count, .. } => {
                    // Acceptable when this is the first tail after a checkpoint:
                    // there's a prior implicit watermark via the WAL-restart bit.
                    assert!(frame_count > 0);
                }
                other => panic!("expected Streamed/Restarted, got {other:?}"),
            }
        } // drop seam (and its turso_core connection) before restore opens its own.

        // Restore into a fresh dest. Replay must produce the post-append state.
        let outcome = restore_latest_stream(&target, dest.path()).await.unwrap();
        assert_eq!(outcome.base_snapshot_key, base_key);
        assert!(outcome.frames_replayed > 0);
        assert!(outcome.generation_count >= 1);

        // The restored DB is openable via vanilla turso and has the right rows.
        // (snapshot::restore_latest's tests already cover the vanilla-sqlite3
        // exit ramp; here the contract is "turso reopens" since wal_insert_*
        // is the only consumer side that knows the engine's frame layout.)
        let restored_rows = count_rows(dest.path()).await;
        assert_eq!(restored_rows, 75, "expected 50 (base) + 25 (replayed) = 75");
    }

    /// Crash-consistency: the last frames captured by `tail_frames` are
    /// uncommitted (mid-transaction, `db_size == 0`). Restore must drop that
    /// suffix and end up at the previous commit's state — `wal_insert_end`
    /// with `force_commit = false` is the engine knob that does this.
    ///
    /// We can't easily force turso to commit-then-leak-a-partial-write in a
    /// unit test, so we drive the seam directly: a single `MockInsertSeam`-
    /// recorded restore would show the begin/end protocol, but to verify the
    /// engine actually truncates we need the real `CoreWalSeam`. The
    /// `manifest_with_uncommitted_tail_rolls_back_via_insert_end` test below
    /// stages frame bytes in the store by hand and aims them at a real DB.
    #[tokio::test]
    async fn manifest_with_uncommitted_tail_rolls_back_via_insert_end() {
        // Seed two rows via real turso so we have committed page 1 + page 2.
        // Then capture the WAL frames from the source. Then stage a *fake*
        // extra frame whose db_size = 0 (mid-transaction) in the sink, and
        // append it to the manifest range. Restore should drop that fake
        // frame and leave the dest at the 2-row state.
        let src = TempDb::new("ccsrc");
        let dest = TempDb::new("ccdest");
        seed_rows(src.path(), 0, 2).await;
        let target = BackupTarget {
            store: Arc::new(InMemory::new()),
            prefix: "backups".into(),
        };
        let base_key = match crate::snapshot::snapshot_and_upload(src.path(), &target)
            .await
            .unwrap()
        {
            crate::snapshot::SnapshotOutcome::Uploaded { key, .. } => key,
            other => panic!("expected Uploaded, got {other:?}"),
        };
        checkpoint_truncate(src.path()).await;
        seed_rows(src.path(), 100, 1).await; // one more row → at least one commit frame.

        let (checkpoint_seq, last_committed_frame, frame_size) = {
            let seam = CoreWalSeam::open(src.path()).unwrap();
            let cfg = StreamConfig {
                base_snapshot_key: &base_key,
                page_size: 4096,
                backpressure: BackpressureConfig::default(),
                rpo_target: None,
                epoch: 0,
                owner: None,
                pointer_generation: 0,
            };
            let _ = tail_frames(&seam, &target, &cfg).await.unwrap();
            let w = seam.wal_state().unwrap();
            (w.checkpoint_seq, w.last_frame, WAL_FRAME_HEADER_SIZE + 4096)
        };

        // Manually upload one extra "uncommitted" frame: copy the last
        // committed frame's bytes but zero db_size in the header. This
        // simulates tail_frames having captured a mid-transaction tail.
        // R761-F2: the frames live in one batch object, so take the last
        // frame's slice out of it rather than fetching a per-frame key.
        let batch_key = target.frame_batch_key(0, checkpoint_seq, 1, last_committed_frame);
        let batch = target
            .store
            .get(&batch_key)
            .await
            .unwrap()
            .bytes()
            .await
            .unwrap();
        let at = (last_committed_frame as usize - 1) * frame_size;
        let mut tail_bytes = batch[at..at + frame_size].to_vec();
        // Zero the big-endian db_size at offset 4..8 to mark this as non-commit.
        tail_bytes[4..8].copy_from_slice(&0u32.to_be_bytes());
        // Repad to ensure exact length — paranoia.
        assert_eq!(tail_bytes.len(), frame_size);
        let phantom_frame_no = last_committed_frame + 1;
        target
            .store
            .put(
                &target.frame_batch_key(0, checkpoint_seq, phantom_frame_no, phantom_frame_no),
                tail_bytes.into(),
            )
            .await
            .unwrap();

        // Extend the latest generation manifest to claim the phantom frame.
        let mut keys: Vec<_> = target
            .store
            .list_with_delimiter(Some(&join_key(&target.prefix, "generations")))
            .await
            .unwrap()
            .objects
            .into_iter()
            .map(|o| o.location)
            .collect();
        keys.sort();
        let last_manifest_key = keys.last().unwrap().clone();
        let bytes = target
            .store
            .get(&last_manifest_key)
            .await
            .unwrap()
            .bytes()
            .await
            .unwrap();
        let mut m = parse_generation_manifest(&String::from_utf8_lossy(&bytes)).unwrap();
        m.last_frame = phantom_frame_no;
        // The phantom frame went up as its own single-frame batch object, so
        // the manifest's index has to name it too — the range alone is no
        // longer enough to find a frame (R761-F2).
        m.frame_batches.push((phantom_frame_no, phantom_frame_no));
        let new_text = format_generation_manifest(GenerationManifest {
            base_snapshot_key: &m.base_snapshot_key,
            page_size: m.page_size,
            checkpoint_seq: m.checkpoint_seq,
            salt: None,
            first_frame: m.first_frame,
            last_frame: m.last_frame,
            epoch: m.epoch,
            owner: m.owner.as_deref(),
            frame_batches: &m.frame_batches,
        });
        target
            .store
            .put(&last_manifest_key, new_text.into_bytes().into())
            .await
            .unwrap();

        // Restore. The phantom uncommitted frame must be dropped by
        // wal_insert_end(false); the row count reflects the last *commit*.
        let _ = restore_latest_stream(&target, dest.path()).await.unwrap();
        let restored = count_rows(dest.path()).await;
        assert_eq!(restored, 3, "expected 2 (base) + 1 (committed) — phantom rolled back");
    }

    /// `replay_wal_onto_main` with an empty WAL returns the main bytes
    /// untouched — the main file alone is the committed image.
    #[test]
    fn replay_empty_wal_returns_main_unchanged() {
        let seam = MockWal::new(4096);
        let mut main = vec![0u8; 4096 * 3];
        main[0..16].copy_from_slice(b"SQLite format 3\0");
        let img = replay_wal_onto_main(&seam, main.clone(), 4096).unwrap();
        assert_eq!(img, main);
    }

    /// A single commit frame updates its page slot and the final image is
    /// sized to the commit's `db_size`. Unrelated pages in `main` are left
    /// in place.
    #[test]
    fn replay_single_commit_frame_applies_page() {
        let seam = MockWal::new(4096);
        // Frame 1: page 2, commit at db_size = 3 pages.
        seam.append(2, 3, 0xCC);
        let mut main = vec![0u8; 4096 * 3];
        main[0..16].copy_from_slice(b"SQLite format 3\0");

        let img = replay_wal_onto_main(&seam, main, 4096).unwrap();
        assert_eq!(img.len(), 4096 * 3);
        // Page 1 untouched (still has the magic + zeros).
        assert!(img.starts_with(b"SQLite format 3\0"));
        // Page 2 = 0xCC fill from the WAL frame.
        assert!(img[4096..4096 * 2].iter().all(|&b| b == 0xCC));
        // Page 3 untouched.
        assert!(img[4096 * 2..].iter().all(|&b| b == 0));
    }

    /// Uncommitted frames past the last commit are dropped and the image is
    /// truncated to the last commit's `db_size` — the crash-consistency story
    /// matching restore's `wal_insert_end(false)`.
    #[test]
    fn replay_drops_uncommitted_tail() {
        let seam = MockWal::new(4096);
        seam.append(1, 0, 0x11); // mid-txn
        seam.append(2, 2, 0x22); // commit at db_size = 2
        seam.append(3, 0, 0x33); // uncommitted (dropped)
        seam.append(4, 0, 0x44); // uncommitted (dropped)

        let main = vec![0u8; 4096 * 4]; // pre-grown so a buggy replay would keep junk
        let img = replay_wal_onto_main(&seam, main, 4096).unwrap();
        assert_eq!(img.len(), 4096 * 2, "image must be truncated to db_size=2");
        assert!(img[0..4096].iter().all(|&b| b == 0x11));
        assert!(img[4096..].iter().all(|&b| b == 0x22));
    }

    /// Multi-commit chain: the full commit prefix is applied and the image is
    /// grown to the final commit's `db_size`. Intermediate uncommitted frames
    /// between commits are also applied (they are part of the committed
    /// suffix once a later frame in the same batch becomes a commit).
    #[test]
    fn replay_applies_full_commit_prefix_and_grows_image() {
        let seam = MockWal::new(4096);
        seam.append(1, 0, 0xAA);
        seam.append(2, 2, 0xBB); // first commit
        seam.append(3, 0, 0xCC);
        seam.append(4, 4, 0xDD); // second commit

        let main = vec![0u8; 4096 * 2];
        let img = replay_wal_onto_main(&seam, main, 4096).unwrap();
        assert_eq!(img.len(), 4096 * 4, "image grown to db_size=4");
        assert!(img[0..4096].iter().all(|&b| b == 0xAA));
        assert!(img[4096..4096 * 2].iter().all(|&b| b == 0xBB));
        assert!(img[4096 * 2..4096 * 3].iter().all(|&b| b == 0xCC));
        assert!(img[4096 * 3..].iter().all(|&b| b == 0xDD));
    }

    /// WAL has frames but none are committed (writer mid-transaction at the
    /// instant we sampled). The main file alone is the image; the uncommitted
    /// suffix is dropped wholesale.
    #[test]
    fn replay_returns_main_when_only_uncommitted_frames_present() {
        let seam = MockWal::new(4096);
        seam.append(1, 0, 0x11);
        seam.append(2, 0, 0x22);
        let mut main = vec![0u8; 4096 * 3];
        main[0..16].copy_from_slice(b"SQLite format 3\0");
        let img = replay_wal_onto_main(&seam, main.clone(), 4096).unwrap();
        assert_eq!(img, main);
    }

    /// End-to-end against a real turso DB and `CoreWalSeam`: seed, fold the
    /// first batch into main via TRUNCATE, then seed more rows so the WAL has
    /// uncheckpointed committed frames. `raw_consistent_copy_live` must
    /// reproduce the full row count without itself calling TRUNCATE — that is
    /// the live-writer contract a TRUNCATE-busy concurrent writer would
    /// otherwise block.
    #[tokio::test]
    async fn live_db_consistent_copy_without_truncate_round_trips() {
        let src = TempDb::new("live-cc-src");
        seed_rows(src.path(), 0, 25).await;
        checkpoint_truncate(src.path()).await; // first batch into main, WAL reset
        seed_rows(src.path(), 1000, 10).await; // second batch lives in WAL

        // No TRUNCATE here — this is the live-writer path.
        let image = raw_consistent_copy_live(src.path(), 4096).await.unwrap();
        assert!(image.starts_with(b"SQLite format 3\0"));

        // Write the image to a fresh path (no -wal sidecar — the replay
        // folded WAL in) and re-open to confirm all 35 rows are present.
        let restored = TempDb::new("live-cc-restored");
        std::fs::write(restored.path(), &image).unwrap();
        let rows = count_rows(restored.path()).await;
        assert_eq!(
            rows, 35,
            "expected 25 (checkpointed) + 10 (replayed from WAL)"
        );
    }

    // ── R858-B18: reading a source nobody will lock for us ────────────────
    //
    // The property under test throughout: the copy is either a validated
    // point in time or a refusal, never a plausible wrong image.

    /// The two salt readers must agree. [`read_wal_salt`] pulls it out of frame
    /// 1 through the [`WalSeam`] trait (the only path a `from_conn` seam has);
    /// [`WalFileHeader::read`] pulls it out of the 32-byte `-wal` header with no
    /// engine at all. They are separate code paths against separate byte
    /// offsets, and R858-B18 depends on them naming the same generation — if
    /// they could disagree, validation would compare a salt the streamer never
    /// recorded.
    #[tokio::test]
    async fn wal_file_header_salt_matches_the_salt_read_through_the_seam() {
        let src = TempDb::new("b18-salt-agree");
        seed_rows(src.path(), 0, 5).await;

        let seam = CoreWalSeam::open_reader(src.path()).unwrap();
        let state = seam.wal_state().unwrap();
        assert!(state.last_frame > 0, "seed should leave frames in the WAL");
        let via_seam = read_wal_salt(&seam, 4096, state.last_frame).unwrap().unwrap();
        drop(seam);

        let via_file = WalFileHeader::read(src.path()).unwrap().unwrap();
        assert_eq!(via_file.salt, via_seam, "frame 1 carries the WAL header's salt verbatim");
        assert_eq!(
            via_file.checkpoint_seq, state.checkpoint_seq,
            "and the on-disk sequence is the one the engine reports"
        );
    }

    /// A `-wal` shorter than one header names no generation — an honest
    /// unknown, not an error (same posture as `WalGeneration { salt: None }`).
    #[test]
    fn wal_file_header_parse_needs_a_whole_header() {
        assert!(WalFileHeader::parse(&[0u8; WalFileHeader::SIZE - 1]).is_none());
        let mut hdr = [0u8; WalFileHeader::SIZE];
        hdr[8..12].copy_from_slice(&4096u32.to_be_bytes());
        hdr[12..16].copy_from_slice(&7u32.to_be_bytes());
        hdr[16..20].copy_from_slice(&0xb83c_03f5u32.to_be_bytes());
        hdr[20..24].copy_from_slice(&0x7acf_42a3u32.to_be_bytes());
        let parsed = WalFileHeader::parse(&hdr).unwrap();
        assert_eq!(parsed.page_size, 4096);
        assert_eq!(parsed.checkpoint_seq, 7);
        assert_eq!(parsed.salt, WalSalt { salt1: 0xb83c_03f5, salt2: 0x7acf_42a3 });
    }

    /// The central asymmetry of [`SourceFingerprint::stable_across`]: a plain
    /// append is NOT movement (frames `1..=max_frame` are immutable within a
    /// generation, so our image is merely an earlier point in time), while a
    /// checkpoint IS (it rewrites the main file our copy already read).
    ///
    /// Getting this backwards is not a small error in either direction: treat
    /// an append as movement and every copy of a database that is actually in
    /// use is refused; treat a checkpoint as harmless and we hand back spliced
    /// state that passes `integrity_check`.
    #[tokio::test]
    async fn fingerprint_ignores_an_append_and_catches_a_checkpoint() {
        let src = TempDb::new("b18-fingerprint");
        seed_rows(src.path(), 0, 5).await;

        let before = SourceFingerprint::read(src.path()).unwrap();
        seed_rows(src.path(), 100, 5).await;
        let appended = SourceFingerprint::read(src.path()).unwrap();
        assert!(
            appended.wal_len > before.wal_len,
            "the append must actually have grown the WAL, or this proves nothing \
             (before {}B, after {}B)",
            before.wal_len,
            appended.wal_len
        );
        assert!(
            before.stable_across(&appended),
            "an append is not movement: {} -> {}",
            before.describe(),
            appended.describe()
        );

        checkpoint_truncate(src.path()).await;
        let folded = SourceFingerprint::read(src.path()).unwrap();
        assert!(
            !before.stable_across(&folded),
            "a checkpoint IS movement and must be caught: {} -> {}",
            before.describe(),
            folded.describe()
        );
    }

    /// The protocol accepts when the source holds still, and the accepted value
    /// is the one the attempt produced.
    #[tokio::test]
    async fn validation_accepts_a_quiescent_source() {
        let src = TempDb::new("b18-quiescent");
        seed_rows(src.path(), 0, 3).await;
        let got = validated_against_source(src.path(), "test read", || async { Ok(41 + 1) })
            .await
            .unwrap();
        assert_eq!(got, 42);
    }

    /// A source that moves under every attempt yields a REFUSAL, not a value —
    /// and the message names both samples so an operator can see what moved.
    #[tokio::test]
    async fn validation_refuses_when_the_source_moves_under_every_attempt() {
        let src = TempDb::new("b18-moving");
        seed_rows(src.path(), 0, 3).await;
        let path = src.path().to_string();
        let attempts = std::cell::Cell::new(0u32);
        let err = validated_against_source(src.path(), "test read", || {
            // Grow the main file inside the attempt window, which is exactly
            // the shape of a foreign checkpoint folding pages into it.
            attempts.set(attempts.get() + 1);
            let path = path.clone();
            async move {
                let mut f = std::fs::OpenOptions::new().append(true).open(&path)?;
                std::io::Write::write_all(&mut f, &[0u8; 4096])?;
                Ok(())
            }
        })
        .await
        .unwrap_err();
        assert_eq!(
            attempts.get(),
            COPY_VALIDATION_ATTEMPTS,
            "every attempt in the budget must be spent before refusing"
        );
        let msg = format!("{err:#}");
        assert!(msg.contains("refusing a test read"), "{msg}");
        assert!(msg.contains("before and"), "message must name both samples: {msg}");
    }

    /// An attempt that fails against a source that did NOT move is a real
    /// error, not a race — it is returned as itself on the first attempt rather
    /// than retried and then reported as concurrency.
    #[tokio::test]
    async fn validation_surfaces_a_real_error_without_burning_the_budget() {
        let src = TempDb::new("b18-real-error");
        seed_rows(src.path(), 0, 3).await;
        let attempts = std::cell::Cell::new(0u32);
        let err = validated_against_source(src.path(), "test read", || {
            attempts.set(attempts.get() + 1);
            async { Err::<(), _>(anyhow::anyhow!("page 3 checksum mismatch")) }
        })
        .await
        .unwrap_err();
        assert_eq!(attempts.get(), 1, "a stable source means retrying cannot help");
        let msg = format!("{err:#}");
        assert!(msg.contains("did NOT move"), "{msg}");
        assert!(msg.contains("page 3 checksum mismatch"), "{msg}");
    }

    /// The reader open is the one the backup path uses, and it must produce the
    /// same watermark the writable open does. (Cross-process non-exclusivity —
    /// the point of the flag — is measured in `examples/foreign_checkpoint_probe.rs`
    /// probes G2r/G3c, since it needs a second process.)
    #[tokio::test]
    async fn read_only_seam_reports_the_same_watermark_as_the_writable_one() {
        let src = TempDb::new("b18-reader-watermark");
        seed_rows(src.path(), 0, 4).await;
        let writable = CoreWalSeam::open(src.path()).unwrap().wal_state().unwrap();
        let reader = CoreWalSeam::open_reader(src.path()).unwrap().wal_state().unwrap();
        assert_eq!(writable, reader);
    }

    /// A second call to `raw_consistent_copy_live` after more writes captures
    /// the new state — the primitive is callable repeatedly without per-call
    /// setup, mirroring how a snapshot loop would drive it.
    #[tokio::test]
    async fn live_db_consistent_copy_reflects_new_writes() {
        let src = TempDb::new("live-cc-incr-src");
        seed_rows(src.path(), 0, 5).await;
        let img1 = raw_consistent_copy_live(src.path(), 4096).await.unwrap();
        let r1 = TempDb::new("live-cc-incr-r1");
        std::fs::write(r1.path(), &img1).unwrap();
        assert_eq!(count_rows(r1.path()).await, 5);

        seed_rows(src.path(), 100, 7).await;
        let img2 = raw_consistent_copy_live(src.path(), 4096).await.unwrap();
        let r2 = TempDb::new("live-cc-incr-r2");
        std::fs::write(r2.path(), &img2).unwrap();
        assert_eq!(count_rows(r2.path()).await, 12);
    }

    // ── R732-F2 (W245): tenant fencing epochs on the R2 write path ────────
    //
    // The property under test throughout: a writer holding a stale fencing
    // token is REJECTED, not merely unlucky. Every test below names the
    // split-brain it rules out.

    fn cfg_at_epoch(epoch: u64, owner: Option<&'static str>) -> StreamConfig<'static> {
        StreamConfig { epoch, owner, ..cfg() }
    }

    /// R736-T2: like `cfg_at_epoch`, but also names the cross-cell pointer
    /// generation this writer believes it holds.
    fn cfg_at(epoch: u64, pointer_generation: u64, owner: Option<&'static str>) -> StreamConfig<'static> {
        StreamConfig { epoch, pointer_generation, owner, ..cfg() }
    }

    /// THE canonical F2 test, and the reason the whole relay exists: two
    /// owners tailing the same tenant. The one at the older epoch bounces and
    /// writes nothing; the sink is byte-for-byte what the newer owner left.
    #[tokio::test]
    async fn a_stale_writer_is_fenced_and_writes_zero_frames() {
        let target = fresh_target();

        // The real owner (epoch 2) streams three frames.
        let winner = MockWal::new(4096);
        for i in 1..=3u32 {
            winner.append(i, i, i as u8);
        }
        let out = tail_frames(&winner, &target, &cfg_at_epoch(2, Some("node-2")))
            .await
            .unwrap();
        assert!(matches!(out, StreamOutcome::Streamed { frame_count: 3, .. }));

        let manifests_before = list_and_parse_generation_manifests(&target).await.unwrap();
        let watermark_before = read_watermark(&target.store, &target.watermark_key())
            .await
            .unwrap()
            .unwrap();

        // The partitioned old owner (epoch 1) wakes up with its own frames and
        // tails the same sink, unaware it has been transferred away.
        let loser = MockWal::new(4096);
        for i in 1..=9u32 {
            loser.append(i, i, 0xff);
        }
        let out = tail_frames(&loser, &target, &cfg_at_epoch(1, Some("node-1")))
            .await
            .unwrap();
        assert_eq!(
            out,
            StreamOutcome::Fenced {
                current_epoch: 2,
                our_epoch: 1,
                current_pointer_generation: 0,
                our_pointer_generation: 0,
            },
            "the stale owner must be told it lost, with both epochs"
        );

        // Zero side effects: no frames under its own epoch prefix, no extra
        // generation, and the watermark still names the winner.
        // Nothing at all under the loser's epoch prefix — asserted by listing
        // rather than by probing one key, so it holds whatever object layout
        // the writer would have used (R761-F2).
        let loser_prefix = "backups/frames/00000000000000000001/";
        assert!(
            !objects_under(&target)
                .await
                .iter()
                .any(|k| k.starts_with(loser_prefix)),
            "a fenced writer must not upload a single frame"
        );
        let manifests_after = list_and_parse_generation_manifests(&target).await.unwrap();
        assert_eq!(
            manifests_after, manifests_before,
            "a fenced writer must not write a generation manifest"
        );
        let watermark_after = read_watermark(&target.store, &target.watermark_key())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(watermark_after.epoch, 2);
        assert_eq!(watermark_after.watermark, watermark_before.watermark);
    }

    /// An UNFENCED (epoch 0) legacy streamer is fenced by a sink that has been
    /// claimed. This is the migration case: a node still running the old
    /// single-writer configuration must not be allowed to scribble over a
    /// tenant that yubaba has since handed to someone else.
    #[tokio::test]
    async fn an_unfenced_writer_is_fenced_by_a_claimed_sink() {
        let target = fresh_target();
        let owner = MockWal::new(4096);
        owner.append(1, 1, 1);
        tail_frames(&owner, &target, &cfg_at_epoch(1, None)).await.unwrap();

        let legacy = MockWal::new(4096);
        legacy.append(1, 1, 2);
        assert_eq!(
            tail_frames(&legacy, &target, &cfg_at_epoch(0, None)).await.unwrap(),
            StreamOutcome::Fenced {
                current_epoch: 1,
                our_epoch: 0,
                current_pointer_generation: 0,
                our_pointer_generation: 0,
            }
        );
    }

    // ── R736-T2 (W250): the second, cross-cell fence ───────────────────────
    //
    // The epoch alone is a *local* raft counter — it cannot see a tenant
    // that moved to a different cell's independent raft group. These tests
    // prove the pointer generation catches exactly the case the epoch can't:
    // a stale cell that is current on its own epoch.

    /// THE canonical T2 test: a writer whose epoch is perfectly current for
    /// its own (now-stale) cell still bounces, because the global pointer
    /// says ownership moved elsewhere. Proves the epoch is necessary but not
    /// sufficient — this is the gap W250 exists to close.
    #[tokio::test]
    async fn a_stale_pointer_generation_fences_even_at_a_current_epoch() {
        let target = fresh_target();

        // The new cell streams at generation 2, epoch 1 (its own local raft
        // is fresh — it just took ownership).
        let winner = MockWal::new(4096);
        for i in 1..=3u32 {
            winner.append(i, i, i as u8);
        }
        let out = tail_frames(&winner, &target, &cfg_at(1, 2, Some("cell-b/node-1")))
            .await
            .unwrap();
        assert!(matches!(out, StreamOutcome::Streamed { frame_count: 3, .. }));

        // The old cell's writer wakes up unaware of the move. Its own local
        // epoch (1) is perfectly current for its own raft group — nothing
        // local told it to step down — but its pointer generation (1) is
        // behind the sink's (2).
        let loser = MockWal::new(4096);
        for i in 1..=9u32 {
            loser.append(i, i, 0xff);
        }
        let out = tail_frames(&loser, &target, &cfg_at(1, 1, Some("cell-a/node-1")))
            .await
            .unwrap();
        assert_eq!(
            out,
            StreamOutcome::Fenced {
                current_epoch: 1,
                our_epoch: 1,
                current_pointer_generation: 2,
                our_pointer_generation: 1,
            },
            "an equal, non-stale epoch must not mask a stale pointer generation"
        );

        // Zero side effects, exactly like the epoch-only fence.
        let manifests = list_and_parse_generation_manifests(&target).await.unwrap();
        assert_eq!(manifests.len(), 1, "only the winner's generation was written");
    }

    /// The watermark sidecar carries the pointer generation as a fifth
    /// positional field, and a four-field sidecar (pre-R736-T2 writer) reads
    /// back generation 0 — unfenced, exactly like a pre-R732 sidecar reads
    /// back epoch 0.
    #[tokio::test]
    async fn watermark_sidecar_round_trips_the_pointer_generation() {
        let target = fresh_target();
        let key = target.watermark_key();
        write_watermark(
            &target.store,
            &key,
            Watermark { checkpoint_seq: 2, last_frame: 11 },
            None,
            6,
            3,
            None,
        )
        .await
        .unwrap();
        let read = read_watermark(&target.store, &key).await.unwrap().unwrap();
        assert_eq!((read.epoch, read.pointer_generation), (6, 3));

        // A four-field sidecar (epoch, no generation) — the R732-F2 shape.
        target
            .store
            .put(&key, b"2 11 12345 6\n".to_vec().into())
            .await
            .unwrap();
        let legacy = read_watermark(&target.store, &key).await.unwrap().unwrap();
        assert_eq!(legacy.epoch, 6);
        assert_eq!(legacy.pointer_generation, 0, "a pre-R736-T2 sidecar fences nobody on generation");
    }

    /// R869: `read_fence_state` reports exactly the two comparands
    /// [`tail_frames`] checks — including the pre-fencing sidecar shapes, which
    /// must read back as unfenced rather than erroring, since a rebuild will
    /// meet them on any sink written before R732-F2.
    #[tokio::test]
    async fn read_fence_state_reports_what_tail_frames_would_check() {
        let target = fresh_target();
        assert_eq!(
            read_fence_state(&target).await.unwrap(),
            None,
            "no sidecar means no fence at all, which is not the same as a zero fence"
        );

        write_watermark(
            &target.store,
            &target.watermark_key(),
            Watermark {
                checkpoint_seq: 2,
                last_frame: 11,
            },
            None,
            6,
            3,
            None,
        )
        .await
        .unwrap();
        assert_eq!(
            read_fence_state(&target).await.unwrap(),
            Some(FenceState {
                epoch: 6,
                pointer_generation: 3
            })
        );

        // A two-field sidecar — the original pre-fencing shape. Both comparands
        // read back 0, so a rebuild sees "unfenced" rather than a parse error.
        target
            .store
            .put(&target.watermark_key(), b"2 11\n".to_vec().into())
            .await
            .unwrap();
        assert_eq!(
            read_fence_state(&target).await.unwrap(),
            Some(FenceState::default())
        );
    }

    /// The property a rebuild's epoch floor rests on, stated as a test rather
    /// than as a comment: seeding one above `read_fence_state().epoch` is
    /// exactly enough to stop being fenced, and one *below* it is not.
    #[tokio::test]
    async fn a_floor_taken_from_read_fence_state_is_what_unfences_a_rebuild() {
        let target = fresh_target();
        let seam = MockWal::new(4096);
        for i in 1..=3u32 {
            seam.append(i, i, i as u8);
        }
        tail_frames(&seam, &target, &cfg_at_epoch(5, Some("dead-fleet")))
            .await
            .unwrap();

        let fence = read_fence_state(&target).await.unwrap().unwrap();
        assert_eq!(fence.epoch, 5);

        // A rebuilt cluster that restarted its epochs at 1 is refused.
        seam.append(4, 4, 4);
        let out = tail_frames(&seam, &target, &cfg_at_epoch(1, Some("rebuilt")))
            .await
            .unwrap();
        assert!(matches!(out, StreamOutcome::Fenced { .. }), "got {out:?}");

        // Seeded from the fence, it is not.
        let out = tail_frames(
            &seam,
            &target,
            &cfg_at_epoch(fence.epoch + 1, Some("rebuilt")),
        )
        .await
        .unwrap();
        assert!(matches!(out, StreamOutcome::Streamed { .. }), "got {out:?}");
    }

    /// The same owner resuming at the same epoch is NOT fenced — fencing is
    /// strictly "someone newer exists", not "someone else wrote here". An
    /// owner that restarts under an unchanged token must keep streaming, or
    /// every process restart would wedge the tenant.
    #[tokio::test]
    async fn an_equal_epoch_writer_resumes_normally() {
        let target = fresh_target();
        let seam = MockWal::new(4096);
        seam.append(1, 1, 1);
        tail_frames(&seam, &target, &cfg_at_epoch(3, None)).await.unwrap();
        seam.append(2, 2, 2);
        let out = tail_frames(&seam, &target, &cfg_at_epoch(3, None)).await.unwrap();
        match out {
            StreamOutcome::Streamed { first_frame, last_frame, .. } => {
                assert_eq!((first_frame, last_frame), (2, 2), "resumes after the watermark");
            }
            other => panic!("expected Streamed, got {other:?}"),
        }
    }

    /// Frame keys are namespaced by epoch, and epoch 0 keeps the pre-fencing
    /// two-level layout so existing backups stay addressable.
    #[test]
    fn frame_keys_are_namespaced_by_epoch_with_zero_keeping_the_legacy_layout() {
        let target = fresh_target();
        assert_eq!(
            target.frame_key(0, 7, 42).to_string(),
            "backups/frames/0000000007/00000000000000000042",
            "epoch 0 must keep the original key shape"
        );
        assert_eq!(
            target.frame_key(5, 7, 42).to_string(),
            "backups/frames/00000000000000000005/0000000007/00000000000000000042"
        );
        assert_ne!(target.frame_key(5, 7, 42), target.frame_key(6, 7, 42));
    }

    /// R761-F2: a batch key carries its frame range, keeps the epoch
    /// namespacing and the zero-padding (so lexical order is still frame
    /// order), and cannot be confused with a pre-batching per-frame key even
    /// when the batch holds exactly one frame.
    #[test]
    fn batch_keys_carry_the_range_and_never_collide_with_a_per_frame_key() {
        let target = fresh_target();
        assert_eq!(
            target.frame_batch_key(0, 7, 42, 99).to_string(),
            "backups/frames/0000000007/00000000000000000042-00000000000000000099",
            "epoch 0 keeps the two-level layout for batches too"
        );
        assert_eq!(
            target.frame_batch_key(5, 7, 42, 99).to_string(),
            "backups/frames/00000000000000000005/0000000007/00000000000000000042-00000000000000000099"
        );
        assert_ne!(
            target.frame_batch_key(0, 7, 42, 42),
            target.frame_key(0, 7, 42),
            "a one-frame batch is still a batch — the two layouts must stay distinguishable"
        );
        // Lexical order matches frame order for the batches of one stream,
        // which never overlap.
        assert!(
            target.frame_batch_key(0, 7, 1, 8) < target.frame_batch_key(0, 7, 9, 16),
            "zero-padding must keep batches lexically ordered by first frame"
        );
    }

    /// A takeover mid-stream leaves both owners' frames intact under their own
    /// prefixes — the key namespacing is the backstop behind the epoch check.
    #[tokio::test]
    async fn a_takeover_writes_under_its_own_epoch_prefix_without_disturbing_the_old_one() {
        let target = fresh_target();
        let seam = MockWal::new(4096);
        seam.append(1, 1, 0xaa);
        tail_frames(&seam, &target, &cfg_at_epoch(1, None)).await.unwrap();
        seam.append(2, 2, 0xbb);
        tail_frames(&seam, &target, &cfg_at_epoch(2, None)).await.unwrap();

        let first = target.store.get(&target.frame_batch_key(1, 0, 1, 1)).await.unwrap();
        assert_eq!(first.bytes().await.unwrap().len(), WAL_FRAME_HEADER_SIZE + 4096);
        target
            .store
            .get(&target.frame_batch_key(2, 0, 2, 2))
            .await
            .expect("the new owner's frame lives under its own epoch");
        assert!(
            target.store.get(&target.frame_batch_key(1, 0, 2, 2)).await.is_err(),
            "the new owner must not write into the old owner's prefix"
        );
    }

    /// Restore refuses a chain whose epoch goes backwards: a generation
    /// written by an owner that had already been fenced. Replaying it would
    /// interleave a stale owner's frames into the live stream.
    #[test]
    fn validate_chain_refuses_an_epoch_regression() {
        let mut newer = mk_manifest("base.db", 4096, 0, 1, 5);
        newer.epoch = 4;
        let mut stale = mk_manifest("base.db", 4096, 0, 6, 9);
        stale.epoch = 3;
        let err = validate_generation_chain(&[newer, stale]).unwrap_err();
        let msg = format!("{err}");
        assert!(msg.contains("epoch 3"), "err was {msg}");
        assert!(msg.contains("fenced"), "err must name the cause: {msg}");
    }

    /// …but a chain that spans an ownership TRANSFER is fine. Epochs may
    /// advance mid-stream; only regression is corruption.
    #[test]
    fn validate_chain_accepts_a_transfer_mid_chain() {
        let mut first = mk_manifest("base.db", 4096, 0, 1, 5);
        first.epoch = 3;
        let mut second = mk_manifest("base.db", 4096, 0, 6, 9);
        second.epoch = 4;
        let chain = validate_generation_chain(&[first, second]).unwrap();
        assert_eq!(chain.total_frames, 9);
        assert_eq!(chain.epoch, 4, "the chain reports the most recent owner");
    }

    /// A pre-fencing chain still validates and reports epoch 0.
    #[test]
    fn validate_chain_of_pre_fencing_manifests_reports_epoch_zero() {
        let chain = validate_generation_chain(&[mk_manifest("base.db", 4096, 0, 1, 5)]).unwrap();
        assert_eq!(chain.epoch, 0);
    }

    /// Manifest v2 round-trips the epoch and the owner label.
    #[test]
    fn manifest_v2_round_trips_epoch_and_owner() {
        let text = format_generation_manifest(GenerationManifest {
            base_snapshot_key: "backups/snapshots/snapshot-1.db",
            page_size: 4096,
            checkpoint_seq: 7,
            salt: None,
            first_frame: 12,
            last_frame: 34,
            epoch: 9,
            owner: Some("node-3"),
            frame_batches: &[],
        });
        assert!(text.starts_with("TURSO-BACKUP STREAM v2\n"), "{text}");
        let parsed = parse_generation_manifest(&text).unwrap();
        assert_eq!(parsed.epoch, 9);
        assert_eq!(parsed.owner.as_deref(), Some("node-3"));

        // No owner label → the key is omitted entirely, not written empty.
        let text = format_generation_manifest(GenerationManifest {
            base_snapshot_key: "b.db",
            page_size: 4096,
            checkpoint_seq: 0,
            salt: None,
            first_frame: 1,
            last_frame: 1,
            epoch: 1,
            owner: None,
            frame_batches: &[],
        });
        assert!(!text.contains("owner"), "{text}");
        assert_eq!(parse_generation_manifest(&text).unwrap().owner, None);
    }

    /// R761-F2: a batch list round-trips, and its presence is what moves the
    /// header to v3 — the version and the layout are one fact, so a reader can
    /// never see a v3 header without an index or a v2 header with one.
    #[test]
    fn manifest_v3_round_trips_the_frame_batch_list() {
        let batches = [(12u64, 20u64), (21, 34)];
        let text = format_generation_manifest(GenerationManifest {
            base_snapshot_key: "backups/snapshots/snapshot-1.db",
            page_size: 4096,
            checkpoint_seq: 7,
            salt: None,
            first_frame: 12,
            last_frame: 34,
            epoch: 9,
            owner: Some("node-3"),
            frame_batches: &batches,
        });
        assert!(text.starts_with("TURSO-BACKUP STREAM v3\n"), "{text}");
        let parsed = parse_generation_manifest(&text).unwrap();
        assert_eq!(parsed.frame_batches, batches.to_vec());
        assert_eq!(parsed.first_frame, 12);
        assert_eq!(parsed.last_frame, 34);
        assert_eq!(parsed.owner.as_deref(), Some("node-3"));
    }

    /// A v3 batch list that does not exactly tile the manifest's own frame
    /// range is corruption, and it has to fail at parse: the list IS the frame
    /// index, so a hole in it becomes a 404 halfway through a replay — after
    /// the destination has already been overwritten with the base snapshot.
    #[test]
    fn a_v3_manifest_whose_batches_do_not_tile_its_range_is_rejected() {
        let head = "TURSO-BACKUP STREAM v3\nbase_snapshot b.db\npage_size 4096\ncheckpoint_seq 0\nepoch 0\n";
        for (batches, want, why) in [
            ("frame_batch 1-3\nframe_batch 5-9\n", "does not continue", "gap"),
            ("frame_batch 1-3\n", "but the manifest claims", "short"),
            ("frame_batch 2-9\n", "does not continue", "wrong start"),
            ("frame_batch 1-4\nframe_batch 4-9\n", "does not continue", "overlap"),
            ("", "no `frame_batch` lines", "missing index"),
        ] {
            let text = format!("{head}first_frame 1\nlast_frame 9\n{batches}");
            let err = parse_generation_manifest(&text).unwrap_err();
            assert!(
                format!("{err}").contains(want),
                "{why}: expected {want:?}, err was {err}"
            );
        }
    }

    /// The other direction: a `frame_batch` line under a v1/v2 header is
    /// corrupt too. Silently honouring it would let a hand-edited manifest
    /// claim a layout its header says it does not have.
    #[test]
    fn a_pre_v3_manifest_carrying_a_batch_line_is_rejected() {
        let bad = "TURSO-BACKUP STREAM v2\nbase_snapshot b.db\npage_size 4096\ncheckpoint_seq 0\nfirst_frame 1\nlast_frame 3\nepoch 0\nframe_batch 1-3\n";
        let err = parse_generation_manifest(bad).unwrap_err();
        assert!(format!("{err}").contains("corrupt or hand-edited"), "err was {err}");
    }

    /// A v1 manifest — one written before fencing existed — still parses, as
    /// epoch 0. Those backups have to stay restorable.
    #[test]
    fn a_v1_manifest_parses_as_epoch_zero() {
        let legacy = "TURSO-BACKUP STREAM v1\nbase_snapshot b.db\npage_size 4096\ncheckpoint_seq 0\nfirst_frame 1\nlast_frame 3\n";
        let parsed = parse_generation_manifest(legacy).unwrap();
        assert_eq!(parsed.epoch, 0);
        assert_eq!(parsed.owner, None);
        assert_eq!(parsed.last_frame, 3);
    }

    /// A v2 manifest missing its epoch is corrupt, not legacy. Defaulting it
    /// to 0 would silently demote a fenced generation to unfenced — the one
    /// direction this mechanism must never fail in.
    #[test]
    fn a_v2_manifest_without_an_epoch_is_rejected() {
        let bad = "TURSO-BACKUP STREAM v2\nbase_snapshot b.db\npage_size 4096\ncheckpoint_seq 0\nfirst_frame 1\nlast_frame 3\n";
        let err = parse_generation_manifest(bad).unwrap_err();
        assert!(format!("{err}").contains("missing `epoch`"), "err was {err}");
    }

    /// The watermark sidecar carries the epoch as a fourth positional field,
    /// and a three-field sidecar (pre-R732 writer) reads back as epoch 0.
    #[tokio::test]
    async fn watermark_sidecar_round_trips_the_epoch() {
        let target = fresh_target();
        let key = target.watermark_key();
        write_watermark(
            &target.store,
            &key,
            Watermark { checkpoint_seq: 2, last_frame: 11 },
            None,
            6,
            0,
            None,
        )
        .await
        .unwrap();
        let read = read_watermark(&target.store, &key).await.unwrap().unwrap();
        assert_eq!(read.epoch, 6);
        assert_eq!(read.watermark.last_frame, 11);
        assert!(read.written_at_nanos.is_some());

        target
            .store
            .put(&key, b"2 11 12345\n".to_vec().into())
            .await
            .unwrap();
        let legacy = read_watermark(&target.store, &key).await.unwrap().unwrap();
        assert_eq!(legacy.epoch, 0, "a pre-R732 sidecar fences nobody");
        assert_eq!(legacy.written_at_nanos, Some(12345));
    }

    /// End-to-end on a real DB: an ownership transfer happens mid-stream and
    /// the restore is clean — every frame from both owners replays, and the
    /// outcome reports the winning epoch. This is the "the N+1 writer wins;
    /// restore is clean" half of the ticket's ask, run against turso_core
    /// rather than a mock.
    #[tokio::test]
    async fn a_transfer_mid_stream_restores_cleanly_and_reports_the_new_epoch() {
        let src = TempDb::new("src-epoch");
        let dest = TempDb::new("dest-epoch");
        seed_rows(src.path(), 0, 50).await;

        let target = fresh_target();
        let base_key = match crate::snapshot::snapshot_and_upload(src.path(), &target)
            .await
            .unwrap()
        {
            crate::snapshot::SnapshotOutcome::Uploaded { key, .. } => key,
            other => panic!("expected Uploaded base snapshot, got {other:?}"),
        };
        checkpoint_truncate(src.path()).await;

        // Owner A (epoch 1) streams the first batch of writes.
        seed_rows(src.path(), 1000, 10).await;
        {
            let seam = CoreWalSeam::open(src.path()).unwrap();
            let cfg = StreamConfig {
                base_snapshot_key: &base_key,
                epoch: 1,
                owner: Some("node-a"),
                ..cfg()
            };
            let out = tail_frames(&seam, &target, &cfg).await.unwrap();
            assert!(
                matches!(
                    out,
                    StreamOutcome::Streamed { .. } | StreamOutcome::Restarted { .. }
                ),
                "owner A should have streamed, got {out:?}"
            );
        }

        // Ownership transfers. Owner B (epoch 2) picks up where A stopped.
        seed_rows(src.path(), 2000, 15).await;
        {
            let seam = CoreWalSeam::open(src.path()).unwrap();
            let cfg = StreamConfig {
                base_snapshot_key: &base_key,
                epoch: 2,
                owner: Some("node-b"),
                ..cfg()
            };
            let out = tail_frames(&seam, &target, &cfg).await.unwrap();
            assert!(
                matches!(
                    out,
                    StreamOutcome::Streamed { .. } | StreamOutcome::Restarted { .. }
                ),
                "owner B should have streamed, got {out:?}"
            );
        }

        let outcome = restore_latest_stream(&target, dest.path()).await.unwrap();
        assert_eq!(outcome.base_snapshot_key, base_key);
        assert_eq!(outcome.epoch, 2, "restore reports the most recent owner");
        assert_eq!(
            count_rows(dest.path()).await,
            75,
            "50 (base) + 10 (owner A) + 15 (owner B)"
        );
    }

    // ── R732-T3 (W245): the watermark advance is a compare-and-swap ───────

    /// Bootstrapping a fresh sink uses `PutMode::Create`, so two writers
    /// racing to claim a brand-new tenant cannot both succeed. Without this
    /// the very first write — the one with no prior version to swap on —
    /// would be the one unguarded moment in the whole protocol.
    #[tokio::test]
    async fn a_second_bootstrap_of_a_fresh_sink_is_contended() {
        let target = fresh_target();
        let key = target.watermark_key();
        let w = Watermark { checkpoint_seq: 0, last_frame: 1 };
        assert_eq!(
            write_watermark(&target.store, &key, w, None, 1, 0, None).await.unwrap(),
            WatermarkCas::Advanced
        );
        assert_eq!(
            write_watermark(&target.store, &key, w, None, 1, 0, None).await.unwrap(),
            WatermarkCas::Contended,
            "the sink already exists — Create must not silently overwrite it"
        );
    }

    /// An advance is conditional on the version actually read. A writer
    /// holding a version somebody else has already replaced loses.
    #[tokio::test]
    async fn an_advance_on_a_replaced_version_is_contended() {
        let target = fresh_target();
        let key = target.watermark_key();
        write_watermark(
            &target.store,
            &key,
            Watermark { checkpoint_seq: 0, last_frame: 1 },
            None,
            1,
            0,
            None,
        )
        .await
        .unwrap();

        // Our writer reads the sidecar and holds onto that version…
        let stale = read_watermark(&target.store, &key).await.unwrap().unwrap();
        // …while somebody else advances it out from under us.
        write_watermark(
            &target.store,
            &key,
            Watermark { checkpoint_seq: 0, last_frame: 5 },
            None,
            2,
            0,
            read_watermark(&target.store, &key)
                .await
                .unwrap()
                .unwrap()
                .version
                .as_ref(),
        )
        .await
        .unwrap();

        assert_eq!(
            write_watermark(
                &target.store,
                &key,
                Watermark { checkpoint_seq: 0, last_frame: 2 },
                None,
                1,
                0,
                stale.version.as_ref(),
            )
            .await
            .unwrap(),
            WatermarkCas::Contended,
            "the stale version must not be allowed to overwrite the newer one"
        );
        // And the sink still names the winner, not us.
        let now = read_watermark(&target.store, &key).await.unwrap().unwrap();
        assert_eq!((now.epoch, now.watermark.last_frame), (2, 5));
    }

    /// Re-reading before advancing works: the point is the version, not the
    /// identity of the writer.
    #[tokio::test]
    async fn an_advance_on_the_current_version_succeeds() {
        let target = fresh_target();
        let key = target.watermark_key();
        write_watermark(
            &target.store,
            &key,
            Watermark { checkpoint_seq: 0, last_frame: 1 },
            None,
            1,
            0,
            None,
        )
        .await
        .unwrap();
        let cur = read_watermark(&target.store, &key).await.unwrap().unwrap();
        assert_eq!(
            write_watermark(
                &target.store,
                &key,
                Watermark { checkpoint_seq: 0, last_frame: 7 },
                None,
                1,
                0,
                cur.version.as_ref(),
            )
            .await
            .unwrap(),
            WatermarkCas::Advanced
        );
    }

    /// A store that lets a test slip a competing writer in between our read
    /// of the watermark and our conditional write of it — the interleaving
    /// that a sequential epoch check cannot catch and the CAS must.
    ///
    /// On the first `put_opts` aimed at the watermark key it writes a rival
    /// sidecar (at `rival_epoch`) straight through to the inner store, then
    /// forwards our conditional put, which now finds a version it does not
    /// hold.
    struct RacingStore {
        inner: Arc<dyn ObjectStore>,
        watermark: ObjPath,
        rival_epoch: u64,
        /// R736-T2: the rival's pointer generation, so the same interleaving
        /// can be exercised for the cross-cell fence too.
        rival_pointer_generation: u64,
        fired: std::sync::atomic::AtomicBool,
    }

    impl std::fmt::Display for RacingStore {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "RacingStore({})", self.inner)
        }
    }
    impl std::fmt::Debug for RacingStore {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "RacingStore({:?})", self.inner)
        }
    }

    #[async_trait::async_trait]
    impl ObjectStore for RacingStore {
        async fn put_opts(
            &self,
            location: &ObjPath,
            payload: object_store::PutPayload,
            opts: PutOptions,
        ) -> object_store::Result<object_store::PutResult> {
            if location == &self.watermark
                && !self
                    .fired
                    .swap(true, std::sync::atomic::Ordering::SeqCst)
            {
                let rival =
                    format!("0 99 1 {} {}\n", self.rival_epoch, self.rival_pointer_generation);
                self.inner
                    .put(location, rival.into_bytes().into())
                    .await?;
            }
            self.inner.put_opts(location, payload, opts).await
        }

        async fn put_multipart_opts(
            &self,
            location: &ObjPath,
            opts: object_store::PutMultipartOptions,
        ) -> object_store::Result<Box<dyn object_store::MultipartUpload>> {
            self.inner.put_multipart_opts(location, opts).await
        }

        async fn get_opts(
            &self,
            location: &ObjPath,
            options: object_store::GetOptions,
        ) -> object_store::Result<object_store::GetResult> {
            self.inner.get_opts(location, options).await
        }

        fn delete_stream(
            &self,
            locations: futures_util::stream::BoxStream<'static, object_store::Result<ObjPath>>,
        ) -> futures_util::stream::BoxStream<'static, object_store::Result<ObjPath>> {
            self.inner.delete_stream(locations)
        }

        fn list(
            &self,
            prefix: Option<&ObjPath>,
        ) -> futures_util::stream::BoxStream<'static, object_store::Result<object_store::ObjectMeta>>
        {
            self.inner.list(prefix)
        }

        async fn list_with_delimiter(
            &self,
            prefix: Option<&ObjPath>,
        ) -> object_store::Result<object_store::ListResult> {
            self.inner.list_with_delimiter(prefix).await
        }

        async fn copy_opts(
            &self,
            from: &ObjPath,
            to: &ObjPath,
            options: object_store::CopyOptions,
        ) -> object_store::Result<()> {
            self.inner.copy_opts(from, to, options).await
        }
    }

    fn racing_target(rival_epoch: u64) -> BackupTarget {
        racing_target_at(rival_epoch, 0)
    }

    /// R736-T2: like `racing_target`, but also names the rival's pointer
    /// generation.
    fn racing_target_at(rival_epoch: u64, rival_pointer_generation: u64) -> BackupTarget {
        let inner: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
        let prefix = "backups".to_string();
        let watermark = join_key(&prefix, "latest.stream-watermark");
        BackupTarget {
            store: Arc::new(RacingStore {
                inner,
                watermark,
                rival_epoch,
                rival_pointer_generation,
                fired: std::sync::atomic::AtomicBool::new(false),
            }),
            prefix,
        }
    }

    /// The race the sequential check cannot see: a newer owner claims the
    /// sink *after* we read it and *before* we write it. The CAS catches it,
    /// and we report Fenced without publishing a manifest — which is why the
    /// manifest write had to move after the watermark advance.
    #[tokio::test]
    async fn losing_the_watermark_race_to_a_newer_owner_fences_us_before_we_publish() {
        let target = racing_target(9);
        let seam = MockWal::new(4096);
        for i in 1..=2u32 {
            seam.append(i, i, i as u8);
        }
        let out = tail_frames(&seam, &target, &cfg_at_epoch(4, Some("node-loser")))
            .await
            .unwrap();
        assert_eq!(
            out,
            StreamOutcome::Fenced {
                current_epoch: 9,
                our_epoch: 4,
                current_pointer_generation: 0,
                our_pointer_generation: 0,
            }
        );

        // No manifest published — the chain stays clean, so restore is not
        // poisoned by a regressed generation from a writer that lost.
        assert!(
            list_and_parse_generation_manifests(&target)
                .await
                .unwrap()
                .is_empty(),
            "a writer that loses the CAS must publish nothing"
        );
        // The rival's sidecar survived untouched.
        let now = read_watermark(&target.store, &target.watermark_key())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(now.epoch, 9);
    }

    /// R736-T2: the same race, but the rival is a different cell claiming the
    /// tenant via a pointer CAS — our epoch is unchanged (we were never told
    /// to step down locally) but the generation moved under us mid-write.
    #[tokio::test]
    async fn losing_the_watermark_race_to_a_cross_cell_move_fences_us_before_we_publish() {
        let target = racing_target_at(4, 2);
        let seam = MockWal::new(4096);
        for i in 1..=2u32 {
            seam.append(i, i, i as u8);
        }
        let out = tail_frames(&seam, &target, &cfg_at(4, 1, Some("cell-a/node-loser")))
            .await
            .unwrap();
        assert_eq!(
            out,
            StreamOutcome::Fenced {
                current_epoch: 4,
                our_epoch: 4,
                current_pointer_generation: 2,
                our_pointer_generation: 1,
            },
            "an unchanged epoch must not mask a generation that moved under us"
        );
        assert!(
            list_and_parse_generation_manifests(&target)
                .await
                .unwrap()
                .is_empty(),
            "a writer that loses the CAS on generation must publish nothing"
        );
    }

    /// Losing the race to a writer at our own epoch is NOT a fencing event —
    /// it means two streamers were handed the same token. That is a caller
    /// bug and must surface as a loud error, never be retried into success.
    #[tokio::test]
    async fn losing_the_race_to_an_equal_epoch_writer_is_a_loud_error() {
        let target = racing_target(4);
        let seam = MockWal::new(4096);
        seam.append(1, 1, 1);
        let err = tail_frames(&seam, &target, &cfg_at_epoch(4, None))
            .await
            .unwrap_err();
        let msg = format!("{err}");
        assert!(
            msg.contains("two streamers share one fencing token"),
            "err was {msg}"
        );
    }

    /// Generation manifest round-trip and rejection of garbage.
    #[test]
    fn manifest_round_trip_and_rejection() {
        let text = format_generation_manifest(GenerationManifest {
            base_snapshot_key: "backups/snapshots/snapshot-1.db",
            page_size: 4096,
            checkpoint_seq: 7,
            salt: None,
            first_frame: 12,
            last_frame: 34,
            epoch: 0,
            owner: None,
            frame_batches: &[],
        });
        let parsed = parse_generation_manifest(&text).unwrap();
        assert_eq!(parsed.base_snapshot_key, "backups/snapshots/snapshot-1.db");
        assert_eq!(parsed.page_size, 4096);
        assert_eq!(parsed.checkpoint_seq, 7);
        assert_eq!(parsed.first_frame, 12);
        assert_eq!(parsed.last_frame, 34);

        assert!(parse_generation_manifest("not a manifest").is_err());
        assert!(parse_generation_manifest("TURSO-BACKUP STREAM v1\nunknown 1").is_err());
        assert!(parse_generation_manifest("TURSO-BACKUP STREAM v1\nbase_snapshot k").is_err()); // missing fields
    }

    // ---- R850-T3: tier-2 GC (gc_stream) --------------------------------

    /// A tier-2 sink built object-by-object on an `InMemory` store.
    ///
    /// Hand-placed rather than produced by running `tail_frames`, because every
    /// one of these tests is about a sink in a state a *live* tailer never
    /// leaves behind — a superseded base, a generation's frames with no
    /// manifest. Driving the writer could not produce them without also
    /// producing the rebase that is the thing under test.
    struct GcSink {
        target: BackupTarget,
    }

    impl GcSink {
        fn new() -> Self {
            GcSink {
                target: BackupTarget {
                    store: Arc::new(InMemory::new()),
                    prefix: "sink".to_string(),
                },
            }
        }

        /// A base snapshot at `nanos`. Body is not a real database — nothing in
        /// the GC path parses it.
        async fn put_snapshot(&self, nanos: u128) -> String {
            let key = join_key(
                &self.target.prefix,
                &format!("snapshots/snapshot-{nanos:020}.db"),
            );
            self.target
                .store
                .put(&key, format!("base {nanos}").into_bytes().into())
                .await
                .unwrap();
            key.to_string()
        }

        /// One batch object holding frames `first..=last` of generation
        /// `(epoch, seq)`, keyed exactly as `tail_frames` would key it.
        async fn put_frame_batch(&self, epoch: u64, seq: u32, first: u64, last: u64) -> String {
            let key = self.target.frame_batch_key(epoch, seq, first, last);
            self.target
                .store
                .put(&key, vec![0u8; 16].into())
                .await
                .unwrap();
            key.to_string()
        }

        /// A generation manifest naming `base` and one batch `first..=last`.
        async fn put_manifest(
            &self,
            nanos: u128,
            base: &str,
            epoch: u64,
            seq: u32,
            first: u64,
            last: u64,
        ) {
            let batches = [(first, last)];
            let text = format_generation_manifest(GenerationManifest {
                base_snapshot_key: base,
                page_size: 4096,
                checkpoint_seq: seq,
                salt: Some(WalSalt {
                    salt1: 1,
                    salt2: 2,
                }),
                first_frame: first,
                last_frame: last,
                epoch,
                owner: None,
                frame_batches: &batches,
            });
            self.target
                .store
                .put(&self.target.generation_key(nanos), text.into_bytes().into())
                .await
                .unwrap();
        }

        async fn exists(&self, key: &str) -> bool {
            self.target.store.get(&ObjPath::from(key)).await.is_ok()
        }
    }

    /// Collect everything collectable: no grace, really delete.
    fn sweep_now() -> StreamGcConfig {
        StreamGcConfig {
            grace: Duration::ZERO,
            dry_run: false,
        }
    }

    /// (a) The base a rebase superseded is reclaimed; the one a restore would
    /// pick is not.
    #[tokio::test]
    async fn gc_collects_a_superseded_base_snapshot_and_keeps_the_current_one() {
        let sink = GcSink::new();
        let old_base = sink.put_snapshot(1).await;
        let new_base = sink.put_snapshot(2).await;
        // The post-rebase steady state: the only chain names the new base.
        sink.put_manifest(10, &new_base, 0, 5, 1, 10).await;
        sink.put_frame_batch(0, 5, 1, 10).await;

        let out = gc_stream(&sink.target, &sweep_now()).await.unwrap();

        assert_eq!(out.collected_snapshots, vec![old_base.clone()]);
        assert_eq!(out.retained_snapshots, 1);
        assert_eq!(out.live_base_snapshot_keys, vec![new_base.clone()]);
        assert!(!sink.exists(&old_base).await, "superseded base survived");
        assert!(sink.exists(&new_base).await, "live base was collected");
    }

    /// The rebase-crash window: a new base is published before the old
    /// manifests are deleted, so the surviving chain names the OLD base. Both
    /// are live — the chain's pick and the newest — and the GC must refuse to
    /// choose between them.
    #[tokio::test]
    async fn gc_keeps_both_bases_while_a_rebase_is_half_landed() {
        let sink = GcSink::new();
        let old_base = sink.put_snapshot(1).await;
        let new_base = sink.put_snapshot(2).await;
        sink.put_manifest(10, &old_base, 0, 5, 1, 10).await;

        let out = gc_stream(&sink.target, &sweep_now()).await.unwrap();

        assert!(out.collected_snapshots.is_empty());
        assert_eq!(out.retained_snapshots, 2);
        assert_eq!(out.live_base_snapshot_keys, vec![old_base, new_base]);
    }

    /// (b) An orphaned generation's frames are reclaimed — under both key
    /// shapes — and a live generation's are not.
    #[tokio::test]
    async fn gc_collects_orphaned_frame_prefixes_and_keeps_the_live_generation() {
        let sink = GcSink::new();
        let base = sink.put_snapshot(2).await;
        sink.put_manifest(10, &base, 0, 5, 1, 10).await;
        let live = sink.put_frame_batch(0, 5, 1, 10).await;
        // Generation 4 was rebased away: its manifests are gone, its frames
        // are not. Once under the epoch-0 layout, once under the fenced one.
        let orphan_unfenced = sink.put_frame_batch(0, 4, 1, 6).await;
        let orphan_fenced = sink.put_frame_batch(7, 4, 1, 6).await;

        let out = gc_stream(&sink.target, &sweep_now()).await.unwrap();

        let mut expected = vec![orphan_unfenced.clone(), orphan_fenced.clone()];
        expected.sort();
        assert_eq!(out.collected_frame_objects, expected);
        assert_eq!(out.retained_frame_objects, 1);
        assert_eq!(out.collected_bytes, 32, "two 16-byte batch objects");
        assert!(sink.exists(&live).await, "live generation's frames collected");
        assert!(!sink.exists(&orphan_unfenced).await);
        assert!(!sink.exists(&orphan_fenced).await);
    }

    /// (c) Nothing inside the grace window is touched, however unreachable it
    /// looks. Everything an `InMemory` store holds was written moments ago, so
    /// a non-zero grace must spare the whole sweep.
    #[tokio::test]
    async fn gc_spares_everything_inside_the_grace_window() {
        let sink = GcSink::new();
        let stale_base = sink.put_snapshot(1).await;
        let base = sink.put_snapshot(2).await;
        sink.put_manifest(10, &base, 0, 5, 1, 10).await;
        sink.put_frame_batch(0, 5, 1, 10).await;
        let orphan = sink.put_frame_batch(0, 4, 1, 6).await;

        let cfg = StreamGcConfig {
            grace: Duration::from_secs(3600),
            dry_run: false,
        };
        let out = gc_stream(&sink.target, &cfg).await.unwrap();

        assert!(out.collected_frame_objects.is_empty());
        assert!(out.collected_snapshots.is_empty());
        assert_eq!(out.collected_bytes, 0);
        assert_eq!(out.spared_by_grace, 2, "the orphan batch and the stale base");
        assert!(sink.exists(&orphan).await);
        assert!(sink.exists(&stale_base).await);
    }

    /// (d) A dry run reports exactly what the real sweep would collect, and
    /// deletes none of it.
    #[tokio::test]
    async fn gc_dry_run_reports_without_deleting() {
        let sink = GcSink::new();
        let stale_base = sink.put_snapshot(1).await;
        let base = sink.put_snapshot(2).await;
        sink.put_manifest(10, &base, 0, 5, 1, 10).await;
        sink.put_frame_batch(0, 5, 1, 10).await;
        let orphan = sink.put_frame_batch(0, 4, 1, 6).await;

        let dry = gc_stream(
            &sink.target,
            &StreamGcConfig {
                grace: Duration::ZERO,
                dry_run: true,
            },
        )
        .await
        .unwrap();

        assert!(dry.dry_run);
        assert_eq!(dry.collected_snapshots, vec![stale_base.clone()]);
        assert_eq!(dry.collected_frame_objects, vec![orphan.clone()]);
        assert!(dry.collected_bytes > 0);
        assert!(sink.exists(&stale_base).await, "dry run deleted a snapshot");
        assert!(sink.exists(&orphan).await, "dry run deleted a frame object");

        // The proposal is what the real sweep then executes, key for key.
        let wet = gc_stream(&sink.target, &sweep_now()).await.unwrap();
        assert!(!wet.dry_run);
        assert_eq!(wet.collected_snapshots, dry.collected_snapshots);
        assert_eq!(wet.collected_frame_objects, dry.collected_frame_objects);
        assert_eq!(wet.collected_bytes, dry.collected_bytes);
        assert!(!sink.exists(&stale_base).await);
        assert!(!sink.exists(&orphan).await);
    }

    /// The pin `gc_stream`'s docs promise: its live set is the exact complement
    /// of what a restore selects, on BOTH restore branches. If either selection
    /// rule is ever changed without changing the other, this fails.
    #[tokio::test]
    async fn gc_liveness_is_the_complement_of_restore_selection() {
        // Branch 1 — no generations. `hydrate::restore_subject` falls back to
        // `snapshot::restore_latest`, which takes the lexically-greatest key.
        // That key is live; the older ones are not collected either, because a
        // chainless prefix is indistinguishable from a tier-1a sink whose older
        // snapshots are history (see `base_snapshots_skipped`).
        let sink = GcSink::new();
        let _older = sink.put_snapshot(1).await;
        let newest = sink.put_snapshot(2).await;

        let restore_would_pick = crate::snapshot::latest_snapshot_key(&sink.target)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(restore_would_pick, newest);

        let out = gc_stream(
            &sink.target,
            &StreamGcConfig {
                grace: Duration::ZERO,
                dry_run: true,
            },
        )
        .await
        .unwrap();
        assert_eq!(out.live_base_snapshot_keys, vec![restore_would_pick]);
        assert!(out.base_snapshots_skipped);
        assert!(out.collected_snapshots.is_empty());
        assert_eq!(out.retained_snapshots, 2);

        // Branch 2 — a validating chain. `restore_stream_from_manifests` takes
        // the chain's base, which here is the OLDER snapshot. The GC must keep
        // it, and keep the newest as well, since the fallback branch is still
        // reachable for any reader that finds no manifests.
        let chained = GcSink::new();
        let chain_base = chained.put_snapshot(1).await;
        let unreferenced_newer = chained.put_snapshot(2).await;
        chained.put_manifest(10, &chain_base, 0, 5, 1, 10).await;
        chained.put_frame_batch(0, 5, 1, 10).await;

        let manifests = list_and_parse_generation_manifests(&chained.target)
            .await
            .unwrap();
        let chain = validate_generation_chain(&manifests).unwrap();
        assert_eq!(chain.base_snapshot_key, chain_base);

        let out = gc_stream(
            &chained.target,
            &StreamGcConfig {
                grace: Duration::ZERO,
                dry_run: true,
            },
        )
        .await
        .unwrap();
        assert!(!out.base_snapshots_skipped, "a chain makes this a tier-2 sink");
        assert!(out.live_base_snapshot_keys.contains(&chain.base_snapshot_key));
        assert!(out.live_base_snapshot_keys.contains(&unreferenced_newer));
        assert!(out.collected_snapshots.is_empty());

        // And the retained frame set is exactly what a replay would fetch.
        let replayed: Vec<String> = manifests
            .iter()
            .flat_map(|m| chained.target.frame_objects_of(m))
            .map(|(key, _, _)| key.to_string())
            .collect();
        assert_eq!(out.retained_frame_objects, replayed.len());
        assert!(out.collected_frame_objects.is_empty());
    }

    /// A prefix with no chain keeps every snapshot — its older ones could be a
    /// tier-1a sink's history — but its `frames/` are unreachable by
    /// construction and still go.
    #[tokio::test]
    async fn gc_without_a_chain_keeps_snapshots_but_still_collects_frames() {
        let sink = GcSink::new();
        let older = sink.put_snapshot(1).await;
        let newest = sink.put_snapshot(2).await;
        let orphan = sink.put_frame_batch(0, 4, 1, 6).await;

        let out = gc_stream(&sink.target, &sweep_now()).await.unwrap();

        assert!(out.base_snapshots_skipped);
        assert!(out.collected_snapshots.is_empty());
        assert_eq!(out.collected_frame_objects, vec![orphan.clone()]);
        assert!(sink.exists(&older).await);
        assert!(sink.exists(&newest).await);
        assert!(!sink.exists(&orphan).await);
    }

    /// An empty prefix is a no-op, not an error — a GC pointed at a sink that
    /// has never been written to must not fail.
    #[tokio::test]
    async fn gc_on_an_empty_prefix_collects_nothing() {
        let sink = GcSink::new();
        let out = gc_stream(&sink.target, &sweep_now()).await.unwrap();
        assert_eq!(
            out,
            StreamGcOutcome {
                base_snapshots_skipped: true,
                ..Default::default()
            }
        );
    }
}