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
18pub struct Restored {
21 pub registry: TableRegistry,
22 pub outcome: RestoreOutcome,
23 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
40fn 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
74fn 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
125fn 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 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#[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#[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
247pub 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, ®istry_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, ®istry, "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, ®istry, "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 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, ®istry, "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, ®istry, "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, ®istry, "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, ®istry, "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 #[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, ®istry, "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, ®istry, "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, ®istry, "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, ®istry, "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, ®istry, "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, ®istry, "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, ®istry, "users", SCHEMA_HASH, 0, &log).unwrap();
828
829 assert!(!dir.path().join("users.idx.tmp").exists());
830 }
831}