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
21pub struct Restored {
24 pub registry: TableRegistry,
25 pub outcome: RestoreOutcome,
26 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
43fn 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
77fn 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
128fn 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 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#[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#[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
250pub 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
268pub 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, ®istry_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, ®istry, "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, ®istry, "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 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, ®istry, "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, ®istry, "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, ®istry, "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, ®istry, "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 #[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, ®istry, "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, ®istry, "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, ®istry, "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, ®istry, "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, ®istry, "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, ®istry, "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, ®istry, "users", SCHEMA_HASH, 0, &log).unwrap();
844
845 assert!(!dir.path().join("users.idx.tmp").exists());
846 }
847}