turso_backup/

dedup.rs

//! Tier 1b — incremental page-dedup snapshot.
//!
//! Raw point-in-time-consistent copy of the live `.db` (NOT `VACUUM INTO`,
//! which reshuffles pages and defeats dedup) → split into fixed 4 KB pages →
//! content-address (hash) each → upload only changed pages + a per-snapshot
//! manifest. Restore reassembles from the manifest. Engine-agnostic.
//!
//! @yah:relay(R004, "Tier 1b — incremental page-dedup snapshot")
//! @yah:at(2026-05-26T22:28:31Z)
//! @yah:status(open)
//! @yah:phase(P2)
//! @yah:parent(Q002)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//!
//! @yah:ticket(R004-T1, "Spike: raw point-in-time-consistent copy primitive (main+WAL); confirm pages keep fixed offsets across snapshots")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-05-26T22:30:06Z)
//! @yah:status(review)
//! @yah:phase(P2)
//! @yah:parent(R004)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//! @yah:handoff("Built dedup::raw_consistent_copy(db_path)->Vec<u8> + dedup::page_size(image). Primitive = PRAGMA wal_checkpoint(TRUNCATE) (turso supports it) to fold WAL->main, drop conn, fs::read main file. NOT vacuum (no repack). Confirmed point-in-time consistent main+WAL: writing only the main bytes to a fresh path reopens with the full row count.")
//! @yah:handoff("Page-offset stability CONFIRMED and it is structural: page N is always at (N-1)*page_size; existing pages never renumber on a raw write. Measured 1-row insert into a 5000-row DB = 1/18 pages changed. page_size read from header u16@16 = 4096 (turso default).")
//! @yah:handoff("CORRECTION to working doc: 'vacuum reshuffles pages and defeats dedup' is too strong. Append-only: raw 1/18 == vacuum 1/18 (equivalent). Vacuum only diverges under reclamation: delete-low-block = raw 10/18 vs vacuum 18/18 (vacuum renumbers all survivors). Reason to copy raw = its offset stability holds for every mutation shape; vacuum's is incidental. Doc updated with the spike findings section.")
//! @yah:verify("cargo test -p turso-backup dedup -- --nocapture: 3 spike tests green (raw_copy_keeps_page_offsets_stable, raw_copy_beats_vacuum_as_dedup_base, raw_copy_folds_wal_and_is_self_contained). Full suite 6/6, clippy --all-targets clean.")
//! @yah:next("R004-F2: chunk raw_consistent_copy output on page_size(), sha256 each page, diff against prior manifest, put only changed pages content-addressed + write manifest. Caller must release write conns first (snapshot/DR model) — a live writer can make the TRUNCATE checkpoint return busy; read-only WAL-replay copy is tier-2.")
//!
//! @yah:ticket(R004-F2, "Page chunking + content-addressed dedup upload (changed pages only) + manifest format")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-05-26T22:30:07Z)
//! @yah:status(review)
//! @yah:phase(P2)
//! @yah:parent(R004)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//! @yah:depends_on(R004-T1)
//! @yah:handoff("Built snapshot_dedup(db_path, &BackupTarget)->DedupOutcome on top of raw_consistent_copy: chunk on header page_size, sha256 each page, diff vs latest manifest, put only new pages content-addressed at pages/{hash}, write manifests/manifest-{nanos:020}.manifest last. Reuses tier-1a BackupTarget. DedupOutcome::{Snapshotted{total/uploaded/reused},Unchanged}. Manifest = dep-free text format (header + page_size + page_count + ordered hex hashes); to_bytes/from_bytes + image_len. load_latest_manifest is pub for F3.")
//! @yah:handoff("Measured: first snapshot of 5000 rows puts all 18 pages + 1 manifest; a 1-row insert then puts exactly 1 page + a 2nd manifest (pages/ grows by 1); both manifests reassemble to valid SQLite (5000/5001 rows). Unchanged path: re-snapshot with no change uploads nothing, writes no new manifest. Note: page-blob GC (unreferenced pages) is out of scope — flag for a later tier.")
//! @yah:verify("cargo test -p turso-backup: 9/9 green (3 dedup-F2: dedup_uploads_only_changed_pages_and_reassembles, dedup_unchanged_when_pages_identical, manifest_round_trips). clippy --all-targets clean. Run from turso-backup/ dir, NOT db_test root (root yah workspace has unrelated pre-existing compile errors in crates/yah/forms).")
//! @yah:next("R004-F3: implement restore_from_manifest(target, manifest, dest) — fetch each pages/{hash} in order, concat, write file. The test helper reassemble_latest already does this; F3 packages it + a restore_latest_dedup that calls load_latest_manifest first. Then R004-T4: harness e2e (two snapshots, assert only changed pages uploaded, restore both).")
//!
//! @yah:ticket(R004-F3, "Restore from manifest: fetch pages by hash, reassemble valid SQLite")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-05-26T22:30:07Z)
//! @yah:status(review)
//! @yah:phase(P2)
//! @yah:parent(R004)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//! @yah:depends_on(R004-F2)
//! @yah:handoff("Built restore_from_manifest(target, manifest, dest) + restore_latest_dedup(target, dest). Restore fetches each pages/{hash} in manifest order, VERIFIES content hash (corrupt/truncated blob -> loud error, not a silent-corrupt DB), concats, asserts SQLite magic, writes dest. No -wal needed (snapshot folded WAL in). restore_latest_dedup = load_latest_manifest (newest by nanos key) then restore_from_manifest; errors if no snapshot. Tier-1b counterpart to tier-1a snapshot::restore_latest.")
//! @yah:verify("cargo test -p turso-backup: 11/11 green (3 F3: restore_latest_dedup_picks_newest_and_round_trips, restore_rejects_corrupt_page; empty-prefix error covered). clippy --all-targets clean. Run from turso-backup/ dir.")
//! @yah:next("R004-T4: harness e2e against MinIO — wire writer to snapshot_dedup, take two snapshots with a mutation between, assert only changed pages uploaded (DedupOutcome.uploaded_pages), restore_latest_dedup both, sqlite3 verify. Lib round-trips use InMemory; T4 proves it on the object_store aws backend.")
//!
//! @yah:ticket(R004-T4, "Harness e2e incremental: two snapshots, assert only changed pages uploaded, restore both")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-05-26T22:30:08Z)
//! @yah:status(review)
//! @yah:phase(P2)
//! @yah:parent(R004)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//! @yah:depends_on(R004-F3)
//! @yah:handoff("GREEN e2e against MinIO. Added a tier-1b docker flow parallel to tier-1a (own bins + bucket prefix 'dedup'; tier-1a writer/verifier/run.sh untouched): harness bins dedup_writer (2 snapshots w/ a 5-row append between; ASSERTS uploaded2<uploaded1 && reused2>0 so a dedup regression fails the build) + dedup_restore (by MANIFEST_KEY or latest); shared row helpers in harness/src/lib.rs; verify_dedup.sh restores BOTH (snapshot1 by key, snapshot2 latest) + sqlite3 integrity/rows/boundary/hash; Dockerfile dedup-writer/dedup-verifier targets; compose services; ci/run_dedup.sh (staged). Lib gained dedup::restore_manifest_key (point-in-time restore by key).")
//! @yah:handoff("Result at 5000 rows + 5-row append: snapshot1 109/109 pages uploaded (446KB), snapshot2 1/109 uploaded + 108 reused (only the tail leaf changed; page 1 byte-identical since change_counter is dead + file didn't grow). Both restored to vanilla SQLite: by-key=5000 rows, latest=5005 rows, full-table hashes matched byte-for-byte, exit 0. Restoring snapshot1 by key proves its pages survived snapshot2's deduplicated upload.")
//! @yah:verify("bash ci/run_dedup.sh (staged: minio -> minio-init -> dedup-writer -> dedup-verifier; verifier exit 0 = green). Lib: cargo test -p turso-backup = 12/12, clippy --all-targets clean (run from turso-backup/ dir).")
//! @yah:next("R004 complete pending sign-off: all 4 children (T1/F2/F3/T4) in review. Followups filed: page-blob GC = relay R006; live-writer path = R005-F4. Tier 2 (R005, WAL-frame streaming) is the next major relay.")
//!
//! @yah:relay(R006, "Tier 1b — snapshot retention + page-blob GC")
//! @yah:at(2026-05-27T03:10:00Z)
//! @yah:status(open)
//! @yah:phase(P2)
//! @yah:parent(Q002)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//!
//! @yah:ticket(R006-F1, "Retention + mark-and-sweep GC: keep the most recent N manifests, delete pages/{hash} blobs no retained manifest references, prune the pruned manifests")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-05-27T03:10:01Z)
//! @yah:status(review)
//! @yah:phase(P2)
//! @yah:parent(R006)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//! @yah:gotcha("R004's snapshot_dedup never deletes pages — pages/{hash} blobs accumulate forever as content changes. GC = compute the union of page hashes across the retained manifests, delete every pages/* blob outside that union, then delete the non-retained manifests. Design the retained-set so a concurrent restore can't lose a page mid-sweep (delete manifests last / GC infrequently).")
//! @yah:handoff("Implemented dedup::gc_dedup(target, keep_n) -> GcOutcome. Mark-and-sweep: list manifests/, the lexically-greatest keep_n are retained; download each, union their page_hashes into the retain set; list pages/, delete every blob whose leaf isn't in the retain union; finally delete the pruned manifests. Order is pages-first / manifests-last so a retained manifest never references a missing page at any moment of a partial GC, and restore_latest_dedup (which targets the newest = retained) is safe through the whole sweep.")
//! @yah:handoff("GcOutcome reports {retained_manifests, pruned_manifests, deleted_pages, retained_pages}. keep_n >= total is a no-op (saturating_sub). restore_manifest_key against a soon-to-be-pruned manifest is documented as inherently racy — schedule GC infrequently, in windows where no point-in-time restores are in flight.")
//! @yah:next("R006-T2: the F1 lib tests already cover keep-N retains restorable + orphan pages deleted + newest still restores byte-for-byte. The piece T2 still owns: prove an *in-window older* snapshot (e.g. GC keep_n=2 with 3 snapshots → assert older retained manifest still restores byte-for-byte via restore_manifest_key). Optionally extend the dedup-writer/verifier harness to call gc_dedup between snapshots and re-verify both restores against MinIO, parallel to R004-T4's tier-1b e2e shape.")
//! @yah:verify("cargo test -p turso-backup: 41/41 lib green (3 new GC tests: gc_prunes_older_manifests_and_sweeps_orphan_pages, gc_keep_n_at_or_above_total_is_noop, gc_keeps_pages_shared_with_retained_manifests). cargo clippy --all-targets clean. Run from turso-backup/ dir.")
//!
//! @yah:ticket(R006-T2, "Test GC: keep-N retains restorable snapshots, orphaned pages deleted, latest + an in-window older snapshot still restore byte-for-byte")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-05-27T03:10:02Z)
//! @yah:status(review)
//! @yah:phase(P2)
//! @yah:parent(R006)
//! @arch:see(.yah/docs/working/turso-s3-backup.md)
//! @yah:depends_on(R006-F1)
//! @yah:handoff("Added gc_keep_two_lets_in_window_older_snapshot_restore_byte_for_byte: takes 3 snapshots with mutations between, restores the middle snapshot via restore_manifest_key BEFORE GC as ground-truth bytes, runs gc_dedup(target, 2), then asserts (a) snapshot 1's pruned key errors on restore, (b) the in-window older snapshot 2 restores BYTE-FOR-BYTE equal to the pre-GC bytes, (c) restore_latest_dedup picks snapshot 3 with the post-second-mutation row count. Locks in the property that GC's mark set covers every retained manifest, not just the one restore_latest_dedup picks.")
//! @yah:handoff("F1's three tests already covered keep-N retain + orphan delete + newest-restores-byte-for-byte + shared-page survival; T2 added the missing piece (in-window older snapshot byte-equality). 42/42 lib tests green, clippy clean.")
//! @yah:handoff("NB: the turso-backup crate was relocated from external/db_test/turso-backup to external/turso-backup during this session — board tools now need path=/Users/user/ss/yah to resolve R006 tickets.")
//! @yah:next("R006 complete pending sign-off: both children (F1, T2) in review. Optional follow-up — extend dedup-writer/verifier docker harness to call gc_dedup between snapshots and re-verify both restores against MinIO, parallel to R004-T4's tier-1b e2e shape. Not blocking R006 archive.")
//! @yah:verify("cd /Users/user/ss/yah/external/turso-backup && cargo test --lib: 42/42 green (4 GC tests under dedup::tests::gc*). cargo clippy --all-targets clean.")

use crate::snapshot::BackupTarget;
use anyhow::{Context, Result};
use object_store::path::Path as ObjPath;
use object_store::{ObjectStore, ObjectStoreExt};
use sha2::{Digest, Sha256};
use std::collections::{HashMap, HashSet};
use std::time::{SystemTime, UNIX_EPOCH};
use turso::Builder;

/// SQLite stores the database page size as a big-endian `u16` at byte offset 16
/// of the file header. The encoded value `1` means the maximum 65 536-byte page.
const PAGE_SIZE_OFFSET: usize = 16;

/// The SQLite file-format magic, present at the start of every valid database
/// (and every `VACUUM INTO` / raw-copy image we produce).
const SQLITE_MAGIC: &[u8] = b"SQLite format 3\0";

/// Read the page size (in bytes) from a SQLite file image's header.
///
/// Tier 1b chunks on **real page boundaries**, so the dedup unit is whatever the
/// engine actually uses — read it from the header rather than assuming 4 KB.
pub fn page_size(image: &[u8]) -> Result<usize> {
    anyhow::ensure!(image.starts_with(SQLITE_MAGIC), "not a SQLite database (bad magic)");
    anyhow::ensure!(image.len() >= PAGE_SIZE_OFFSET + 2, "image too short for a header");
    let raw = u16::from_be_bytes([image[PAGE_SIZE_OFFSET], image[PAGE_SIZE_OFFSET + 1]]);
    Ok(if raw == 1 { 65_536 } else { raw as usize })
}

/// Take a raw, point-in-time-consistent byte image of the database at `db_path`,
/// preserving its native page layout. This is the tier-1b foundation and is
/// deliberately **not** `VACUUM INTO`: vacuum repacks the b-tree into a fresh
/// file, relocating pages and defeating content-addressed dedup.
///
/// We fold any pending WAL frames into the main file with a `TRUNCATE`
/// checkpoint, then read the main file. After the checkpoint the main file is
/// the complete, self-consistent image — page `N` lives at byte offset
/// `(N - 1) * page_size`, so a logically-unchanged page hashes identically
/// across snapshots. That fixed-offset stability is the precondition that makes
/// "upload only changed pages" actually save bytes (confirmed by the spike tests
/// below: a one-row insert into a 5 000-row DB rewrites only a handful of pages
/// via this path, but a large fraction of them via `VACUUM INTO`).
///
/// Consistency comes from folding **main + WAL** into one image: any committed
/// frame still sitting in `-wal` is written back to its page slot, so the
/// returned bytes open in vanilla SQLite with no sidecar.
///
/// Assumes no live writer holds the WAL — the snapshot/DR model, where the
/// caller has released its write connections (as the harness writer does before
/// backing up). A concurrent writer can make a `TRUNCATE` checkpoint return
/// `busy`; the read-only "copy main + replay WAL frames ourselves" path that
/// would cover live writers is tier-2 territory.
pub async fn raw_consistent_copy(db_path: &str) -> Result<Vec<u8>> {
    // Fold WAL -> main and reset the WAL so the main file alone is complete.
    {
        let db = Builder::new_local(db_path)
            .build()
            .await
            .with_context(|| format!("opening turso db {db_path}"))?;
        let conn = db.connect().context("connecting to turso db")?;
        let mut rows = conn
            .query("PRAGMA wal_checkpoint(TRUNCATE)", ())
            .await
            .context("PRAGMA wal_checkpoint(TRUNCATE)")?;
        while rows.next().await.context("draining wal_checkpoint")?.is_some() {}
    } // drop the connection before reading the file off disk

    let bytes = std::fs::read(db_path).with_context(|| format!("reading db image {db_path}"))?;
    anyhow::ensure!(
        bytes.starts_with(SQLITE_MAGIC),
        "raw copy of {db_path} is not a SQLite database"
    );
    Ok(bytes)
}

/// The first line of every serialized [`Manifest`] — magic + format version.
const MANIFEST_HEADER: &str = "TURSO-BACKUP MANIFEST v1";

/// A per-snapshot manifest: the page size plus the ordered content hashes of the
/// database's pages. Page `i` of the restored file is the blob stored at
/// `pages/{page_hashes[i]}`, so the manifest alone is enough to reassemble the
/// whole image (see [`restore_from_manifest`]).
///
/// Serialized as a small, dependency-free text format (no serde):
///
/// ```text
/// TURSO-BACKUP MANIFEST v1
/// page_size 4096
/// page_count 18
/// <hex sha256 of page 0>
/// <hex sha256 of page 1>
/// ...
/// ```
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Manifest {
    pub page_size: usize,
    /// Lowercase-hex SHA-256 of each page, in file order.
    pub page_hashes: Vec<String>,
}

impl Manifest {
    /// Serialize to the text wire format.
    pub fn to_bytes(&self) -> Vec<u8> {
        let mut s = String::with_capacity(64 + self.page_hashes.len() * 65);
        s.push_str(MANIFEST_HEADER);
        s.push('\n');
        s.push_str(&format!("page_size {}\n", self.page_size));
        s.push_str(&format!("page_count {}\n", self.page_hashes.len()));
        for h in &self.page_hashes {
            s.push_str(h);
            s.push('\n');
        }
        s.into_bytes()
    }

    /// Parse the text wire format, validating the header and page count.
    pub fn from_bytes(bytes: &[u8]) -> Result<Manifest> {
        let text = std::str::from_utf8(bytes).context("manifest is not valid UTF-8")?;
        let mut lines = text.lines();
        let header = lines.next().context("empty manifest")?;
        anyhow::ensure!(header == MANIFEST_HEADER, "unexpected manifest header: {header:?}");
        let page_size = parse_header_field(lines.next(), "page_size")?;
        let page_count = parse_header_field(lines.next(), "page_count")?;
        let page_hashes: Vec<String> = lines.map(|l| l.trim().to_string()).collect();
        anyhow::ensure!(
            page_hashes.len() == page_count,
            "manifest page_count {page_count} != {} hash lines",
            page_hashes.len()
        );
        Ok(Manifest { page_size, page_hashes })
    }

    /// Total reassembled image size in bytes.
    pub fn image_len(&self) -> usize {
        self.page_size * self.page_hashes.len()
    }
}

/// Parse a `"<name> <usize>"` header line.
fn parse_header_field(line: Option<&str>, name: &str) -> Result<usize> {
    let line = line.with_context(|| format!("manifest missing {name}"))?;
    let value = line
        .strip_prefix(name)
        .and_then(|rest| rest.trim().parse::<usize>().ok())
        .with_context(|| format!("malformed manifest {name} line: {line:?}"))?;
    Ok(value)
}

/// Outcome of a [`snapshot_dedup`] call.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DedupOutcome {
    /// A new manifest was written. `uploaded_pages` is the number of distinct
    /// pages put this run (changed or new); `reused_pages` were already present
    /// from an earlier snapshot and so were *not* re-uploaded.
    Snapshotted {
        manifest_key: String,
        total_pages: usize,
        uploaded_pages: usize,
        reused_pages: usize,
    },
    /// The page set was byte-identical to the latest manifest, so nothing was
    /// uploaded and no new manifest was written — the prior one still stands.
    Unchanged { manifest_key: String },
}

/// Take a raw consistent snapshot of the database at `db_path` and upload only
/// the pages that changed since the last snapshot, plus a per-snapshot manifest.
/// Tier 1b.
///
/// 1. [`raw_consistent_copy`] the database (page layout preserved — the
///    precondition for dedup, see that fn).
/// 2. Split on the header [`page_size`] and SHA-256 each page.
/// 3. Load the latest manifest (if any). If this snapshot's ordered page hashes
///    are identical, return [`Unchanged`] without uploading.
/// 4. Otherwise upload every page whose hash isn't already referenced by the
///    prior manifest — content-addressed at `{prefix}/pages/{hash}`, so
///    identical pages (within or across snapshots) are stored once — then write
///    the manifest at `{prefix}/manifests/manifest-{unix_nanos:020}.manifest`.
///
/// The prior-manifest diff is conservative: a page present in the store but not
/// in the prior manifest is re-uploaded (idempotent — same content-addressed
/// key), so the failure direction is a redundant put, never a missing page.
///
/// [`Unchanged`]: DedupOutcome::Unchanged
pub async fn snapshot_dedup(db_path: &str, target: &BackupTarget) -> Result<DedupOutcome> {
    let image = raw_consistent_copy(db_path).await?;
    snapshot_dedup_image(&image, target).await
}

/// Steps 2–4 of [`snapshot_dedup`] against an image the caller already holds.
///
/// R850-F1 split this out for the caller [`snapshot_dedup`] cannot serve: a tail
/// running **beside a live application** that holds the database open. Step 1's
/// [`raw_consistent_copy`] takes a `PRAGMA wal_checkpoint(TRUNCATE)` through an
/// ordinary writable connection, and turso locks the whole file at open — so
/// against a held database that step does not merely block the checkpoint, it
/// fails to open at all (measured: `examples/appliance_tail_probe.rs`, A1).
/// Such a caller produces its image with
/// [`stream::raw_consistent_copy_live`](crate::stream::raw_consistent_copy_live)
/// instead, which opens read-only and validates optimistically, and hands it
/// here.
///
/// The page-layout precondition [`snapshot_dedup`] documents is unchanged and is
/// now the caller's to keep: the image must be a raw page-preserving copy, not
/// `VACUUM INTO` output, or every page hash moves on every run and the dedup
/// stores a fresh full copy each time.
pub async fn snapshot_dedup_image(image: &[u8], target: &BackupTarget) -> Result<DedupOutcome> {
    let ps = page_size(image)?;

    let page_hashes: Vec<String> = image.chunks(ps).map(sha256_hex).collect();
    let total_pages = page_hashes.len();

    // Latest manifest = the prior snapshot's page set (empty on the first run).
    let prior = load_latest_manifest(target).await?;
    if let Some((key, manifest)) = &prior {
        if manifest.page_hashes == page_hashes {
            return Ok(DedupOutcome::Unchanged { manifest_key: key.clone() });
        }
    }
    let already: HashSet<&str> = prior
        .as_ref()
        .map(|(_, m)| m.page_hashes.iter().map(String::as_str).collect())
        .unwrap_or_default();

    // Distinct pages new to this snapshot (dedups identical pages within it too).
    let mut to_upload: HashMap<&str, &[u8]> = HashMap::new();
    for (hash, page) in page_hashes.iter().zip(image.chunks(ps)) {
        if !already.contains(hash.as_str()) {
            to_upload.insert(hash.as_str(), page);
        }
    }
    let uploaded_pages = to_upload.len();
    for (hash, page) in &to_upload {
        let key = page_key(target, hash);
        target
            .store
            .put(&key, page.to_vec().into())
            .await
            .with_context(|| format!("uploading page {key}"))?;
    }

    // Write the manifest last, so a manifest never references a missing page.
    let manifest = Manifest { page_size: ps, page_hashes };
    let key = manifest_key(target, unix_nanos());
    target
        .store
        .put(&key, manifest.to_bytes().into())
        .await
        .with_context(|| format!("uploading manifest {key}"))?;

    Ok(DedupOutcome::Snapshotted {
        manifest_key: key.to_string(),
        total_pages,
        uploaded_pages,
        reused_pages: total_pages - uploaded_pages,
    })
}

/// Load the most recent manifest under `{prefix}/manifests/`, returning its key
/// and parsed contents, or `None` if no snapshot has been taken yet. Manifest
/// keys are zero-padded nanosecond timestamps, so the lexically-greatest key is
/// the newest (clock-skew-immune, matching tier 1a's `restore_latest`). Shared
/// with restore (R004-F3).
pub async fn load_latest_manifest(target: &BackupTarget) -> Result<Option<(String, Manifest)>> {
    let prefix = join_key(&target.prefix, "manifests");
    let listing = target
        .store
        .list_with_delimiter(Some(&prefix))
        .await
        .with_context(|| format!("listing manifests under {prefix}"))?;
    let Some(newest) = listing.objects.into_iter().max_by(|a, b| a.location.cmp(&b.location)) else {
        return Ok(None);
    };
    let bytes = target
        .store
        .get(&newest.location)
        .await
        .with_context(|| format!("downloading manifest {}", newest.location))?
        .bytes()
        .await
        .with_context(|| format!("reading manifest body {}", newest.location))?;
    let manifest = Manifest::from_bytes(&bytes)
        .with_context(|| format!("parsing manifest {}", newest.location))?;
    Ok(Some((newest.location.to_string(), manifest)))
}

/// Reassemble a database file from a manifest + content-addressed pages, writing
/// it to `dest`. Tier 1b restore.
///
/// Fetches each `pages/{hash}` in manifest order, **verifies its content hash**
/// (the page is content-addressed, so a hash mismatch means a corrupt or
/// truncated blob — fail loudly rather than write a silently-broken database),
/// concatenates, and writes the image. The result is a valid vanilla-SQLite file
/// needing no `-wal` sidecar — the snapshot folded WAL into the image.
pub async fn restore_from_manifest(
    target: &BackupTarget,
    manifest: &Manifest,
    dest: &str,
) -> Result<()> {
    let mut image = Vec::with_capacity(manifest.image_len());
    for (i, hash) in manifest.page_hashes.iter().enumerate() {
        let key = page_key(target, hash);
        let bytes = target
            .store
            .get(&key)
            .await
            .with_context(|| format!("fetching page {i} ({key})"))?
            .bytes()
            .await
            .with_context(|| format!("reading page {i} body ({key})"))?;
        let got = sha256_hex(&bytes);
        anyhow::ensure!(
            &got == hash,
            "page {i} ({key}) content hash {got} != manifest {hash} — corrupt or truncated blob"
        );
        image.extend_from_slice(&bytes);
    }
    anyhow::ensure!(
        image.starts_with(SQLITE_MAGIC),
        "reassembled image is not a SQLite database"
    );
    std::fs::write(dest, &image).with_context(|| format!("writing restored db to {dest}"))?;
    Ok(())
}

/// Restore the most recent snapshot for `target` to `dest`, returning the
/// manifest key it restored from. Errors if no snapshot exists under the prefix.
/// This is the tier-1b counterpart to tier 1a's `snapshot::restore_latest`.
pub async fn restore_latest_dedup(target: &BackupTarget, dest: &str) -> Result<String> {
    let (key, manifest) = load_latest_manifest(target)
        .await?
        .context("no snapshot manifest found under prefix")?;
    restore_from_manifest(target, &manifest, dest).await?;
    Ok(key)
}

/// Restore a *specific* snapshot, named by its manifest object key (as returned
/// in [`DedupOutcome::Snapshotted::manifest_key`]), to `dest`. Use this for a
/// point-in-time restore of an older snapshot; [`restore_latest_dedup`] covers
/// the common "newest" case. Fetches and parses the manifest object, then defers
/// to [`restore_from_manifest`] (so the same per-page content-hash verification
/// applies).
pub async fn restore_manifest_key(target: &BackupTarget, manifest_key: &str, dest: &str) -> Result<()> {
    let key = ObjPath::from(manifest_key);
    let bytes = target
        .store
        .get(&key)
        .await
        .with_context(|| format!("fetching manifest {key}"))?
        .bytes()
        .await
        .with_context(|| format!("reading manifest body {key}"))?;
    let manifest =
        Manifest::from_bytes(&bytes).with_context(|| format!("parsing manifest {key}"))?;
    restore_from_manifest(target, &manifest, dest).await
}

/// Outcome of a [`gc_dedup`] call.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GcOutcome {
    /// Number of manifests retained (the `keep_n` newest, or fewer if the store
    /// holds less).
    pub retained_manifests: usize,
    /// Number of older manifests deleted from `manifests/`.
    pub pruned_manifests: usize,
    /// Number of `pages/{hash}` blobs deleted (referenced by no retained
    /// manifest).
    pub deleted_pages: usize,
    /// Number of distinct page hashes referenced by the retained manifests.
    /// Every blob with one of these hashes survives the sweep.
    pub retained_pages: usize,
}

/// Mark-and-sweep GC: keep the `keep_n` newest manifests, delete every
/// `pages/{hash}` blob no retained manifest references, then prune the older
/// manifests.
///
/// Manifest keys are zero-padded nanosecond timestamps, so lexical order is
/// chronological — the `keep_n` lexically-greatest keys are the newest
/// snapshots. If the store holds `<= keep_n` manifests, nothing is pruned.
///
/// ## Sweep order
///
/// 1. List `manifests/`, split into retained (newest `keep_n`) vs pruned.
/// 2. Download + parse the retained manifests; union their `page_hashes`.
/// 3. List `pages/`, delete every blob whose leaf isn't in the retained union.
/// 4. **Delete the pruned manifests last.**
///
/// Concurrent-restore safety: [`restore_latest_dedup`] always targets the
/// lexically-greatest manifest, which is in the retain set, and *all* of its
/// pages are in the retain union — so it survives a concurrent GC. The
/// `manifests-last` ordering means that even if we crash mid-sweep, every
/// surviving manifest still resolves: orphan pages may briefly outlive their
/// (already-pruned) referrers, but no retained manifest ever points at a
/// missing page. The reverse order would risk a half-deleted retain set.
///
/// `restore_manifest_key` against a manifest in the prune set is inherently a
/// race — the caller is asking to restore something we've decided to discard.
/// Schedule GC to run when no point-in-time restores are in flight.
pub async fn gc_dedup(target: &BackupTarget, keep_n: usize) -> Result<GcOutcome> {
    // 1. List manifests, sort lexically (== chronologically).
    let manifests_prefix = join_key(&target.prefix, "manifests");
    let manifests_listing = target
        .store
        .list_with_delimiter(Some(&manifests_prefix))
        .await
        .with_context(|| format!("listing manifests under {manifests_prefix}"))?;
    let mut all_manifests: Vec<ObjPath> =
        manifests_listing.objects.into_iter().map(|o| o.location).collect();
    all_manifests.sort();

    // Split: the newest `keep_n` retained, the rest pruned. If keep_n exceeds
    // the count, every manifest is retained and there's nothing to prune.
    let total = all_manifests.len();
    let split = total.saturating_sub(keep_n);
    let pruned: Vec<ObjPath> = all_manifests.iter().take(split).cloned().collect();
    let retained: Vec<ObjPath> = all_manifests.into_iter().skip(split).collect();

    // 2. Compute the union of page hashes referenced by the retained manifests.
    let mut retained_hashes: HashSet<String> = HashSet::new();
    for key in &retained {
        let bytes = target
            .store
            .get(key)
            .await
            .with_context(|| format!("fetching retained manifest {key}"))?
            .bytes()
            .await
            .with_context(|| format!("reading retained manifest body {key}"))?;
        let manifest = Manifest::from_bytes(&bytes)
            .with_context(|| format!("parsing retained manifest {key}"))?;
        retained_hashes.extend(manifest.page_hashes);
    }

    // 3. List pages, delete every blob whose hash isn't in the retain union.
    // Pages are stored at `{prefix}/pages/{hash}` with no further slashes, so
    // list_with_delimiter returns each one directly under `objects`.
    let pages_prefix = join_key(&target.prefix, "pages");
    let pages_listing = target
        .store
        .list_with_delimiter(Some(&pages_prefix))
        .await
        .with_context(|| format!("listing pages under {pages_prefix}"))?;
    let mut deleted_pages = 0usize;
    for obj in pages_listing.objects {
        let leaf = obj
            .location
            .filename()
            .with_context(|| format!("page key has no leaf: {}", obj.location))?
            .to_string();
        if !retained_hashes.contains(&leaf) {
            target
                .store
                .delete(&obj.location)
                .await
                .with_context(|| format!("deleting orphan page {}", obj.location))?;
            deleted_pages += 1;
        }
    }

    // 4. Delete the pruned manifests LAST (see fn doc — keeps retained
    // manifests fully resolvable through the entire sweep).
    for key in &pruned {
        target
            .store
            .delete(key)
            .await
            .with_context(|| format!("deleting pruned manifest {key}"))?;
    }

    Ok(GcOutcome {
        retained_manifests: retained.len(),
        pruned_manifests: pruned.len(),
        deleted_pages,
        retained_pages: retained_hashes.len(),
    })
}

/// Object key for a content-addressed page blob.
fn page_key(target: &BackupTarget, hash: &str) -> ObjPath {
    join_key(&target.prefix, &format!("pages/{hash}"))
}

/// Object key for a snapshot's manifest, taken at `unix_nanos` (zero-padded so
/// lexical order matches chronological order).
fn manifest_key(target: &BackupTarget, unix_nanos: u128) -> ObjPath {
    join_key(&target.prefix, &format!("manifests/manifest-{unix_nanos:020}.manifest"))
}

/// Join an object-store prefix and a leaf into a normalized [`ObjPath`],
/// tolerating empty/slash-padded prefixes.
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}"))
    }
}

/// Lowercase-hex SHA-256 of a byte buffer.
fn sha256_hex(bytes: &[u8]) -> String {
    let mut hasher = Sha256::new();
    hasher.update(bytes);
    hex::encode(hasher.finalize())
}

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

#[cfg(test)]
mod tests {
    use super::*;
    use object_store::memory::InMemory;
    use std::sync::Arc;

    /// A throwaway db path under the OS temp dir, cleaned up (with sidecars) on drop.
    struct TempDb(std::path::PathBuf);
    impl TempDb {
        fn new(tag: &str) -> Self {
            TempDb(std::env::temp_dir().join(format!(
                "turso-dedup-test-{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 suffix in ["", "-wal", "-shm"] {
                let _ = std::fs::remove_file(format!("{}{suffix}", self.0.display()));
            }
        }
    }

    async fn seed_db(path: &str, rows: i64) {
        let db = 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 0..rows {
            conn.execute("INSERT INTO t (id, v) VALUES (?, ?)", (i, format!("v{i}")))
                .await
                .unwrap();
        }
        conn.execute("COMMIT", ()).await.unwrap();
    }

    async fn insert_rows(path: &str, start: i64, n: i64) {
        let db = Builder::new_local(path).build().await.unwrap();
        let conn = db.connect().unwrap();
        for i in 0..n {
            conn.execute("INSERT INTO t (id, v) VALUES (?, ?)", (start + i, format!("v{}", start + i)))
                .await
                .unwrap();
        }
    }

    async fn count_rows(path: &str) -> i64 {
        let db = 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()
    }

    /// `VACUUM INTO` a fresh file and return its bytes — the contrast case that
    /// shows why tier 1b copies raw instead of vacuuming.
    async fn vacuum_image(db_path: &str) -> Vec<u8> {
        let out = std::env::temp_dir().join(format!(
            "turso-dedup-vac-{}-{}.db",
            std::process::id(),
            unix_nanos()
        ));
        let _ = std::fs::remove_file(&out);
        let db = Builder::new_local(db_path).build().await.unwrap();
        let conn = db.connect().unwrap();
        let escaped = out.to_string_lossy().replace('\'', "''");
        let mut rows = conn.query(&format!("VACUUM INTO '{escaped}'"), ()).await.unwrap();
        while rows.next().await.unwrap().is_some() {}
        drop(conn);
        let bytes = std::fs::read(&out).unwrap();
        let _ = std::fs::remove_file(&out);
        bytes
    }

    /// Count pages that differ at the same offset between two images, padding the
    /// shorter image's tail to `total` (a grown file = appended pages = changed).
    fn changed_pages(a: &[u8], b: &[u8], ps: usize) -> (usize, usize) {
        let pa: Vec<&[u8]> = a.chunks(ps).collect();
        let pb: Vec<&[u8]> = b.chunks(ps).collect();
        let total = pa.len().max(pb.len());
        let mut changed = 0;
        for i in 0..total {
            if pa.get(i) != pb.get(i) {
                changed += 1;
            }
        }
        (changed, total)
    }

    /// SPIKE R004-T1 (headline): a raw consistent copy keeps unchanged pages at
    /// fixed offsets, so a one-row mutation rewrites only a few pages — the
    /// precondition for content-addressed dedup.
    #[tokio::test]
    async fn raw_copy_keeps_page_offsets_stable() {
        let src = TempDb::new("offsets");
        seed_db(src.path(), 5000).await;

        let a = raw_consistent_copy(src.path()).await.unwrap();
        let ps = page_size(&a).unwrap();
        assert_eq!(ps, 4096, "expected turso's default 4 KB page, got {ps}");

        // A single-row insert: the smallest meaningful mutation.
        insert_rows(src.path(), 1_000_000, 1).await;
        let b = raw_consistent_copy(src.path()).await.unwrap();
        assert_eq!(page_size(&b).unwrap(), ps, "page size must not change");

        let (changed, total) = changed_pages(&a, &b, ps);
        eprintln!(
            "[R004-T1] raw copy: {changed}/{total} pages changed by a 1-row insert (page_size={ps})"
        );
        assert!(changed >= 1, "the mutation must change at least one page");
        assert!(
            changed * 4 < total,
            "raw copy not dedup-friendly: {changed}/{total} pages changed by one row"
        );

        // The second image is itself complete and consistent (opens in turso).
        let restored = TempDb::new("offsets-restored");
        std::fs::write(restored.path(), &b).unwrap();
        assert_eq!(count_rows(restored.path()).await, 5001);
    }

    /// SPIKE R004-T1 (contrast): the honest comparison of raw copy vs
    /// `VACUUM INTO` as the dedup base, across two mutation shapes.
    ///
    /// - **Append** (1-row insert): the two are *equivalent* — both touch a
    ///   single page. The working doc's blanket "vacuum reshuffles and defeats
    ///   dedup" does NOT hold for append-only growth.
    /// - **Reclamation** (delete a low-id block): vacuum repacks the surviving
    ///   rows from page 1, renumbering them, so a large fraction of pages
    ///   change. A raw copy leaves every surviving page at its original offset,
    ///   so only the freed region changes.
    ///
    /// The takeaway for F2: raw copy's offset stability is a *structural
    /// guarantee* (page N is always at `(N-1)*ps`, existing pages never
    /// renumber), independent of workload. Vacuum's dedup-friendliness is
    /// incidental and collapses under reclamation — so tier 1b copies raw.
    #[tokio::test]
    async fn raw_copy_beats_vacuum_as_dedup_base() {
        let src = TempDb::new("contrast");
        seed_db(src.path(), 5000).await;

        let raw_a = raw_consistent_copy(src.path()).await.unwrap();
        let vac_a = vacuum_image(src.path()).await;
        let ps = page_size(&raw_a).unwrap();

        // Phase 1 — append one row: raw and vacuum should be ~equivalent.
        insert_rows(src.path(), 1_000_000, 1).await;
        let raw_b = raw_consistent_copy(src.path()).await.unwrap();
        let vac_b = vacuum_image(src.path()).await;
        let (raw_app, raw_app_total) = changed_pages(&raw_a, &raw_b, ps);
        let (vac_app, vac_app_total) = changed_pages(&vac_a, &vac_b, ps);
        eprintln!(
            "[R004-T1] append 1 row -> raw: {raw_app}/{raw_app_total} | vacuum: {vac_app}/{vac_app_total} (expected ~equal)"
        );

        // Phase 2 — free a low-id block: vacuum renumbers survivors, raw doesn't.
        let src2 = TempDb::new("contrast-reclaim");
        seed_db(src2.path(), 5000).await;
        let raw_c = raw_consistent_copy(src2.path()).await.unwrap();
        let vac_c = vacuum_image(src2.path()).await;
        {
            let db = Builder::new_local(src2.path()).build().await.unwrap();
            let conn = db.connect().unwrap();
            conn.execute("DELETE FROM t WHERE id < 2000", ()).await.unwrap();
        }
        let raw_d = raw_consistent_copy(src2.path()).await.unwrap();
        let vac_d = vacuum_image(src2.path()).await;
        let (raw_rec, raw_rec_total) = changed_pages(&raw_c, &raw_d, ps);
        let (vac_rec, vac_rec_total) = changed_pages(&vac_c, &vac_d, ps);
        eprintln!(
            "[R004-T1] delete low block -> raw: {raw_rec}/{raw_rec_total} | vacuum: {vac_rec}/{vac_rec_total} (vacuum renumbers)"
        );

        // The defensible, workload-independent claim: under reclamation the raw
        // copy preserves more pages than vacuum does.
        assert!(
            vac_rec > raw_rec,
            "under reclamation expected vacuum to perturb more pages than raw copy \
             (raw {raw_rec}, vacuum {vac_rec})"
        );
    }

    /// SPIKE R004-T1: `raw_consistent_copy` folds main + WAL into one image, so
    /// the returned bytes are complete with no `-wal` sidecar.
    #[tokio::test]
    async fn raw_copy_folds_wal_and_is_self_contained() {
        let src = TempDb::new("walfold");
        seed_db(src.path(), 100).await;
        insert_rows(src.path(), 200, 50).await;

        let img = raw_consistent_copy(src.path()).await.unwrap();
        assert!(img.starts_with(SQLITE_MAGIC));

        // Write ONLY the main image to a fresh path (no -wal/-shm): a complete
        // image means the WAL frames were folded in.
        let restored = TempDb::new("walfold-restored");
        std::fs::write(restored.path(), &img).unwrap();
        assert_eq!(count_rows(restored.path()).await, 150);
    }

    // ---- R004-F2: page chunking + content-addressed dedup upload + manifest ----

    /// Number of objects stored under `{prefix}/{leaf}/`.
    async fn count_under(target: &BackupTarget, leaf: &str) -> usize {
        target
            .store
            .list_with_delimiter(Some(&join_key(&target.prefix, leaf)))
            .await
            .unwrap()
            .objects
            .len()
    }

    /// Reassemble the image described by the latest manifest, fetching each page
    /// blob by its content hash (what R004-F3 will package as `restore`).
    async fn reassemble_latest(target: &BackupTarget) -> Vec<u8> {
        let (_, m) = load_latest_manifest(target).await.unwrap().unwrap();
        let mut out = Vec::with_capacity(m.image_len());
        for h in &m.page_hashes {
            let bytes = target.store.get(&page_key(target, h)).await.unwrap().bytes().await.unwrap();
            out.extend_from_slice(&bytes);
        }
        out
    }

    #[test]
    fn manifest_round_trips() {
        let m = Manifest {
            page_size: 4096,
            page_hashes: vec!["aa".repeat(32), "bb".repeat(32)],
        };
        assert_eq!(Manifest::from_bytes(&m.to_bytes()).unwrap(), m);
        assert_eq!(m.image_len(), 8192);
        assert!(Manifest::from_bytes(b"not a manifest\n").is_err());
        assert!(Manifest::from_bytes(&[0xff, 0xfe]).is_err());
    }

    /// First snapshot uploads every distinct page + a manifest; a one-row change
    /// then uploads only the few changed pages and writes a second manifest. Both
    /// manifests reassemble to a valid SQLite with the expected row count.
    #[tokio::test]
    async fn dedup_uploads_only_changed_pages_and_reassembles() {
        let src = TempDb::new("dedup-src");
        seed_db(src.path(), 5000).await;
        let target = BackupTarget { store: Arc::new(InMemory::new()), prefix: "backups".into() };

        // First snapshot: nothing prior -> every distinct page is new.
        let total = match snapshot_dedup(src.path(), &target).await.unwrap() {
            DedupOutcome::Snapshotted { total_pages, uploaded_pages, reused_pages, .. } => {
                assert_eq!(reused_pages, 0, "no prior snapshot -> nothing reused");
                assert_eq!(uploaded_pages, count_under(&target, "pages").await, "every put is a distinct page");
                total_pages
            }
            other => panic!("expected Snapshotted, got {other:?}"),
        };
        assert_eq!(count_under(&target, "manifests").await, 1);
        let pages_after_first = count_under(&target, "pages").await;

        // Reassembles to a valid SQLite with all 5000 rows.
        let img = reassemble_latest(&target).await;
        assert!(img.starts_with(SQLITE_MAGIC));
        let r1 = TempDb::new("dedup-r1");
        std::fs::write(r1.path(), &img).unwrap();
        assert_eq!(count_rows(r1.path()).await, 5000);

        // One-row mutation -> only a handful of changed pages uploaded.
        insert_rows(src.path(), 1_000_000, 1).await;
        match snapshot_dedup(src.path(), &target).await.unwrap() {
            DedupOutcome::Snapshotted { uploaded_pages, total_pages, .. } => {
                assert!(
                    uploaded_pages >= 1 && uploaded_pages * 4 < total_pages,
                    "expected few changed pages, got {uploaded_pages}/{total_pages}"
                );
                // Content-addressed store grew by exactly the new pages.
                assert_eq!(
                    count_under(&target, "pages").await - pages_after_first,
                    uploaded_pages
                );
            }
            other => panic!("expected Snapshotted, got {other:?}"),
        }
        assert_eq!(count_under(&target, "manifests").await, 2, "second snapshot -> second manifest");
        assert_eq!(total, 18, "sanity: 5000 short rows -> 18 pages");

        // The newest manifest reassembles to the post-mutation database.
        let img2 = reassemble_latest(&target).await;
        let r2 = TempDb::new("dedup-r2");
        std::fs::write(r2.path(), &img2).unwrap();
        assert_eq!(count_rows(r2.path()).await, 5001);
    }

    /// Re-snapshotting an unchanged database uploads nothing and writes no new
    /// manifest — the prior one still stands.
    #[tokio::test]
    async fn dedup_unchanged_when_pages_identical() {
        let src = TempDb::new("dedup-unch");
        seed_db(src.path(), 50).await;
        let target = BackupTarget { store: Arc::new(InMemory::new()), prefix: String::new() };

        let first_key = match snapshot_dedup(src.path(), &target).await.unwrap() {
            DedupOutcome::Snapshotted { manifest_key, .. } => manifest_key,
            other => panic!("expected Snapshotted, got {other:?}"),
        };
        let pages = count_under(&target, "pages").await;

        match snapshot_dedup(src.path(), &target).await.unwrap() {
            DedupOutcome::Unchanged { manifest_key } => assert_eq!(manifest_key, first_key),
            other => panic!("expected Unchanged, got {other:?}"),
        }
        assert_eq!(count_under(&target, "manifests").await, 1, "no new manifest on unchanged");
        assert_eq!(count_under(&target, "pages").await, pages, "no new pages on unchanged");
    }

    // ---- R004-F3: restore from manifest ----

    /// `restore_latest_dedup` errors when empty, then restores the newest of two
    /// snapshots (post-mutation row count) as a valid vanilla-SQLite file.
    #[tokio::test]
    async fn restore_latest_dedup_picks_newest_and_round_trips() {
        let src = TempDb::new("f3-src");
        seed_db(src.path(), 200).await;
        let target = BackupTarget { store: Arc::new(InMemory::new()), prefix: "backups".into() };

        // No snapshot yet -> error, no file written.
        let dest = TempDb::new("f3-dest");
        assert!(restore_latest_dedup(&target, dest.path()).await.is_err());

        snapshot_dedup(src.path(), &target).await.unwrap();
        insert_rows(src.path(), 1_000_000, 5).await;
        let newest = match snapshot_dedup(src.path(), &target).await.unwrap() {
            DedupOutcome::Snapshotted { manifest_key, .. } => manifest_key,
            other => panic!("expected Snapshotted, got {other:?}"),
        };

        let restored_from = restore_latest_dedup(&target, dest.path()).await.unwrap();
        assert_eq!(restored_from, newest, "should restore the newest manifest");

        let on_disk = std::fs::read(dest.path()).unwrap();
        assert!(on_disk.starts_with(SQLITE_MAGIC), "restored file is not SQLite");
        assert_eq!(count_rows(dest.path()).await, 205);
    }

    /// A corrupt page blob is caught by the content-hash check rather than
    /// producing a silently-broken database.
    #[tokio::test]
    async fn restore_rejects_corrupt_page() {
        let src = TempDb::new("f3-corrupt");
        seed_db(src.path(), 100).await;
        let target = BackupTarget { store: Arc::new(InMemory::new()), prefix: String::new() };
        snapshot_dedup(src.path(), &target).await.unwrap();

        let (_, manifest) = load_latest_manifest(&target).await.unwrap().unwrap();
        // Overwrite one page blob with garbage of the right length (so only the
        // hash check, not a size check, can catch it).
        let victim = &manifest.page_hashes[0];
        target
            .store
            .put(&page_key(&target, victim), vec![0xab; manifest.page_size].into())
            .await
            .unwrap();

        let dest = TempDb::new("f3-corrupt-dest");
        let err = restore_from_manifest(&target, &manifest, dest.path()).await.unwrap_err();
        assert!(
            err.to_string().contains("content hash"),
            "expected a hash-mismatch error, got: {err}"
        );
    }

    /// `restore_manifest_key` targets an *older* snapshot by key while the latest
    /// reflects the post-mutation state — both reassemble to their own row counts.
    #[tokio::test]
    async fn restore_manifest_key_targets_older_snapshot() {
        let src = TempDb::new("f3-key-src");
        seed_db(src.path(), 100).await;
        let target = BackupTarget { store: Arc::new(InMemory::new()), prefix: "backups".into() };

        let key1 = match snapshot_dedup(src.path(), &target).await.unwrap() {
            DedupOutcome::Snapshotted { manifest_key, .. } => manifest_key,
            other => panic!("expected Snapshotted, got {other:?}"),
        };
        insert_rows(src.path(), 1_000_000, 10).await;
        snapshot_dedup(src.path(), &target).await.unwrap();

        // Older snapshot by key -> 100 rows.
        let d1 = TempDb::new("f3-key-d1");
        restore_manifest_key(&target, &key1, d1.path()).await.unwrap();
        assert_eq!(count_rows(d1.path()).await, 100);

        // Latest -> 110 rows.
        let d2 = TempDb::new("f3-key-d2");
        restore_latest_dedup(&target, d2.path()).await.unwrap();
        assert_eq!(count_rows(d2.path()).await, 110);
    }

    // ---- R006-F1: retention + page-blob GC ----

    /// Take three snapshots, GC down to the newest one. Older manifests are
    /// pruned, every page hash referenced *only* by them is swept, the pages
    /// the surviving manifest still references stay put, and the surviving
    /// snapshot restores byte-for-byte against the un-GC'd image.
    #[tokio::test]
    async fn gc_prunes_older_manifests_and_sweeps_orphan_pages() {
        let src = TempDb::new("gc-src");
        seed_db(src.path(), 5000).await;
        let target = BackupTarget { store: Arc::new(InMemory::new()), prefix: "backups".into() };

        // Three snapshots with a row appended between each — every snapshot
        // produces at least one fresh page that the next one doesn't reference.
        snapshot_dedup(src.path(), &target).await.unwrap();
        insert_rows(src.path(), 1_000_000, 1).await;
        snapshot_dedup(src.path(), &target).await.unwrap();
        insert_rows(src.path(), 2_000_000, 1).await;
        let newest_key = match snapshot_dedup(src.path(), &target).await.unwrap() {
            DedupOutcome::Snapshotted { manifest_key, .. } => manifest_key,
            other => panic!("expected Snapshotted, got {other:?}"),
        };
        assert_eq!(count_under(&target, "manifests").await, 3);
        let pages_before_gc = count_under(&target, "pages").await;

        // Snapshot the bytes-on-disk of every page the newest manifest needs,
        // so we can prove the GC leaves them byte-identical.
        let (_, newest_manifest) = load_latest_manifest(&target).await.unwrap().unwrap();
        let mut expected_pages: HashMap<String, Vec<u8>> = HashMap::new();
        for h in &newest_manifest.page_hashes {
            let bytes = target.store.get(&page_key(&target, h)).await.unwrap().bytes().await.unwrap();
            expected_pages.insert(h.clone(), bytes.to_vec());
        }

        // GC down to keep=1 (just the newest manifest).
        let outcome = gc_dedup(&target, 1).await.unwrap();
        assert_eq!(outcome.retained_manifests, 1);
        assert_eq!(outcome.pruned_manifests, 2);
        assert_eq!(outcome.retained_pages, newest_manifest.page_hashes.len());
        assert!(
            outcome.deleted_pages > 0,
            "earlier snapshots wrote at least one page the newest manifest no longer references"
        );

        // Store state matches the outcome counts.
        assert_eq!(count_under(&target, "manifests").await, 1, "only the newest manifest survives");
        assert_eq!(
            count_under(&target, "pages").await,
            pages_before_gc - outcome.deleted_pages,
            "pages/ shrank by exactly the orphan count"
        );

        // Every page the newest manifest references is still there, byte-identical.
        for (h, want) in &expected_pages {
            let got = target.store.get(&page_key(&target, h)).await.unwrap().bytes().await.unwrap();
            assert_eq!(&got[..], &want[..], "retained page {h} was mutated by GC");
        }

        // And the newest manifest still restores to a valid SQLite with the
        // post-second-mutation row count (5000 + 1 + 1 = 5002).
        let dest = TempDb::new("gc-restored");
        let restored_from = restore_latest_dedup(&target, dest.path()).await.unwrap();
        assert_eq!(restored_from, newest_key);
        assert_eq!(count_rows(dest.path()).await, 5002);
    }

    /// keep_n >= total manifests is a no-op: nothing pruned, no pages deleted.
    #[tokio::test]
    async fn gc_keep_n_at_or_above_total_is_noop() {
        let src = TempDb::new("gc-noop");
        seed_db(src.path(), 100).await;
        let target = BackupTarget { store: Arc::new(InMemory::new()), prefix: String::new() };

        snapshot_dedup(src.path(), &target).await.unwrap();
        insert_rows(src.path(), 1_000_000, 1).await;
        snapshot_dedup(src.path(), &target).await.unwrap();
        let manifests_before = count_under(&target, "manifests").await;
        let pages_before = count_under(&target, "pages").await;

        // keep_n == total
        let outcome = gc_dedup(&target, 2).await.unwrap();
        assert_eq!(outcome.pruned_manifests, 0);
        assert_eq!(outcome.deleted_pages, 0);
        assert_eq!(outcome.retained_manifests, manifests_before);

        // keep_n > total
        let outcome = gc_dedup(&target, 999).await.unwrap();
        assert_eq!(outcome.pruned_manifests, 0);
        assert_eq!(outcome.deleted_pages, 0);

        assert_eq!(count_under(&target, "manifests").await, manifests_before);
        assert_eq!(count_under(&target, "pages").await, pages_before);
    }

    /// Pages shared between a pruned and a retained manifest survive — content
    /// addressing means they're one blob, and the retain union still needs it.
    /// This is the load-bearing safety property: incremental snapshots reuse
    /// most pages, so a naive "delete every page from pruned manifests" would
    /// destroy the retained snapshot.
    #[tokio::test]
    async fn gc_keeps_pages_shared_with_retained_manifests() {
        let src = TempDb::new("gc-shared");
        seed_db(src.path(), 5000).await;
        let target = BackupTarget { store: Arc::new(InMemory::new()), prefix: "backups".into() };

        // Snapshot 1 — all pages new.
        snapshot_dedup(src.path(), &target).await.unwrap();
        let (_, m1) = load_latest_manifest(&target).await.unwrap().unwrap();
        let s1_hashes: HashSet<String> = m1.page_hashes.iter().cloned().collect();

        // Small mutation -> snapshot 2 shares almost every page with snapshot 1.
        insert_rows(src.path(), 1_000_000, 1).await;
        snapshot_dedup(src.path(), &target).await.unwrap();
        let (_, m2) = load_latest_manifest(&target).await.unwrap().unwrap();
        let s2_hashes: HashSet<String> = m2.page_hashes.iter().cloned().collect();
        let shared: HashSet<&String> = s1_hashes.intersection(&s2_hashes).collect();
        assert!(
            shared.len() > s1_hashes.len() / 2,
            "incremental snapshot should share most pages with the prior \
             (shared {}/{})",
            shared.len(),
            s1_hashes.len()
        );

        // GC keep=1: snapshot 1 is pruned, but every page snapshot 2 still
        // references must survive — including the ones it shares with snapshot 1.
        gc_dedup(&target, 1).await.unwrap();
        for h in &s2_hashes {
            assert!(
                target.store.get(&page_key(&target, h)).await.is_ok(),
                "shared page {h} (in retained snapshot 2) was deleted"
            );
        }
        // Pages exclusive to snapshot 1 are gone.
        for h in s1_hashes.difference(&s2_hashes) {
            assert!(
                target.store.get(&page_key(&target, h)).await.is_err(),
                "page {h} exclusive to pruned snapshot 1 still exists"
            );
        }
    }

    // ---- R006-T2: GC preserves byte-for-byte restorability of every
    // retained snapshot, not just the latest. Three snapshots, keep_n=2 ----
    //
    // The headline contract: an in-window *older* snapshot (the middle one
    // here) must restore byte-for-byte after GC, proving that GC's mark set
    // covers every retained manifest — not just the lexically-newest one
    // restore_latest_dedup uses. Ground truth is the pre-GC restore of the
    // same key; post-GC bytes are asserted equal.
    #[tokio::test]
    async fn gc_keep_two_lets_in_window_older_snapshot_restore_byte_for_byte() {
        let src = TempDb::new("gc-window");
        seed_db(src.path(), 5000).await;
        let target = BackupTarget { store: Arc::new(InMemory::new()), prefix: "backups".into() };

        // Three snapshots with a mutation between each.
        let key1 = match snapshot_dedup(src.path(), &target).await.unwrap() {
            DedupOutcome::Snapshotted { manifest_key, .. } => manifest_key,
            other => panic!("expected Snapshotted, got {other:?}"),
        };
        insert_rows(src.path(), 1_000_000, 1).await;
        let key2 = match snapshot_dedup(src.path(), &target).await.unwrap() {
            DedupOutcome::Snapshotted { manifest_key, .. } => manifest_key,
            other => panic!("expected Snapshotted, got {other:?}"),
        };
        insert_rows(src.path(), 2_000_000, 1).await;
        let key3 = match snapshot_dedup(src.path(), &target).await.unwrap() {
            DedupOutcome::Snapshotted { manifest_key, .. } => manifest_key,
            other => panic!("expected Snapshotted, got {other:?}"),
        };

        // Ground truth: restore the middle snapshot BEFORE GC.
        let pre = TempDb::new("gc-window-pre");
        restore_manifest_key(&target, &key2, pre.path()).await.unwrap();
        let pre_bytes = std::fs::read(pre.path()).unwrap();
        assert_eq!(count_rows(pre.path()).await, 5001);

        // GC keep_n=2 -> snapshot 1 pruned, 2 and 3 retained.
        let outcome = gc_dedup(&target, 2).await.unwrap();
        assert_eq!(outcome.pruned_manifests, 1);
        assert_eq!(outcome.retained_manifests, 2);
        assert!(
            outcome.deleted_pages > 0,
            "snapshot 1 had at least one page neither snapshot 2 nor 3 references"
        );

        // Snapshot 1 is gone — its manifest object no longer exists.
        let gone = TempDb::new("gc-window-gone");
        assert!(
            restore_manifest_key(&target, &key1, gone.path()).await.is_err(),
            "pruned snapshot 1 should not be restorable"
        );

        // Snapshot 2 (in-window older) restores BYTE-FOR-BYTE to the pre-GC bytes.
        let post = TempDb::new("gc-window-post");
        restore_manifest_key(&target, &key2, post.path()).await.unwrap();
        let post_bytes = std::fs::read(post.path()).unwrap();
        assert_eq!(
            pre_bytes, post_bytes,
            "in-window older snapshot must restore byte-for-byte after GC"
        );

        // And the newest (snapshot 3) restores to its own row count via the
        // latest path — proving keep_n=2 left both retained snapshots whole.
        let latest = TempDb::new("gc-window-latest");
        let restored_from = restore_latest_dedup(&target, latest.path()).await.unwrap();
        assert_eq!(restored_from, key3);
        assert_eq!(count_rows(latest.path()).await, 5002);
    }
}