hayson-engine 0.16.0

use std::io;
use std::path::{Path, PathBuf};

use crate::catalog::ExpandedIndex;
use crate::errors::HaysonIndexMismatchError;
use crate::registry::table_registry::{PrimaryEntry, TableRegistry};
use crate::schema::{TableSchema, UNPARTITIONED};
use crate::storage::{LogFile, LogSet};

use super::snapshot::{log_fingerprint, log_prefix_fingerprint, prefix_window, read_snapshot, Snapshot};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RestoreOutcome {
    Restored,
    Rebuilt,
}

/// What a restore attempt produced: the registry to open the table on, whether it came from the
/// snapshot or from a full scan, and the log generation to carry on counting from.
pub struct Restored {
    pub registry: TableRegistry,
    pub outcome: RestoreOutcome,
    /// The `gen` the restored snapshot carried, and `0` on a rebuild — a snapshot that could not be
    /// used says nothing about how many times its log was compacted. Same rule as the TS reference,
    /// where the generation is only carried over from a snapshot that paired.
    pub gen: u64,
}

enum PairingState {
    Stale,
    Ok,
    HashMismatch {
        log_path: PathBuf,
        watermark: u64,
        expected: String,
        actual: String,
    },
}

/// Checks one log against the watermark and fingerprint a snapshot claims for it.
///
/// A watermark equal to the log size is the only shape that can restore. A watermark SHORT of it is
/// still checked, not waved through: the prefix up to the watermark is fingerprinted, and only if it
/// still matches is the excess treated as a plain append — a tail this module does not own
/// (`table/` scans it), so the snapshot counts as stale and the caller rebuilds. When the prefix
/// disagrees the extra bytes are not an append at all: the log was replaced under a snapshot that
/// outlived it, which is the same tampering/mixed-generation case as an equal-length mismatch and is
/// reported the same way (ADR-r2pd/0019). Compaction always deletes the snapshot before swapping a
/// log, so a live rewrite never reaches here legitimately.
///
/// A watermark PAST the log size is checked against `prefix` instead — see [`short_log_pairing`].
fn pairing_at(log: &LogFile, watermark: u64, expected: String, prefix: Option<&str>) -> PairingState {
    if watermark > log.size() {
        return short_log_pairing(log, watermark, prefix);
    }
    match log_fingerprint(log, watermark) {
        Ok(actual) if actual == expected => {
            if watermark == log.size() {
                PairingState::Ok
            } else {
                PairingState::Stale
            }
        }
        Ok(actual) => PairingState::HashMismatch {
            log_path: log.path().to_path_buf(),
            watermark,
            expected,
            actual,
        },
        Err(_) => PairingState::Stale,
    }
}

/// Checks a log that is SHORTER than the watermark a snapshot claims for it.
///
/// Losing bytes off the end is a legal recovery shape, not on its own a disagreement: the log may
/// have vanished and been recreated empty (a writable `open()` does that), or `sync: "never"` may
/// have left the tail unflushed when the snapshot — always `fsync`ed — survived the crash. What
/// tells those apart from a snapshot laid over an unrelated file is the leading window, which
/// `logHash` cannot serve because it takes the watermark into its preimage and so cannot be
/// recomputed at another length. `prefixHash` can (HAY-197): the same byte range hashes
/// the same whatever the log's length became, so a disagreement there is proof of a different file
/// and is reported like any other mismatch.
///
/// Two cases stay stale for want of evidence, not by preference: a snapshot written before the
/// engine carried `prefixHash` (backward compatibility — it is `None` and there is nothing to
/// compare), and a log now shorter than the window itself, where the two sides would be hashing
/// different byte ranges. That leaves the check decisive exactly for a watermark past 4 KiB against
/// a log of at least 4 KiB.
fn short_log_pairing(log: &LogFile, watermark: u64, prefix: Option<&str>) -> PairingState {
    let Some(expected) = prefix else {
        return PairingState::Stale;
    };
    let window = prefix_window(watermark);
    if log.size() < window {
        return PairingState::Stale;
    }
    match log_prefix_fingerprint(log, window) {
        Ok(actual) if actual == expected => PairingState::Stale,
        Ok(actual) => PairingState::HashMismatch {
            log_path: log.path().to_path_buf(),
            watermark,
            expected: expected.to_string(),
            actual,
        },
        Err(_) => PairingState::Stale,
    }
}

fn pairing_of(snapshot: &Snapshot, log: &LogFile, schema_hash: &str) -> PairingState {
    let Some(expected) = snapshot.header.log_hash.clone() else {
        return PairingState::Stale;
    };
    if snapshot.header.schema_hash != schema_hash {
        return PairingState::Stale;
    }
    pairing_at(
        log,
        snapshot.header.watermark,
        expected,
        snapshot.header.prefix_hash.as_deref(),
    )
}

/// The partitioned counterpart of [`pairing_of`]: the check runs per partition and the snapshot is
/// taken as a whole or not at all, so a partition that appeared, grew or vanished since the
/// snapshot was written costs a full rebuild rather than a partial restore (`docs/format.md`).
///
/// A `parts` entry carries no prefix fingerprint, so a partition shorter than its watermark stays
/// stale the way every log did before [`short_log_pairing`] existed (HAY-197).
fn pairing_of_set(snapshot: &Snapshot, logs: &LogSet, schema_hash: &str) -> PairingState {
    if !logs.is_partitioned() {
        return match logs.log_of(UNPARTITIONED) {
            Some(log) => pairing_of(snapshot, log, schema_hash),
            None => PairingState::Stale,
        };
    }
    let Some(claimed) = snapshot.header.parts.as_ref() else {
        return PairingState::Stale;
    };
    if snapshot.header.schema_hash != schema_hash || claimed.len() != logs.parts().len() {
        return PairingState::Stale;
    }
    let mut stale = false;
    for part in claimed {
        let Some(log) = logs.log_of(&part.part) else {
            return PairingState::Stale;
        };
        match pairing_at(log, part.watermark, part.log_hash.clone(), None) {
            PairingState::Ok => {}
            // NOTE: keep scanning -- a later partition may hold a mismatch, which outranks staleness.
            PairingState::Stale => stale = true,
            mismatch => return mismatch,
        }
    }
    if stale {
        return PairingState::Stale;
    }
    PairingState::Ok
}

fn restore_from(schema: TableSchema, snapshot: &Snapshot, declared: &[ExpandedIndex]) -> TableRegistry {
    let mut registry = TableRegistry::with_declared_indexes(schema, declared);
    for entry in &snapshot.entries {
        registry.hydrate(
            &entry.key,
            &entry.projection,
            PrimaryEntry {
                offset: entry.offset,
                length: entry.length,
                seq: entry.seq,
            },
            entry.part.as_deref().unwrap_or(""),
        );
    }
    match &snapshot.header.parts {
        Some(parts) => {
            for part in parts {
                registry.set_dead_bytes(part.dead_bytes, &part.part);
            }
        }
        None => registry.set_dead_bytes(snapshot.header.dead_bytes, ""),
    }
    registry.note_seq(snapshot.header.seq);
    registry
}

/// Loads a snapshot and restores a registry from it when the snapshot's schema hash, watermark and
/// content fingerprint all still match `log` — a cache hit, `RestoreOutcome::Restored`. An ORDINARY
/// stale case (no snapshot, corrupt file, stale schema, a log shorter than the watermark whose
/// leading window still agrees, or a log that grew by a genuine append on top of a still-matching
/// prefix) falls back to
/// `TableRegistry::build_from_log`, `RestoreOutcome::Rebuilt` — the snapshot is an optimization only
/// there, so nothing fails because of it. A snapshot whose schema matches but whose content
/// fingerprint does NOT match the log up to its watermark is a different, louder case
/// (ADR-r2pd/0019): that combination means the log or the snapshot was tampered with or corrupted in
/// a way that watermark/schema checks can't catch, so it is left on disk and reported as
/// `HaysonIndexMismatchError` instead of silently rebuilding underneath a caller who might otherwise
/// never notice. The only other propagated error is a real IO failure scanning `log` itself.
#[allow(dead_code)]
pub fn restore_or_rebuild(
    path: &Path,
    schema: TableSchema,
    log: &LogFile,
    schema_hash: &str,
) -> io::Result<(TableRegistry, RestoreOutcome)> {
    restore_or_rebuild_with_declared(path, schema, log, schema_hash, &[])
}

/// Same as [`restore_or_rebuild`], but both the restored and the rebuilt registry are built with
/// `declared` — the catalog's fully expanded index list (`catalog::expand_indexes`) — so a
/// composite or `where`-scoped index is enforced/served the same way whichever path is taken.
#[allow(dead_code)]
pub fn restore_or_rebuild_with_declared(
    path: &Path,
    schema: TableSchema,
    log: &LogFile,
    schema_hash: &str,
    declared: &[ExpandedIndex],
) -> io::Result<(TableRegistry, RestoreOutcome)> {
    if let Some(snapshot) = read_snapshot(path).unwrap_or(None) {
        match pairing_of(&snapshot, log, schema_hash) {
            PairingState::Ok => {
                return Ok((restore_from(schema, &snapshot, declared), RestoreOutcome::Restored));
            }
            PairingState::HashMismatch {
                log_path,
                watermark,
                expected,
                actual,
            } => {
                return Err(io::Error::other(HaysonIndexMismatchError {
                    path: path.display().to_string(),
                    log_path: log_path.display().to_string(),
                    watermark,
                    expected_hash: expected,
                    actual_hash: actual,
                }));
            }
            PairingState::Stale => {}
        }
    }
    let registry = TableRegistry::build_from_log_with_declared(schema, log, declared)?;
    Ok((registry, RestoreOutcome::Rebuilt))
}

/// Same as [`restore_or_rebuild_with_declared`], for a table held as a [`LogSet`] — one log file
/// when it declares no `partitionBy`, one per partition value when it does. Identical rules, with
/// the pairing check running per partition ([`pairing_of_set`]); the rebuild falls back to scanning
/// every partition.
///
/// The set is taken by `&mut` for the rebuild path: see
/// [`TableRegistry::build_from_log_set_with_declared`]. The restore path never touches the files —
/// a snapshot only pairs when its watermark is the whole log, which a half-written tail is not.
pub fn restore_or_rebuild_set_with_declared(
    path: &Path,
    schema: TableSchema,
    logs: &mut LogSet,
    schema_hash: &str,
    declared: &[ExpandedIndex],
) -> io::Result<Restored> {
    if let Some(snapshot) = read_snapshot(path).unwrap_or(None) {
        match pairing_of_set(&snapshot, logs, schema_hash) {
            PairingState::Ok => {
                return Ok(Restored {
                    registry: restore_from(schema, &snapshot, declared),
                    outcome: RestoreOutcome::Restored,
                    gen: snapshot.header.gen,
                });
            }
            PairingState::HashMismatch {
                log_path,
                watermark,
                expected,
                actual,
            } => {
                return Err(io::Error::other(HaysonIndexMismatchError {
                    path: path.display().to_string(),
                    log_path: log_path.display().to_string(),
                    watermark,
                    expected_hash: expected,
                    actual_hash: actual,
                }));
            }
            PairingState::Stale => {}
        }
    }
    let registry = TableRegistry::build_from_log_set_with_declared(schema, logs, declared)?;
    Ok(Restored {
        registry,
        outcome: RestoreOutcome::Rebuilt,
        gen: 0,
    })
}

#[cfg(test)]
mod tests {
    use std::fs;
    use std::os::unix::fs::symlink;

    use serde_json::json;

    use super::*;
    use crate::checkpoint::{write_snapshot, write_snapshot_set};
    use crate::registry::fulltext_index::SearchMode;
    use crate::registry::sorted_index::{RangeBounds, SortValue};
    use crate::schema::parse_schema;
    use crate::storage::record::{serialize_entry, LogMeta, LogOp};

    const SCHEMA_HASH: &str = "schema-hash-1";

    fn users_schema() -> TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "name": { "type": "string", "fulltext": true },
                "role": { "type": "string", "index": "exact" },
                "age": { "type": "int", "index": "sorted" },
            },
        }))
        .unwrap()
    }

    fn notes_schema() -> TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "notes",
            "primaryKey": "id",
            "partitionBy": "kind",
            "fields": {
                "id": { "type": "string" },
                "kind": { "type": "string", "required": true },
            },
        }))
        .unwrap()
    }

    fn note_line(id: &str, seq: u64, kind: &str) -> String {
        let mut data = crate::schema::HaysonRecord::new();
        data.insert("id".into(), json!(id));
        data.insert("kind".into(), json!(kind));
        serialize_entry(
            &LogMeta {
                key: id.into(),
                op: LogOp::Put,
                seq,
                ts: 1,
            },
            Some(&data),
        )
    }

    fn put_line(id: &str, seq: u64, name: &str, role: &str, age: i64) -> String {
        let mut data = crate::schema::HaysonRecord::new();
        data.insert("id".into(), json!(id));
        data.insert("name".into(), json!(name));
        data.insert("role".into(), json!(role));
        data.insert("age".into(), json!(age));
        serialize_entry(
            &LogMeta {
                key: id.into(),
                op: LogOp::Put,
                seq,
                ts: 1,
            },
            Some(&data),
        )
    }

    fn queries(registry: &mut TableRegistry) -> (Vec<String>, Vec<String>, Vec<String>) {
        let mut by_role: Vec<String> = registry
            .lookup_exact("role", &json!("admin"))
            .cloned()
            .unwrap_or_default()
            .into_iter()
            .collect();
        by_role.sort();

        let by_age = registry.sorted_index("age").map(|index| {
            index.range(&RangeBounds {
                gte: Some(vec![SortValue::Num(30.0)]),
                ..Default::default()
            })
        });
        let mut by_age = by_age.unwrap_or_default();
        by_age.sort();

        let mut fulltext: Vec<String> = registry.search("kowalska", SearchMode::All).into_keys().collect();
        fulltext.sort();

        (by_role, by_age, fulltext)
    }

    fn seeded_log(dir: &std::path::Path) -> LogFile {
        let path = dir.join("users.jsonl");
        let mut log = LogFile::open(&path, false).unwrap();
        log.append(&[
            put_line("user_1", 1, "Ania Kowalska", "admin", 31),
            put_line("user_2", 2, "Piotr Nowak", "user", 24),
        ])
        .unwrap();
        log
    }

    #[test]
    fn restores_a_registry_that_answers_queries_identically_to_the_one_it_was_snapshotted_from() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("users.idx");
        let log = seeded_log(dir.path());

        let mut original = TableRegistry::build_from_log(users_schema(), &log).unwrap();
        let before = queries(&mut original);
        write_snapshot(&idx_path, &original, "users", SCHEMA_HASH, 0, &log).unwrap();

        let (mut restored, outcome) = restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH).unwrap();
        assert_eq!(outcome, RestoreOutcome::Restored);
        assert_eq!(queries(&mut restored), before);
        assert_eq!(restored.records(), original.records());
        assert_eq!(restored.max_seq(), original.max_seq());
    }

    #[test]
    fn falls_back_to_a_full_scan_when_the_log_grew_after_the_snapshot_was_taken() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("users.idx");
        let mut log = seeded_log(dir.path());

        let registry_at_snapshot = TableRegistry::build_from_log(users_schema(), &log).unwrap();
        write_snapshot(&idx_path, &registry_at_snapshot, "users", SCHEMA_HASH, 0, &log).unwrap();

        log.append(&[put_line("user_3", 3, "Zosia Wojcik", "admin", 40)])
            .unwrap();

        let (mut restored, outcome) = restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH).unwrap();
        assert_eq!(outcome, RestoreOutcome::Rebuilt);

        let mut full = TableRegistry::build_from_log(users_schema(), &log).unwrap();
        assert_eq!(queries(&mut restored), queries(&mut full));
        assert_eq!(restored.records(), 3);
    }

    #[test]
    fn rejects_a_snapshot_of_a_log_generation_that_was_rewritten_and_then_grown_past_it() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("users.idx");
        let mut log = seeded_log(dir.path());

        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();
        let stranded = fs::read(&idx_path).unwrap();

        log.truncate(0).unwrap();
        let mut filler = 1;
        while log.size() <= stranded.len() as u64 {
            log.append(&[put_line(
                &format!("filler_{filler}"),
                filler,
                &format!("Filler {filler}"),
                "user",
                30,
            )])
            .unwrap();
            filler += 1;
        }
        assert!(
            log.size() > stranded.len() as u64,
            "the rewritten log outgrew the snapshot it will be paired against"
        );
        fs::write(&idx_path, &stranded).unwrap();

        let err = match restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH) {
            Err(err) => err,
            Ok((_, outcome)) => {
                panic!("expected a snapshot of a replaced log generation to be rejected, got {outcome:?}")
            }
        };
        assert!(err
            .get_ref()
            .unwrap()
            .downcast_ref::<crate::errors::HaysonIndexMismatchError>()
            .is_some());
        assert_eq!(fs::read(&idx_path).unwrap(), stranded);
    }

    #[test]
    fn rejects_a_partitioned_snapshot_of_a_partition_that_was_rewritten_and_then_grown_past_it() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("notes.idx");
        let mut logs = LogSet::partitioned(dir.path().join("notes"), false).unwrap();
        logs.append("adr", &[note_line("a1", 1, "adr")]).unwrap();
        logs.append("note", &[note_line("n1", 2, "note")]).unwrap();

        let registry = TableRegistry::build_from_log_set_with_declared(notes_schema(), &mut logs, &[]).unwrap();
        write_snapshot_set(&idx_path, &registry, "notes", SCHEMA_HASH, 0, &logs).unwrap();
        let stranded = fs::read(&idx_path).unwrap();

        let replaced = logs.log_of_mut("note").unwrap();
        replaced.truncate(0).unwrap();
        replaced
            .append(&[note_line("n2", 3, "note"), note_line("n3", 4, "note")])
            .unwrap();

        let err = match restore_or_rebuild_set_with_declared(&idx_path, notes_schema(), &mut logs, SCHEMA_HASH, &[]) {
            Err(err) => err,
            Ok(_) => panic!("expected a snapshot of a replaced partition generation to be rejected"),
        };
        let mismatch = err
            .get_ref()
            .unwrap()
            .downcast_ref::<crate::errors::HaysonIndexMismatchError>()
            .expect("a replaced partition generation is rejected as an index mismatch");
        assert_eq!(mismatch.log_path, logs.path_of("note").display().to_string());
        assert_eq!(fs::read(&idx_path).unwrap(), stranded);
    }

    /// A log comfortably past `PAIRING_WINDOW` (4 KiB), so the leading-window check has a full
    /// window on both sides — below that the short-log comparison declines to judge.
    fn fat_log(path: &std::path::Path, tag: &str, at_least: u64) -> LogFile {
        let mut log = LogFile::open(path, false).unwrap();
        let mut seq = 0;
        while log.size() < at_least {
            seq += 1;
            log.append(&[put_line(
                &format!("{tag}_{seq}"),
                seq,
                &format!("{tag} Kowalska {seq}"),
                "admin",
                30 + seq as i64,
            )])
            .unwrap();
        }
        log
    }

    #[test]
    fn keeps_rebuilding_quietly_when_the_log_vanished_and_reopened_empty_under_its_snapshot() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("users.idx");
        let log_path = dir.path().join("users.jsonl");
        let log = fat_log(&log_path, "before", 8192);

        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();

        fs::remove_file(&log_path).unwrap();
        let recreated = LogFile::open(&log_path, false).unwrap();
        assert_eq!(recreated.size(), 0, "a writable open recreates the log empty");

        let (rebuilt, outcome) = restore_or_rebuild(&idx_path, users_schema(), &recreated, SCHEMA_HASH).unwrap();
        assert_eq!(outcome, RestoreOutcome::Rebuilt);
        assert_eq!(rebuilt.records(), 0);
    }

    #[test]
    fn keeps_rebuilding_quietly_when_an_unflushed_tail_was_lost_but_the_leading_window_agrees() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("users.idx");
        let mut log = fat_log(&dir.path().join("users.jsonl"), "kept", 8192);

        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();
        let watermark = log.size();

        log.truncate(6144).unwrap();
        assert!(
            log.size() < watermark && log.size() > 4096,
            "a lost tail over an intact window"
        );

        let (_, outcome) = restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH).unwrap();
        assert_eq!(outcome, RestoreOutcome::Rebuilt);
    }

    #[test]
    fn rejects_a_snapshot_laid_over_a_shorter_log_it_was_never_taken_of() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("users.idx");
        let stranger = fat_log(&dir.path().join("stranger.jsonl"), "stranger", 8192);

        let theirs = fat_log(&dir.path().join("users.jsonl"), "theirs", 12288);
        let registry = TableRegistry::build_from_log(users_schema(), &theirs).unwrap();
        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &theirs).unwrap();
        assert!(
            stranger.size() < theirs.size(),
            "the planted snapshot outruns the log it meets"
        );

        let err = match restore_or_rebuild(&idx_path, users_schema(), &stranger, SCHEMA_HASH) {
            Err(err) => err,
            Ok((_, outcome)) => {
                panic!("expected a snapshot of an unrelated log to be rejected, got {outcome:?}")
            }
        };
        let mismatch = err
            .get_ref()
            .unwrap()
            .downcast_ref::<crate::errors::HaysonIndexMismatchError>()
            .expect("an unrelated log is rejected as an index mismatch");
        assert_eq!(mismatch.log_path, stranger.path().display().to_string());
        assert_eq!(mismatch.watermark, theirs.size());
        assert!(idx_path.exists(), "a refused snapshot is left on disk to be looked at");
    }

    #[test]
    fn a_snapshot_written_without_a_prefix_fingerprint_still_rebuilds_quietly_under_a_short_log() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("users.idx");
        let stranger = fat_log(&dir.path().join("stranger.jsonl"), "stranger", 8192);

        let theirs = fat_log(&dir.path().join("users.jsonl"), "theirs", 12288);
        let registry = TableRegistry::build_from_log(users_schema(), &theirs).unwrap();
        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &theirs).unwrap();

        let raw = fs::read_to_string(&idx_path).unwrap();
        let mut lines: Vec<&str> = raw.lines().collect();
        let aged_header = lines[0].replace(
            &format!(
                ",\"prefixHash\":\"{}\"",
                log_prefix_fingerprint(&theirs, prefix_window(theirs.size())).unwrap()
            ),
            "",
        );
        assert!(
            !aged_header.contains("prefixHash"),
            "the header now looks like a pre-HAY-197 one"
        );
        lines[0] = &aged_header;
        fs::write(&idx_path, format!("{}\n", lines.join("\n"))).unwrap();

        let (_, outcome) = restore_or_rebuild(&idx_path, users_schema(), &stranger, SCHEMA_HASH).unwrap();
        assert_eq!(outcome, RestoreOutcome::Rebuilt);
    }

    /// A `.idx` written before the `"snapshotVersion"` rename is not read back — and must not be
    /// an error either: the snapshot is a cache, so the unrecognized header rebuilds like any
    /// stale one, which is why this rename needs no old-key tolerance.
    #[test]
    fn rebuilds_from_the_log_when_the_snapshot_carries_the_pre_rename_marker() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("users.idx");
        let log = seeded_log(dir.path());
        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();

        let raw = fs::read_to_string(&idx_path).unwrap();
        let aged = raw.replacen("\"snapshotVersion\"", "\"hayson\"", 1);
        assert_ne!(aged, raw, "the fixture has to be the header this rename touched");
        fs::write(&idx_path, &aged).unwrap();

        let (mut restored, outcome) = restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH).unwrap();
        assert_eq!(outcome, RestoreOutcome::Rebuilt);

        let mut full = TableRegistry::build_from_log(users_schema(), &log).unwrap();
        assert_eq!(queries(&mut restored), queries(&mut full));
    }

    #[test]
    fn falls_back_without_erroring_when_the_snapshot_file_is_truncated_garbage() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("users.idx");
        let log = seeded_log(dir.path());

        fs::write(&idx_path, b"{\"snapshotVersion\":1,\"kind\":\"index\"").unwrap();

        let (mut restored, outcome) = restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH).unwrap();
        assert_eq!(outcome, RestoreOutcome::Rebuilt);

        let mut full = TableRegistry::build_from_log(users_schema(), &log).unwrap();
        assert_eq!(queries(&mut restored), queries(&mut full));
    }

    #[test]
    fn falls_back_without_erroring_when_no_snapshot_file_exists() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("users.idx");
        let log = seeded_log(dir.path());

        let (registry, outcome) = restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH).unwrap();
        assert_eq!(outcome, RestoreOutcome::Rebuilt);
        assert_eq!(registry.records(), 2);
    }

    #[test]
    fn rejects_a_snapshot_whose_schema_hash_no_longer_matches() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("users.idx");
        let log = seeded_log(dir.path());

        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();

        let (_, outcome) = restore_or_rebuild(&idx_path, users_schema(), &log, "schema-hash-2").unwrap();
        assert_eq!(outcome, RestoreOutcome::Rebuilt);
    }

    #[test]
    fn rejects_a_paired_snapshot_whose_content_fingerprint_was_tampered_with() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("users.idx");
        let log = seeded_log(dir.path());

        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();

        let raw = fs::read_to_string(&idx_path).unwrap();
        let mut lines: Vec<&str> = raw.lines().collect();
        let tampered_header = lines[0].replace(
            &format!("\"logHash\":\"{}\"", log_fingerprint(&log, log.size()).unwrap()),
            "\"logHash\":\"0000000000000000\"",
        );
        lines[0] = &tampered_header;
        fs::write(&idx_path, format!("{}\n", lines.join("\n"))).unwrap();

        let result = restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH);
        let err = match result {
            Err(err) => err,
            Ok(_) => panic!("expected a tampered, paired snapshot to be rejected"),
        };
        assert!(err
            .get_ref()
            .unwrap()
            .downcast_ref::<crate::errors::HaysonIndexMismatchError>()
            .is_some());
        assert!(idx_path.exists());
    }

    #[test]
    fn a_rejected_snapshot_names_the_index_the_log_and_the_watermark_it_claimed() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("users.idx");
        let log = seeded_log(dir.path());

        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();

        let raw = fs::read_to_string(&idx_path).unwrap();
        let mut lines: Vec<&str> = raw.lines().collect();
        let tampered_header = lines[0].replace(
            &format!("\"logHash\":\"{}\"", log_fingerprint(&log, log.size()).unwrap()),
            "\"logHash\":\"0000000000000000\"",
        );
        lines[0] = &tampered_header;
        fs::write(&idx_path, format!("{}\n", lines.join("\n"))).unwrap();

        let err = match restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH) {
            Err(err) => err,
            Ok(_) => panic!("expected a tampered, paired snapshot to be rejected"),
        };
        let mismatch = err
            .get_ref()
            .unwrap()
            .downcast_ref::<crate::errors::HaysonIndexMismatchError>()
            .expect("a tampered, paired snapshot is rejected as an index mismatch");

        assert_eq!(mismatch.path, idx_path.display().to_string());
        assert_eq!(mismatch.log_path, log.path().display().to_string());
        assert_eq!(mismatch.watermark, log.size());
        assert_eq!(mismatch.expected_hash, "0000000000000000");
        assert_eq!(mismatch.actual_hash, log_fingerprint(&log, log.size()).unwrap());
    }

    #[test]
    fn a_rejected_partitioned_snapshot_names_the_partition_whose_fingerprint_disagreed() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("notes.idx");
        let mut logs = LogSet::partitioned(dir.path().join("notes"), false).unwrap();
        logs.append("adr", &[note_line("a1", 1, "adr")]).unwrap();
        logs.append("note", &[note_line("n1", 2, "note")]).unwrap();

        let registry = TableRegistry::build_from_log_set_with_declared(notes_schema(), &mut logs, &[]).unwrap();
        write_snapshot_set(&idx_path, &registry, "notes", SCHEMA_HASH, 0, &logs).unwrap();

        let tampered = logs.log_of("note").unwrap();
        let raw = fs::read_to_string(&idx_path).unwrap();
        let mut lines: Vec<&str> = raw.lines().collect();
        let tampered_header = lines[0].replace(
            &format!("\"h\":\"{}\"", log_fingerprint(tampered, tampered.size()).unwrap()),
            "\"h\":\"0000000000000000\"",
        );
        lines[0] = &tampered_header;
        fs::write(&idx_path, format!("{}\n", lines.join("\n"))).unwrap();

        let err = match restore_or_rebuild_set_with_declared(&idx_path, notes_schema(), &mut logs, SCHEMA_HASH, &[]) {
            Err(err) => err,
            Ok(_) => panic!("expected a tampered, paired partition to be rejected"),
        };
        let mismatch = err
            .get_ref()
            .unwrap()
            .downcast_ref::<crate::errors::HaysonIndexMismatchError>()
            .expect("a tampered, paired partition is rejected as an index mismatch");

        assert_eq!(mismatch.path, idx_path.display().to_string());
        assert_eq!(mismatch.log_path, logs.path_of("note").display().to_string());
        assert_eq!(mismatch.watermark, logs.log_of("note").unwrap().size());
    }

    #[test]
    fn write_snapshot_is_atomic_and_does_not_write_through_a_symlink_at_the_target_path() {
        let dir = tempfile::tempdir().unwrap();
        let victim_dir = tempfile::tempdir().unwrap();
        let victim = victim_dir.path().join("precious.idx");
        fs::write(&victim, "PRECIOUS").unwrap();

        let idx_path = dir.path().join("users.idx");
        symlink(&victim, &idx_path).unwrap();

        let log = seeded_log(dir.path());
        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();

        assert_eq!(fs::read_to_string(&victim).unwrap(), "PRECIOUS");
        assert!(!fs::symlink_metadata(&idx_path).unwrap().file_type().is_symlink());
        assert!(fs::read_to_string(&idx_path).unwrap().contains("\"table\":\"users\""));
    }

    #[test]
    fn write_snapshot_leaves_no_temporary_file_behind() {
        let dir = tempfile::tempdir().unwrap();
        let idx_path = dir.path().join("users.idx");
        let log = seeded_log(dir.path());
        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();

        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();

        assert!(!dir.path().join("users.idx.tmp").exists());
    }
}