Skip to main content

hayson_engine/checkpoint/
restore.rs

1use std::io;
2use std::path::{Path, PathBuf};
3
4use crate::catalog::ExpandedIndex;
5use crate::errors::HaysonIndexMismatchError;
6use crate::registry::table_registry::{PrimaryEntry, TableRegistry};
7use crate::schema::{TableSchema, UNPARTITIONED};
8use crate::storage::{LogFile, LogSet};
9
10use super::snapshot::{
11    log_fingerprint, log_prefix_fingerprint, prefix_window, read_snapshot, read_snapshot_sealed, Snapshot,
12};
13use crate::crypto::Sealer;
14
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16pub enum RestoreOutcome {
17    Restored,
18    Rebuilt,
19}
20
21/// What a restore attempt produced: the registry to open the table on, whether it came from the
22/// snapshot or from a full scan, and the log generation to carry on counting from.
23pub struct Restored {
24    pub registry: TableRegistry,
25    pub outcome: RestoreOutcome,
26    /// The `gen` the restored snapshot carried, and `0` on a rebuild — a snapshot that could not be
27    /// used says nothing about how many times its log was compacted. Same rule as the TS reference,
28    /// where the generation is only carried over from a snapshot that paired.
29    pub gen: u64,
30}
31
32enum PairingState {
33    Stale,
34    Ok,
35    HashMismatch {
36        log_path: PathBuf,
37        watermark: u64,
38        expected: String,
39        actual: String,
40    },
41}
42
43/// Checks one log against the watermark and fingerprint a snapshot claims for it.
44///
45/// A watermark equal to the log size is the only shape that can restore. A watermark SHORT of it is
46/// still checked, not waved through: the prefix up to the watermark is fingerprinted, and only if it
47/// still matches is the excess treated as a plain append — a tail this module does not own
48/// (`table/` scans it), so the snapshot counts as stale and the caller rebuilds. When the prefix
49/// disagrees the extra bytes are not an append at all: the log was replaced under a snapshot that
50/// outlived it, which is the same tampering/mixed-generation case as an equal-length mismatch and is
51/// reported the same way (ADR-r2pd/0019). Compaction always deletes the snapshot before swapping a
52/// log, so a live rewrite never reaches here legitimately.
53///
54/// A watermark PAST the log size is checked against `prefix` instead — see [`short_log_pairing`].
55fn pairing_at(log: &LogFile, watermark: u64, expected: String, prefix: Option<&str>) -> PairingState {
56    if watermark > log.size() {
57        return short_log_pairing(log, watermark, prefix);
58    }
59    match log_fingerprint(log, watermark) {
60        Ok(actual) if actual == expected => {
61            if watermark == log.size() {
62                PairingState::Ok
63            } else {
64                PairingState::Stale
65            }
66        }
67        Ok(actual) => PairingState::HashMismatch {
68            log_path: log.path().to_path_buf(),
69            watermark,
70            expected,
71            actual,
72        },
73        Err(_) => PairingState::Stale,
74    }
75}
76
77/// Checks a log that is SHORTER than the watermark a snapshot claims for it.
78///
79/// Losing bytes off the end is a legal recovery shape, not on its own a disagreement: the log may
80/// have vanished and been recreated empty (a writable `open()` does that), or `sync: "never"` may
81/// have left the tail unflushed when the snapshot — always `fsync`ed — survived the crash. What
82/// tells those apart from a snapshot laid over an unrelated file is the leading window, which
83/// `logHash` cannot serve because it takes the watermark into its preimage and so cannot be
84/// recomputed at another length. `prefixHash` can (HAY-197): the same byte range hashes
85/// the same whatever the log's length became, so a disagreement there is proof of a different file
86/// and is reported like any other mismatch.
87///
88/// Two cases stay stale for want of evidence, not by preference: a snapshot written before the
89/// engine carried `prefixHash` (backward compatibility — it is `None` and there is nothing to
90/// compare), and a log now shorter than the window itself, where the two sides would be hashing
91/// different byte ranges. That leaves the check decisive exactly for a watermark past 4 KiB against
92/// a log of at least 4 KiB.
93fn short_log_pairing(log: &LogFile, watermark: u64, prefix: Option<&str>) -> PairingState {
94    let Some(expected) = prefix else {
95        return PairingState::Stale;
96    };
97    let window = prefix_window(watermark);
98    if log.size() < window {
99        return PairingState::Stale;
100    }
101    match log_prefix_fingerprint(log, window) {
102        Ok(actual) if actual == expected => PairingState::Stale,
103        Ok(actual) => PairingState::HashMismatch {
104            log_path: log.path().to_path_buf(),
105            watermark,
106            expected: expected.to_string(),
107            actual,
108        },
109        Err(_) => PairingState::Stale,
110    }
111}
112
113fn pairing_of(snapshot: &Snapshot, log: &LogFile, schema_hash: &str) -> PairingState {
114    let Some(expected) = snapshot.header.log_hash.clone() else {
115        return PairingState::Stale;
116    };
117    if snapshot.header.schema_hash != schema_hash {
118        return PairingState::Stale;
119    }
120    pairing_at(
121        log,
122        snapshot.header.watermark,
123        expected,
124        snapshot.header.prefix_hash.as_deref(),
125    )
126}
127
128/// The partitioned counterpart of [`pairing_of`]: the check runs per partition and the snapshot is
129/// taken as a whole or not at all, so a partition that appeared, grew or vanished since the
130/// snapshot was written costs a full rebuild rather than a partial restore (`docs/format.md`).
131///
132/// A `parts` entry carries no prefix fingerprint, so a partition shorter than its watermark stays
133/// stale the way every log did before [`short_log_pairing`] existed (HAY-197).
134fn pairing_of_set(snapshot: &Snapshot, logs: &LogSet, schema_hash: &str) -> PairingState {
135    if !logs.is_partitioned() {
136        return match logs.log_of(UNPARTITIONED) {
137            Some(log) => pairing_of(snapshot, log, schema_hash),
138            None => PairingState::Stale,
139        };
140    }
141    let Some(claimed) = snapshot.header.parts.as_ref() else {
142        return PairingState::Stale;
143    };
144    if snapshot.header.schema_hash != schema_hash || claimed.len() != logs.parts().len() {
145        return PairingState::Stale;
146    }
147    let mut stale = false;
148    for part in claimed {
149        let Some(log) = logs.log_of(&part.part) else {
150            return PairingState::Stale;
151        };
152        match pairing_at(log, part.watermark, part.log_hash.clone(), None) {
153            PairingState::Ok => {}
154            // NOTE: keep scanning -- a later partition may hold a mismatch, which outranks staleness.
155            PairingState::Stale => stale = true,
156            mismatch => return mismatch,
157        }
158    }
159    if stale {
160        return PairingState::Stale;
161    }
162    PairingState::Ok
163}
164
165fn restore_from(schema: TableSchema, snapshot: &Snapshot, declared: &[ExpandedIndex]) -> TableRegistry {
166    let mut registry = TableRegistry::with_declared_indexes(schema, declared);
167    for entry in &snapshot.entries {
168        registry.hydrate(
169            &entry.key,
170            &entry.projection,
171            PrimaryEntry {
172                offset: entry.offset,
173                length: entry.length,
174                seq: entry.seq,
175            },
176            entry.part.as_deref().unwrap_or(""),
177        );
178    }
179    match &snapshot.header.parts {
180        Some(parts) => {
181            for part in parts {
182                registry.set_dead_bytes(part.dead_bytes, &part.part);
183            }
184        }
185        None => registry.set_dead_bytes(snapshot.header.dead_bytes, ""),
186    }
187    registry.note_seq(snapshot.header.seq);
188    registry
189}
190
191/// Loads a snapshot and restores a registry from it when the snapshot's schema hash, watermark and
192/// content fingerprint all still match `log` — a cache hit, `RestoreOutcome::Restored`. An ORDINARY
193/// stale case (no snapshot, corrupt file, stale schema, a log shorter than the watermark whose
194/// leading window still agrees, or a log that grew by a genuine append on top of a still-matching
195/// prefix) falls back to
196/// `TableRegistry::build_from_log`, `RestoreOutcome::Rebuilt` — the snapshot is an optimization only
197/// there, so nothing fails because of it. A snapshot whose schema matches but whose content
198/// fingerprint does NOT match the log up to its watermark is a different, louder case
199/// (ADR-r2pd/0019): that combination means the log or the snapshot was tampered with or corrupted in
200/// a way that watermark/schema checks can't catch, so it is left on disk and reported as
201/// `HaysonIndexMismatchError` instead of silently rebuilding underneath a caller who might otherwise
202/// never notice. The only other propagated error is a real IO failure scanning `log` itself.
203#[allow(dead_code)]
204pub fn restore_or_rebuild(
205    path: &Path,
206    schema: TableSchema,
207    log: &LogFile,
208    schema_hash: &str,
209) -> io::Result<(TableRegistry, RestoreOutcome)> {
210    restore_or_rebuild_with_declared(path, schema, log, schema_hash, &[])
211}
212
213/// Same as [`restore_or_rebuild`], but both the restored and the rebuilt registry are built with
214/// `declared` — the catalog's fully expanded index list (`catalog::expand_indexes`) — so a
215/// composite or `where`-scoped index is enforced/served the same way whichever path is taken.
216#[allow(dead_code)]
217pub fn restore_or_rebuild_with_declared(
218    path: &Path,
219    schema: TableSchema,
220    log: &LogFile,
221    schema_hash: &str,
222    declared: &[ExpandedIndex],
223) -> io::Result<(TableRegistry, RestoreOutcome)> {
224    if let Some(snapshot) = read_snapshot(path).unwrap_or(None) {
225        match pairing_of(&snapshot, log, schema_hash) {
226            PairingState::Ok => {
227                return Ok((restore_from(schema, &snapshot, declared), RestoreOutcome::Restored));
228            }
229            PairingState::HashMismatch {
230                log_path,
231                watermark,
232                expected,
233                actual,
234            } => {
235                return Err(io::Error::other(HaysonIndexMismatchError {
236                    path: path.display().to_string(),
237                    log_path: log_path.display().to_string(),
238                    watermark,
239                    expected_hash: expected,
240                    actual_hash: actual,
241                }));
242            }
243            PairingState::Stale => {}
244        }
245    }
246    let registry = TableRegistry::build_from_log_with_declared(schema, log, declared)?;
247    Ok((registry, RestoreOutcome::Rebuilt))
248}
249
250/// Same as [`restore_or_rebuild_with_declared`], for a table held as a [`LogSet`] — one log file
251/// when it declares no `partitionBy`, one per partition value when it does. Identical rules, with
252/// the pairing check running per partition ([`pairing_of_set`]); the rebuild falls back to scanning
253/// every partition.
254///
255/// The set is taken by `&mut` for the rebuild path: see
256/// [`TableRegistry::build_from_log_set_with_declared`]. The restore path never touches the files —
257/// a snapshot only pairs when its watermark is the whole log, which a half-written tail is not.
258pub fn restore_or_rebuild_set_with_declared(
259    path: &Path,
260    schema: TableSchema,
261    logs: &mut LogSet,
262    schema_hash: &str,
263    declared: &[ExpandedIndex],
264) -> io::Result<Restored> {
265    restore_or_rebuild_set_sealed(path, schema, logs, schema_hash, declared, None)
266}
267
268/// [`restore_or_rebuild_set_with_declared`] for sealed snapshots: see
269/// [`super::snapshot::read_snapshot_sealed`].
270pub fn restore_or_rebuild_set_sealed(
271    path: &Path,
272    schema: TableSchema,
273    logs: &mut LogSet,
274    schema_hash: &str,
275    declared: &[ExpandedIndex],
276    sealer: Option<&Sealer>,
277) -> io::Result<Restored> {
278    if let Some(snapshot) = read_snapshot_sealed(path, sealer, &schema.table).unwrap_or(None) {
279        match pairing_of_set(&snapshot, logs, schema_hash) {
280            PairingState::Ok => {
281                return Ok(Restored {
282                    registry: restore_from(schema, &snapshot, declared),
283                    outcome: RestoreOutcome::Restored,
284                    gen: snapshot.header.gen,
285                });
286            }
287            PairingState::HashMismatch {
288                log_path,
289                watermark,
290                expected,
291                actual,
292            } => {
293                return Err(io::Error::other(HaysonIndexMismatchError {
294                    path: path.display().to_string(),
295                    log_path: log_path.display().to_string(),
296                    watermark,
297                    expected_hash: expected,
298                    actual_hash: actual,
299                }));
300            }
301            PairingState::Stale => {}
302        }
303    }
304    let registry = TableRegistry::build_from_log_set_with_declared(schema, logs, declared)?;
305    Ok(Restored {
306        registry,
307        outcome: RestoreOutcome::Rebuilt,
308        gen: 0,
309    })
310}
311
312#[cfg(test)]
313mod tests {
314    use std::fs;
315    use std::os::unix::fs::symlink;
316
317    use serde_json::json;
318
319    use super::*;
320    use crate::checkpoint::{write_snapshot, write_snapshot_set};
321    use crate::registry::fulltext_index::SearchMode;
322    use crate::registry::sorted_index::{RangeBounds, SortValue};
323    use crate::schema::parse_schema;
324    use crate::storage::record::{serialize_entry, LogMeta, LogOp};
325
326    const SCHEMA_HASH: &str = "schema-hash-1";
327
328    fn users_schema() -> TableSchema {
329        parse_schema(&json!({
330            "schemaVersion": 1,
331            "table": "users",
332            "primaryKey": "id",
333            "fields": {
334                "id": { "type": "string" },
335                "name": { "type": "string", "fulltext": true },
336                "role": { "type": "string", "index": "exact" },
337                "age": { "type": "int", "index": "sorted" },
338            },
339        }))
340        .unwrap()
341    }
342
343    fn notes_schema() -> TableSchema {
344        parse_schema(&json!({
345            "schemaVersion": 1,
346            "table": "notes",
347            "primaryKey": "id",
348            "partitionBy": "kind",
349            "fields": {
350                "id": { "type": "string" },
351                "kind": { "type": "string", "required": true },
352            },
353        }))
354        .unwrap()
355    }
356
357    fn note_line(id: &str, seq: u64, kind: &str) -> String {
358        let mut data = crate::schema::HaysonRecord::new();
359        data.insert("id".into(), json!(id));
360        data.insert("kind".into(), json!(kind));
361        serialize_entry(
362            &LogMeta {
363                key: id.into(),
364                op: LogOp::Put,
365                seq,
366                ts: 1,
367            },
368            Some(&data),
369        )
370    }
371
372    fn put_line(id: &str, seq: u64, name: &str, role: &str, age: i64) -> String {
373        let mut data = crate::schema::HaysonRecord::new();
374        data.insert("id".into(), json!(id));
375        data.insert("name".into(), json!(name));
376        data.insert("role".into(), json!(role));
377        data.insert("age".into(), json!(age));
378        serialize_entry(
379            &LogMeta {
380                key: id.into(),
381                op: LogOp::Put,
382                seq,
383                ts: 1,
384            },
385            Some(&data),
386        )
387    }
388
389    fn queries(registry: &mut TableRegistry) -> (Vec<String>, Vec<String>, Vec<String>) {
390        let mut by_role: Vec<String> = registry
391            .lookup_exact("role", &json!("admin"))
392            .cloned()
393            .unwrap_or_default()
394            .into_iter()
395            .collect();
396        by_role.sort();
397
398        let by_age = registry.sorted_index("age").map(|index| {
399            index.range(&RangeBounds {
400                gte: Some(vec![SortValue::Num(30.0)]),
401                ..Default::default()
402            })
403        });
404        let mut by_age = by_age.unwrap_or_default();
405        by_age.sort();
406
407        let mut fulltext: Vec<String> = registry.search("kowalska", SearchMode::All).into_keys().collect();
408        fulltext.sort();
409
410        (by_role, by_age, fulltext)
411    }
412
413    fn seeded_log(dir: &std::path::Path) -> LogFile {
414        let path = dir.join("users.jsonl");
415        let mut log = LogFile::open(&path, false).unwrap();
416        log.append(&[
417            put_line("user_1", 1, "Ania Kowalska", "admin", 31),
418            put_line("user_2", 2, "Piotr Nowak", "user", 24),
419        ])
420        .unwrap();
421        log
422    }
423
424    #[test]
425    fn restores_a_registry_that_answers_queries_identically_to_the_one_it_was_snapshotted_from() {
426        let dir = tempfile::tempdir().unwrap();
427        let idx_path = dir.path().join("users.idx");
428        let log = seeded_log(dir.path());
429
430        let mut original = TableRegistry::build_from_log(users_schema(), &log).unwrap();
431        let before = queries(&mut original);
432        write_snapshot(&idx_path, &original, "users", SCHEMA_HASH, 0, &log).unwrap();
433
434        let (mut restored, outcome) = restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH).unwrap();
435        assert_eq!(outcome, RestoreOutcome::Restored);
436        assert_eq!(queries(&mut restored), before);
437        assert_eq!(restored.records(), original.records());
438        assert_eq!(restored.max_seq(), original.max_seq());
439    }
440
441    #[test]
442    fn falls_back_to_a_full_scan_when_the_log_grew_after_the_snapshot_was_taken() {
443        let dir = tempfile::tempdir().unwrap();
444        let idx_path = dir.path().join("users.idx");
445        let mut log = seeded_log(dir.path());
446
447        let registry_at_snapshot = TableRegistry::build_from_log(users_schema(), &log).unwrap();
448        write_snapshot(&idx_path, &registry_at_snapshot, "users", SCHEMA_HASH, 0, &log).unwrap();
449
450        log.append(&[put_line("user_3", 3, "Zosia Wojcik", "admin", 40)])
451            .unwrap();
452
453        let (mut restored, outcome) = restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH).unwrap();
454        assert_eq!(outcome, RestoreOutcome::Rebuilt);
455
456        let mut full = TableRegistry::build_from_log(users_schema(), &log).unwrap();
457        assert_eq!(queries(&mut restored), queries(&mut full));
458        assert_eq!(restored.records(), 3);
459    }
460
461    #[test]
462    fn rejects_a_snapshot_of_a_log_generation_that_was_rewritten_and_then_grown_past_it() {
463        let dir = tempfile::tempdir().unwrap();
464        let idx_path = dir.path().join("users.idx");
465        let mut log = seeded_log(dir.path());
466
467        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
468        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();
469        let stranded = fs::read(&idx_path).unwrap();
470
471        log.truncate(0).unwrap();
472        let mut filler = 1;
473        while log.size() <= stranded.len() as u64 {
474            log.append(&[put_line(
475                &format!("filler_{filler}"),
476                filler,
477                &format!("Filler {filler}"),
478                "user",
479                30,
480            )])
481            .unwrap();
482            filler += 1;
483        }
484        assert!(
485            log.size() > stranded.len() as u64,
486            "the rewritten log outgrew the snapshot it will be paired against"
487        );
488        fs::write(&idx_path, &stranded).unwrap();
489
490        let err = match restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH) {
491            Err(err) => err,
492            Ok((_, outcome)) => {
493                panic!("expected a snapshot of a replaced log generation to be rejected, got {outcome:?}")
494            }
495        };
496        assert!(err
497            .get_ref()
498            .unwrap()
499            .downcast_ref::<crate::errors::HaysonIndexMismatchError>()
500            .is_some());
501        assert_eq!(fs::read(&idx_path).unwrap(), stranded);
502    }
503
504    #[test]
505    fn rejects_a_partitioned_snapshot_of_a_partition_that_was_rewritten_and_then_grown_past_it() {
506        let dir = tempfile::tempdir().unwrap();
507        let idx_path = dir.path().join("notes.idx");
508        let mut logs = LogSet::partitioned(dir.path().join("notes"), false).unwrap();
509        logs.append("adr", &[note_line("a1", 1, "adr")]).unwrap();
510        logs.append("note", &[note_line("n1", 2, "note")]).unwrap();
511
512        let registry = TableRegistry::build_from_log_set_with_declared(notes_schema(), &mut logs, &[]).unwrap();
513        write_snapshot_set(&idx_path, &registry, "notes", SCHEMA_HASH, 0, &logs).unwrap();
514        let stranded = fs::read(&idx_path).unwrap();
515
516        let replaced = logs.log_of_mut("note").unwrap();
517        replaced.truncate(0).unwrap();
518        replaced
519            .append(&[note_line("n2", 3, "note"), note_line("n3", 4, "note")])
520            .unwrap();
521
522        let err = match restore_or_rebuild_set_with_declared(&idx_path, notes_schema(), &mut logs, SCHEMA_HASH, &[]) {
523            Err(err) => err,
524            Ok(_) => panic!("expected a snapshot of a replaced partition generation to be rejected"),
525        };
526        let mismatch = err
527            .get_ref()
528            .unwrap()
529            .downcast_ref::<crate::errors::HaysonIndexMismatchError>()
530            .expect("a replaced partition generation is rejected as an index mismatch");
531        assert_eq!(mismatch.log_path, logs.path_of("note").display().to_string());
532        assert_eq!(fs::read(&idx_path).unwrap(), stranded);
533    }
534
535    /// A log comfortably past `PAIRING_WINDOW` (4 KiB), so the leading-window check has a full
536    /// window on both sides — below that the short-log comparison declines to judge.
537    fn fat_log(path: &std::path::Path, tag: &str, at_least: u64) -> LogFile {
538        let mut log = LogFile::open(path, false).unwrap();
539        let mut seq = 0;
540        while log.size() < at_least {
541            seq += 1;
542            log.append(&[put_line(
543                &format!("{tag}_{seq}"),
544                seq,
545                &format!("{tag} Kowalska {seq}"),
546                "admin",
547                30 + seq as i64,
548            )])
549            .unwrap();
550        }
551        log
552    }
553
554    #[test]
555    fn keeps_rebuilding_quietly_when_the_log_vanished_and_reopened_empty_under_its_snapshot() {
556        let dir = tempfile::tempdir().unwrap();
557        let idx_path = dir.path().join("users.idx");
558        let log_path = dir.path().join("users.jsonl");
559        let log = fat_log(&log_path, "before", 8192);
560
561        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
562        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();
563
564        fs::remove_file(&log_path).unwrap();
565        let recreated = LogFile::open(&log_path, false).unwrap();
566        assert_eq!(recreated.size(), 0, "a writable open recreates the log empty");
567
568        let (rebuilt, outcome) = restore_or_rebuild(&idx_path, users_schema(), &recreated, SCHEMA_HASH).unwrap();
569        assert_eq!(outcome, RestoreOutcome::Rebuilt);
570        assert_eq!(rebuilt.records(), 0);
571    }
572
573    #[test]
574    fn keeps_rebuilding_quietly_when_an_unflushed_tail_was_lost_but_the_leading_window_agrees() {
575        let dir = tempfile::tempdir().unwrap();
576        let idx_path = dir.path().join("users.idx");
577        let mut log = fat_log(&dir.path().join("users.jsonl"), "kept", 8192);
578
579        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
580        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();
581        let watermark = log.size();
582
583        log.truncate(6144).unwrap();
584        assert!(
585            log.size() < watermark && log.size() > 4096,
586            "a lost tail over an intact window"
587        );
588
589        let (_, outcome) = restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH).unwrap();
590        assert_eq!(outcome, RestoreOutcome::Rebuilt);
591    }
592
593    #[test]
594    fn rejects_a_snapshot_laid_over_a_shorter_log_it_was_never_taken_of() {
595        let dir = tempfile::tempdir().unwrap();
596        let idx_path = dir.path().join("users.idx");
597        let stranger = fat_log(&dir.path().join("stranger.jsonl"), "stranger", 8192);
598
599        let theirs = fat_log(&dir.path().join("users.jsonl"), "theirs", 12288);
600        let registry = TableRegistry::build_from_log(users_schema(), &theirs).unwrap();
601        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &theirs).unwrap();
602        assert!(
603            stranger.size() < theirs.size(),
604            "the planted snapshot outruns the log it meets"
605        );
606
607        let err = match restore_or_rebuild(&idx_path, users_schema(), &stranger, SCHEMA_HASH) {
608            Err(err) => err,
609            Ok((_, outcome)) => {
610                panic!("expected a snapshot of an unrelated log to be rejected, got {outcome:?}")
611            }
612        };
613        let mismatch = err
614            .get_ref()
615            .unwrap()
616            .downcast_ref::<crate::errors::HaysonIndexMismatchError>()
617            .expect("an unrelated log is rejected as an index mismatch");
618        assert_eq!(mismatch.log_path, stranger.path().display().to_string());
619        assert_eq!(mismatch.watermark, theirs.size());
620        assert!(idx_path.exists(), "a refused snapshot is left on disk to be looked at");
621    }
622
623    #[test]
624    fn a_snapshot_written_without_a_prefix_fingerprint_still_rebuilds_quietly_under_a_short_log() {
625        let dir = tempfile::tempdir().unwrap();
626        let idx_path = dir.path().join("users.idx");
627        let stranger = fat_log(&dir.path().join("stranger.jsonl"), "stranger", 8192);
628
629        let theirs = fat_log(&dir.path().join("users.jsonl"), "theirs", 12288);
630        let registry = TableRegistry::build_from_log(users_schema(), &theirs).unwrap();
631        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &theirs).unwrap();
632
633        let raw = fs::read_to_string(&idx_path).unwrap();
634        let mut lines: Vec<&str> = raw.lines().collect();
635        let aged_header = lines[0].replace(
636            &format!(
637                ",\"prefixHash\":\"{}\"",
638                log_prefix_fingerprint(&theirs, prefix_window(theirs.size())).unwrap()
639            ),
640            "",
641        );
642        assert!(
643            !aged_header.contains("prefixHash"),
644            "the header now looks like a pre-HAY-197 one"
645        );
646        lines[0] = &aged_header;
647        fs::write(&idx_path, format!("{}\n", lines.join("\n"))).unwrap();
648
649        let (_, outcome) = restore_or_rebuild(&idx_path, users_schema(), &stranger, SCHEMA_HASH).unwrap();
650        assert_eq!(outcome, RestoreOutcome::Rebuilt);
651    }
652
653    /// A `.idx` written before the `"snapshotVersion"` rename is not read back — and must not be
654    /// an error either: the snapshot is a cache, so the unrecognized header rebuilds like any
655    /// stale one, which is why this rename needs no old-key tolerance.
656    #[test]
657    fn rebuilds_from_the_log_when_the_snapshot_carries_the_pre_rename_marker() {
658        let dir = tempfile::tempdir().unwrap();
659        let idx_path = dir.path().join("users.idx");
660        let log = seeded_log(dir.path());
661        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
662        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();
663
664        let raw = fs::read_to_string(&idx_path).unwrap();
665        let aged = raw.replacen("\"snapshotVersion\"", "\"hayson\"", 1);
666        assert_ne!(aged, raw, "the fixture has to be the header this rename touched");
667        fs::write(&idx_path, &aged).unwrap();
668
669        let (mut restored, outcome) = restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH).unwrap();
670        assert_eq!(outcome, RestoreOutcome::Rebuilt);
671
672        let mut full = TableRegistry::build_from_log(users_schema(), &log).unwrap();
673        assert_eq!(queries(&mut restored), queries(&mut full));
674    }
675
676    #[test]
677    fn falls_back_without_erroring_when_the_snapshot_file_is_truncated_garbage() {
678        let dir = tempfile::tempdir().unwrap();
679        let idx_path = dir.path().join("users.idx");
680        let log = seeded_log(dir.path());
681
682        fs::write(&idx_path, b"{\"snapshotVersion\":1,\"kind\":\"index\"").unwrap();
683
684        let (mut restored, outcome) = restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH).unwrap();
685        assert_eq!(outcome, RestoreOutcome::Rebuilt);
686
687        let mut full = TableRegistry::build_from_log(users_schema(), &log).unwrap();
688        assert_eq!(queries(&mut restored), queries(&mut full));
689    }
690
691    #[test]
692    fn falls_back_without_erroring_when_no_snapshot_file_exists() {
693        let dir = tempfile::tempdir().unwrap();
694        let idx_path = dir.path().join("users.idx");
695        let log = seeded_log(dir.path());
696
697        let (registry, outcome) = restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH).unwrap();
698        assert_eq!(outcome, RestoreOutcome::Rebuilt);
699        assert_eq!(registry.records(), 2);
700    }
701
702    #[test]
703    fn rejects_a_snapshot_whose_schema_hash_no_longer_matches() {
704        let dir = tempfile::tempdir().unwrap();
705        let idx_path = dir.path().join("users.idx");
706        let log = seeded_log(dir.path());
707
708        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
709        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();
710
711        let (_, outcome) = restore_or_rebuild(&idx_path, users_schema(), &log, "schema-hash-2").unwrap();
712        assert_eq!(outcome, RestoreOutcome::Rebuilt);
713    }
714
715    #[test]
716    fn rejects_a_paired_snapshot_whose_content_fingerprint_was_tampered_with() {
717        let dir = tempfile::tempdir().unwrap();
718        let idx_path = dir.path().join("users.idx");
719        let log = seeded_log(dir.path());
720
721        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
722        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();
723
724        let raw = fs::read_to_string(&idx_path).unwrap();
725        let mut lines: Vec<&str> = raw.lines().collect();
726        let tampered_header = lines[0].replace(
727            &format!("\"logHash\":\"{}\"", log_fingerprint(&log, log.size()).unwrap()),
728            "\"logHash\":\"0000000000000000\"",
729        );
730        lines[0] = &tampered_header;
731        fs::write(&idx_path, format!("{}\n", lines.join("\n"))).unwrap();
732
733        let result = restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH);
734        let err = match result {
735            Err(err) => err,
736            Ok(_) => panic!("expected a tampered, paired snapshot to be rejected"),
737        };
738        assert!(err
739            .get_ref()
740            .unwrap()
741            .downcast_ref::<crate::errors::HaysonIndexMismatchError>()
742            .is_some());
743        assert!(idx_path.exists());
744    }
745
746    #[test]
747    fn a_rejected_snapshot_names_the_index_the_log_and_the_watermark_it_claimed() {
748        let dir = tempfile::tempdir().unwrap();
749        let idx_path = dir.path().join("users.idx");
750        let log = seeded_log(dir.path());
751
752        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
753        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();
754
755        let raw = fs::read_to_string(&idx_path).unwrap();
756        let mut lines: Vec<&str> = raw.lines().collect();
757        let tampered_header = lines[0].replace(
758            &format!("\"logHash\":\"{}\"", log_fingerprint(&log, log.size()).unwrap()),
759            "\"logHash\":\"0000000000000000\"",
760        );
761        lines[0] = &tampered_header;
762        fs::write(&idx_path, format!("{}\n", lines.join("\n"))).unwrap();
763
764        let err = match restore_or_rebuild(&idx_path, users_schema(), &log, SCHEMA_HASH) {
765            Err(err) => err,
766            Ok(_) => panic!("expected a tampered, paired snapshot to be rejected"),
767        };
768        let mismatch = err
769            .get_ref()
770            .unwrap()
771            .downcast_ref::<crate::errors::HaysonIndexMismatchError>()
772            .expect("a tampered, paired snapshot is rejected as an index mismatch");
773
774        assert_eq!(mismatch.path, idx_path.display().to_string());
775        assert_eq!(mismatch.log_path, log.path().display().to_string());
776        assert_eq!(mismatch.watermark, log.size());
777        assert_eq!(mismatch.expected_hash, "0000000000000000");
778        assert_eq!(mismatch.actual_hash, log_fingerprint(&log, log.size()).unwrap());
779    }
780
781    #[test]
782    fn a_rejected_partitioned_snapshot_names_the_partition_whose_fingerprint_disagreed() {
783        let dir = tempfile::tempdir().unwrap();
784        let idx_path = dir.path().join("notes.idx");
785        let mut logs = LogSet::partitioned(dir.path().join("notes"), false).unwrap();
786        logs.append("adr", &[note_line("a1", 1, "adr")]).unwrap();
787        logs.append("note", &[note_line("n1", 2, "note")]).unwrap();
788
789        let registry = TableRegistry::build_from_log_set_with_declared(notes_schema(), &mut logs, &[]).unwrap();
790        write_snapshot_set(&idx_path, &registry, "notes", SCHEMA_HASH, 0, &logs).unwrap();
791
792        let tampered = logs.log_of("note").unwrap();
793        let raw = fs::read_to_string(&idx_path).unwrap();
794        let mut lines: Vec<&str> = raw.lines().collect();
795        let tampered_header = lines[0].replace(
796            &format!("\"h\":\"{}\"", log_fingerprint(tampered, tampered.size()).unwrap()),
797            "\"h\":\"0000000000000000\"",
798        );
799        lines[0] = &tampered_header;
800        fs::write(&idx_path, format!("{}\n", lines.join("\n"))).unwrap();
801
802        let err = match restore_or_rebuild_set_with_declared(&idx_path, notes_schema(), &mut logs, SCHEMA_HASH, &[]) {
803            Err(err) => err,
804            Ok(_) => panic!("expected a tampered, paired partition to be rejected"),
805        };
806        let mismatch = err
807            .get_ref()
808            .unwrap()
809            .downcast_ref::<crate::errors::HaysonIndexMismatchError>()
810            .expect("a tampered, paired partition is rejected as an index mismatch");
811
812        assert_eq!(mismatch.path, idx_path.display().to_string());
813        assert_eq!(mismatch.log_path, logs.path_of("note").display().to_string());
814        assert_eq!(mismatch.watermark, logs.log_of("note").unwrap().size());
815    }
816
817    #[test]
818    fn write_snapshot_is_atomic_and_does_not_write_through_a_symlink_at_the_target_path() {
819        let dir = tempfile::tempdir().unwrap();
820        let victim_dir = tempfile::tempdir().unwrap();
821        let victim = victim_dir.path().join("precious.idx");
822        fs::write(&victim, "PRECIOUS").unwrap();
823
824        let idx_path = dir.path().join("users.idx");
825        symlink(&victim, &idx_path).unwrap();
826
827        let log = seeded_log(dir.path());
828        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
829        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();
830
831        assert_eq!(fs::read_to_string(&victim).unwrap(), "PRECIOUS");
832        assert!(!fs::symlink_metadata(&idx_path).unwrap().file_type().is_symlink());
833        assert!(fs::read_to_string(&idx_path).unwrap().contains("\"table\":\"users\""));
834    }
835
836    #[test]
837    fn write_snapshot_leaves_no_temporary_file_behind() {
838        let dir = tempfile::tempdir().unwrap();
839        let idx_path = dir.path().join("users.idx");
840        let log = seeded_log(dir.path());
841        let registry = TableRegistry::build_from_log(users_schema(), &log).unwrap();
842
843        write_snapshot(&idx_path, &registry, "users", SCHEMA_HASH, 0, &log).unwrap();
844
845        assert!(!dir.path().join("users.idx.tmp").exists());
846    }
847}