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