1use crate::fold::{fold, state_hash, SyncState};
53use crate::oplog::{canonical_json, verify_log, ChainError, DeviceLog, Hlc, OpRecord};
54use serde::{Deserialize, Serialize};
55use sha2::{Digest, Sha256};
56use std::collections::{BTreeMap, BTreeSet};
57use std::fmt;
58use std::fs::{self, File};
59use std::io::Write;
60use std::path::{Path, PathBuf};
61
62#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
66pub struct FrontierEntry {
67 pub seq: u64,
68 pub hlc: Hlc,
69 pub head: String,
70}
71
72#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
74pub struct Checkpoint {
75 pub frontier: BTreeMap<String, FrontierEntry>,
78 pub scopes: Vec<String>,
81 pub state_hash: String,
85 pub checkpoint_hash: String,
89 pub state: SyncState,
92}
93
94impl Checkpoint {
95 pub fn assemble(
98 frontier: BTreeMap<String, FrontierEntry>,
99 scopes: Vec<String>,
100 state: SyncState,
101 ) -> Self {
102 let hash = state_hash(&state);
103 let mut checkpoint = Self {
104 frontier,
105 scopes,
106 state_hash: hash,
107 checkpoint_hash: String::new(),
108 state,
109 };
110 checkpoint.checkpoint_hash = checkpoint.content_hash();
111 checkpoint
112 }
113
114 pub fn from_ops(ops: &[OpRecord]) -> Result<Self, ChainError> {
120 verify_log(ops)?;
121 let state = fold(ops);
122 let mut frontier: BTreeMap<String, FrontierEntry> = BTreeMap::new();
123 let mut scopes: BTreeSet<String> = BTreeSet::new();
124 for op in ops {
125 scopes.insert(op.scope.tag());
126 let replace = match frontier.get(&op.device_id) {
127 Some(existing) => op.seq > existing.seq,
128 None => true,
129 };
130 if replace {
131 frontier.insert(
132 op.device_id.clone(),
133 FrontierEntry {
134 seq: op.seq,
135 hlc: op.hlc.clone(),
136 head: op.op_id.clone(),
137 },
138 );
139 }
140 }
141 Ok(Self::assemble(
142 frontier,
143 scopes.into_iter().collect(),
144 state,
145 ))
146 }
147
148 pub fn content_hash(&self) -> String {
153 let mut value = serde_json::to_value(self).expect("Checkpoint serializes");
154 if let Some(obj) = value.as_object_mut() {
155 obj.remove("checkpoint_hash");
156 }
157 let mut hasher = Sha256::new();
158 hasher.update(canonical_json(&value).as_bytes());
159 let digest = hasher.finalize();
160 let hex: String = digest.iter().take(16).map(|b| format!("{b:02x}")).collect();
161 format!("ckpt-{hex}")
162 }
163
164 pub fn verify(&self) -> Result<(), CheckpointError> {
170 let actual_state = state_hash(&self.state);
171 if actual_state != self.state_hash {
172 return Err(CheckpointError::HashMismatch {
173 expected: self.state_hash.clone(),
174 actual: actual_state,
175 });
176 }
177 let actual_content = self.content_hash();
178 if actual_content != self.checkpoint_hash {
179 return Err(CheckpointError::ContentMismatch {
180 expected: self.checkpoint_hash.clone(),
181 actual: actual_content,
182 });
183 }
184 Ok(())
185 }
186
187 pub fn file_name(&self) -> String {
192 format!("{}.checkpoint.json", self.checkpoint_hash)
193 }
194
195 pub fn save(&self, dir: &Path) -> std::io::Result<PathBuf> {
200 fs::create_dir_all(dir)?;
201 let final_path = dir.join(self.file_name());
202 let tmp_path = dir.join(format!("{}.tmp", self.file_name()));
203 {
204 let mut tmp = File::create(&tmp_path)?;
205 tmp.write_all(
206 serde_json::to_string(self)
207 .map_err(std::io::Error::other)?
208 .as_bytes(),
209 )?;
210 tmp.sync_all()?;
211 }
212 fs::rename(&tmp_path, &final_path)?;
213 #[cfg(unix)]
215 {
216 let _ = File::open(dir).and_then(|d| d.sync_all());
217 }
218 Ok(final_path)
219 }
220
221 pub fn load(path: &Path) -> Result<Self, CheckpointError> {
228 let raw = fs::read_to_string(path).map_err(CheckpointError::Io)?;
229 let checkpoint: Checkpoint = serde_json::from_str(&raw).map_err(CheckpointError::Parse)?;
230 checkpoint.verify()?;
231 let expected = checkpoint.file_name();
232 let found = path
233 .file_name()
234 .map(|n| n.to_string_lossy().into_owned())
235 .unwrap_or_default();
236 if found != expected {
237 return Err(CheckpointError::AddressMismatch { expected, found });
238 }
239 Ok(checkpoint)
240 }
241}
242
243#[derive(Debug)]
245pub enum CheckpointError {
246 Io(std::io::Error),
247 Parse(serde_json::Error),
248 HashMismatch {
251 expected: String,
252 actual: String,
253 },
254 ContentMismatch {
258 expected: String,
259 actual: String,
260 },
261 AddressMismatch {
264 expected: String,
265 found: String,
266 },
267}
268
269impl fmt::Display for CheckpointError {
270 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
271 match self {
272 CheckpointError::Io(e) => write!(f, "checkpoint io error: {e}"),
273 CheckpointError::Parse(e) => write!(f, "checkpoint parse error: {e}"),
274 CheckpointError::HashMismatch { expected, actual } => write!(
275 f,
276 "checkpoint state_hash mismatch (stored {expected}, recomputed {actual}) — \
277 tampered or corrupt snapshot, refusing to fold it"
278 ),
279 CheckpointError::ContentMismatch { expected, actual } => write!(
280 f,
281 "checkpoint content-address mismatch (stored {expected}, recomputed {actual}) — \
282 frontier/scopes tampered or record corrupt, refusing to fold or anchor on it"
283 ),
284 CheckpointError::AddressMismatch { expected, found } => write!(
285 f,
286 "checkpoint file name {found} is not its content address {expected} — \
287 renamed or wrongly-stored checkpoint, refusing to load it"
288 ),
289 }
290 }
291}
292
293impl std::error::Error for CheckpointError {}
294
295#[derive(Debug)]
297pub enum AnchorError {
298 Checkpoint(CheckpointError),
300 Chain(ChainError),
302 BrokenAnchor { device_id: String, detail: String },
305 UnanchoredDevice { device_id: String, first_seq: u64 },
308}
309
310impl fmt::Display for AnchorError {
311 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
312 match self {
313 AnchorError::Checkpoint(e) => write!(f, "anchor checkpoint invalid: {e}"),
314 AnchorError::Chain(e) => write!(f, "anchored tail chain invalid: {e}"),
315 AnchorError::BrokenAnchor { device_id, detail } => {
316 write!(
317 f,
318 "device {device_id}: tail does not anchor on checkpoint ({detail})"
319 )
320 }
321 AnchorError::UnanchoredDevice {
322 device_id,
323 first_seq,
324 } => write!(
325 f,
326 "device {device_id}: first op has seq {first_seq} but the checkpoint \
327 records no frontier for it — unanchored truncation"
328 ),
329 }
330 }
331}
332
333impl std::error::Error for AnchorError {}
334
335pub fn verify_anchored(checkpoint: &Checkpoint, tail: &[OpRecord]) -> Result<(), AnchorError> {
351 checkpoint.verify().map_err(AnchorError::Checkpoint)?;
352 verify_log(tail).map_err(AnchorError::Chain)?;
353
354 let mut first: BTreeMap<&str, &OpRecord> = BTreeMap::new();
356 for op in tail {
357 let replace = match first.get(op.device_id.as_str()) {
358 Some(existing) => op.seq < existing.seq,
359 None => true,
360 };
361 if replace {
362 first.insert(&op.device_id, op);
363 }
364 }
365
366 for (device_id, op) in first {
367 match checkpoint.frontier.get(device_id) {
368 Some(anchor) => {
369 if op.seq != anchor.seq + 1 {
370 return Err(AnchorError::BrokenAnchor {
371 device_id: device_id.to_string(),
372 detail: format!(
373 "first retained seq {} does not continue frontier seq {}",
374 op.seq, anchor.seq
375 ),
376 });
377 }
378 if op.prev.as_deref() != Some(anchor.head.as_str()) {
379 return Err(AnchorError::BrokenAnchor {
380 device_id: device_id.to_string(),
381 detail: format!(
382 "first retained op's prev does not link to frontier head {}",
383 anchor.head
384 ),
385 });
386 }
387 if op.hlc <= anchor.hlc {
388 return Err(AnchorError::BrokenAnchor {
389 device_id: device_id.to_string(),
390 detail: "first retained op's hlc does not advance past the frontier"
391 .to_string(),
392 });
393 }
394 }
395 None => {
396 if op.seq != 0 {
397 return Err(AnchorError::UnanchoredDevice {
398 device_id: device_id.to_string(),
399 first_seq: op.seq,
400 });
401 }
402 }
403 }
404 }
405 Ok(())
406}
407
408pub fn resume_anchored(
420 device_id: impl Into<String>,
421 checkpoint: &Checkpoint,
422 tail: &[OpRecord],
423) -> Result<DeviceLog, AnchorError> {
424 verify_anchored(checkpoint, tail)?;
425 let device_id = device_id.into();
426 let mut log = DeviceLog::new(device_id.clone());
427 for entry in checkpoint.frontier.values() {
428 log.clock.observe(&entry.hlc);
429 }
430 if let Some(anchor) = checkpoint.frontier.get(&device_id) {
431 log.next_seq = anchor.seq + 1;
432 log.prev = Some(anchor.head.clone());
433 }
434 for op in tail {
435 log.clock.observe(&op.hlc);
436 if op.device_id == device_id && op.seq >= log.next_seq {
437 log.next_seq = op.seq + 1;
438 log.prev = Some(op.op_id.clone());
439 }
440 }
441 Ok(log)
442}
443
444#[cfg(test)]
445mod tests {
446 use super::*;
447 use crate::fold::fold_onto;
448 use crate::oplog::{Scope, Surface};
449 use serde_json::json;
450
451 fn ops_with_cut() -> (Vec<OpRecord>, usize) {
454 let mut a = DeviceLog::new("dev-a");
455 let mut b = DeviceLog::new("dev-b");
456 let mut ops = vec![
457 a.append(
458 Scope::Personal,
459 Surface::Knowledge,
460 json!({"id": "f1", "v": 1}),
461 ),
462 a.append(
463 Scope::Shared { org: "acme".into() },
464 Surface::Declagent,
465 json!({"id": "agent-1", "rev": "a"}),
466 ),
467 ];
468 for op in &ops {
469 b.observe(&op.hlc);
470 }
471 ops.push(b.append(
472 Scope::Personal,
473 Surface::Knowledge,
474 json!({"id": "f2", "v": 2}),
475 ));
476 let split = ops.len();
477 ops.push(a.append(
478 Scope::Personal,
479 Surface::Conversation,
480 json!({"speaker": "u", "text": "hi", "timestamp": 10}),
481 ));
482 for op in &ops[split..] {
483 b.observe(&op.hlc);
484 }
485 ops.push(b.append(
486 Scope::Shared { org: "acme".into() },
487 Surface::Declagent,
488 json!({"id": "agent-1", "rev": "b"}),
489 ));
490 (ops, split)
491 }
492
493 #[test]
494 fn from_ops_records_frontier_heads_and_scopes() {
495 let (ops, split) = ops_with_cut();
496 let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
497 assert_eq!(
498 ckpt.scopes,
499 vec!["personal".to_string(), "shared:acme".to_string()]
500 );
501 assert_eq!(ckpt.frontier.len(), 2);
502 assert_eq!(ckpt.frontier["dev-a"].seq, 1);
503 assert_eq!(ckpt.frontier["dev-a"].head, ops[1].op_id);
504 assert_eq!(ckpt.frontier["dev-b"].seq, 0);
505 assert_eq!(ckpt.frontier["dev-b"].head, ops[2].op_id);
506 assert_eq!(ckpt.state_hash, state_hash(&fold(&ops[..split])));
507 }
508
509 #[test]
510 fn from_ops_refuses_an_invalid_log() {
511 let (mut ops, _) = ops_with_cut();
512 ops[0].payload = json!({"forged": true}); assert!(matches!(
514 Checkpoint::from_ops(&ops),
515 Err(ChainError::IdMismatch { .. })
516 ));
517 }
518
519 #[test]
520 fn checkpoint_save_load_round_trips_content_addressed() {
521 let dir = tempfile::tempdir().unwrap();
522 let (ops, split) = ops_with_cut();
523 let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
524
525 let path = ckpt.save(dir.path()).unwrap();
526 assert_eq!(
527 path.file_name().unwrap().to_str().unwrap(),
528 format!("{}.checkpoint.json", ckpt.checkpoint_hash),
529 "file name is the WHOLE-RECORD content address"
530 );
531 let loaded = Checkpoint::load(&path).unwrap();
532 assert_eq!(loaded, ckpt);
533
534 let again = ckpt.save(dir.path()).unwrap();
536 assert_eq!(again, path);
537 let names: Vec<String> = fs::read_dir(dir.path())
539 .unwrap()
540 .map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
541 .collect();
542 assert_eq!(
543 names.len(),
544 1,
545 "only the final checkpoint file exists: {names:?}"
546 );
547 }
548
549 #[test]
550 fn tampered_checkpoint_is_rejected_on_load() {
551 let dir = tempfile::tempdir().unwrap();
552 let (ops, split) = ops_with_cut();
553 let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
554 let path = ckpt.save(dir.path()).unwrap();
555
556 let raw = fs::read_to_string(&path).unwrap();
559 let tampered = raw.replace("\"v\":1", "\"v\":999");
560 assert_ne!(
561 raw, tampered,
562 "tamper target must exist in the serialized state"
563 );
564 fs::write(&path, tampered).unwrap();
565 assert!(matches!(
566 Checkpoint::load(&path),
567 Err(CheckpointError::HashMismatch { .. })
568 ));
569
570 let mut forged = ckpt.clone();
572 forged.state.logs.clear();
573 assert!(matches!(
574 forged.verify(),
575 Err(CheckpointError::HashMismatch { .. })
576 ));
577 }
578
579 #[test]
580 fn tampered_frontier_is_rejected_on_load_and_verify() {
581 let dir = tempfile::tempdir().unwrap();
585 let (ops, split) = ops_with_cut();
586 let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
587 let path = ckpt.save(dir.path()).unwrap();
588
589 let mut value: serde_json::Value =
592 serde_json::from_str(&fs::read_to_string(&path).unwrap()).unwrap();
593 value["frontier"]["dev-a"]["seq"] = json!(7);
594 fs::write(&path, serde_json::to_string(&value).unwrap()).unwrap();
595 assert!(matches!(
596 Checkpoint::load(&path),
597 Err(CheckpointError::ContentMismatch { .. })
598 ));
599
600 let mut forged = ckpt.clone();
603 forged.frontier.get_mut("dev-a").unwrap().seq = 7;
604 assert!(matches!(
605 forged.verify(),
606 Err(CheckpointError::ContentMismatch { .. })
607 ));
608 assert!(matches!(
609 verify_anchored(&forged, &ops[split..]),
610 Err(AnchorError::Checkpoint(
611 CheckpointError::ContentMismatch { .. }
612 ))
613 ));
614
615 let good = ckpt.save(dir.path()).unwrap();
617 let renamed = dir.path().join("latest.checkpoint.json");
618 fs::rename(&good, &renamed).unwrap();
619 assert!(matches!(
620 Checkpoint::load(&renamed),
621 Err(CheckpointError::AddressMismatch { .. })
622 ));
623 }
624
625 #[test]
626 fn different_frontiers_never_share_a_content_address() {
627 let mut a = DeviceLog::new("a");
639 let mut b = DeviceLog::new("b");
640 let fact = json!({"id": "f1", "body": "hi"});
641 let oa = a.append(Scope::Personal, Surface::Knowledge, fact.clone());
642 let ob = b.append(Scope::Personal, Surface::Knowledge, fact);
643
644 let just_a = Checkpoint::from_ops(std::slice::from_ref(&oa)).unwrap();
645 let both = Checkpoint::from_ops(&[oa, ob]).unwrap();
646
647 assert_eq!(
648 just_a.state, both.state,
649 "cross-device dedup: identical folded state"
650 );
651 assert_eq!(
652 just_a.state_hash, both.state_hash,
653 "state hash agrees (divergence invariant)"
654 );
655 assert_ne!(just_a.frontier, both.frontier, "but the frontiers differ");
656 assert_ne!(
657 just_a.checkpoint_hash, both.checkpoint_hash,
658 "so the content addresses MUST differ"
659 );
660 assert_ne!(
661 just_a.file_name(),
662 both.file_name(),
663 "…and so must the file names"
664 );
665
666 let dir = tempfile::tempdir().unwrap();
668 let p1 = just_a.save(dir.path()).unwrap();
669 let p2 = both.save(dir.path()).unwrap();
670 assert_ne!(p1, p2);
671 assert_eq!(Checkpoint::load(&p1).unwrap(), just_a);
672 assert_eq!(Checkpoint::load(&p2).unwrap(), both);
673 }
674
675 #[test]
676 fn verify_anchored_accepts_the_truncated_composition() {
677 let (ops, split) = ops_with_cut();
678 let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
679 let tail = &ops[split..];
680 verify_log(tail).expect("verify_log alone accepts a non-zero-seq chain");
681 verify_anchored(&ckpt, tail).expect("checkpoint anchors the truncated tail");
682 assert_eq!(fold_onto(&ckpt.state, tail), fold(&ops));
684 }
685
686 #[test]
687 fn verify_anchored_rejects_breaks() {
688 let (ops, split) = ops_with_cut();
689 let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
690 let tail: Vec<OpRecord> = ops[split..].to_vec();
691
692 let mut a_extra = resume_anchored("dev-a", &ckpt, &tail).unwrap();
699 let extra = a_extra.append(Scope::Personal, Surface::Knowledge, json!({"id": "f9"}));
700 let mut with_extra = tail.clone();
701 with_extra.push(extra.clone());
702 verify_anchored(&ckpt, &with_extra).unwrap();
703
704 let holed: Vec<OpRecord> = with_extra
709 .iter()
710 .filter(|o| o.op_id != tail[0].op_id)
711 .cloned()
712 .collect();
713 assert!(matches!(
714 verify_anchored(&ckpt, &holed),
715 Err(AnchorError::BrokenAnchor { .. })
716 ));
717
718 let mut stranger = DeviceLog::new("dev-c");
720 stranger.append(Scope::Personal, Surface::Knowledge, json!({"id": "s0"}));
721 let s1 = stranger.append(Scope::Personal, Surface::Knowledge, json!({"id": "s1"}));
722 let mut with_stranger = tail.clone();
723 with_stranger.push(s1); assert!(matches!(
725 verify_anchored(&ckpt, &with_stranger),
726 Err(AnchorError::UnanchoredDevice { first_seq: 1, .. })
727 ));
728
729 let mut forged = ckpt.clone();
731 forged.state.logs.clear();
732 assert!(matches!(
733 verify_anchored(&forged, &tail),
734 Err(AnchorError::Checkpoint(
735 CheckpointError::HashMismatch { .. }
736 ))
737 ));
738 }
739
740 #[test]
741 fn resume_anchored_continues_chains_after_truncation() {
742 let (ops, split) = ops_with_cut();
743 let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
744 let tail: Vec<OpRecord> = ops[split..].to_vec();
745
746 let mut a = resume_anchored("dev-a", &ckpt, &tail).unwrap();
748 let next_a = a.append(Scope::Personal, Surface::Knowledge, json!({"id": "na"}));
749 assert_eq!(next_a.seq, 3);
750 assert_eq!(next_a.prev.as_deref(), Some(tail[0].op_id.as_str()));
751
752 let ckpt_all = Checkpoint::from_ops(&ops).unwrap();
755 let mut b = resume_anchored("dev-b", &ckpt_all, &[]).unwrap();
756 let next_b = b.append(Scope::Personal, Surface::Knowledge, json!({"id": "nb"}));
757 assert_eq!(next_b.seq, 2, "continues past the checkpointed chain");
758 assert_eq!(
759 next_b.prev.as_deref(),
760 Some(ckpt_all.frontier["dev-b"].head.as_str())
761 );
762 assert!(
763 next_b.hlc > ckpt_all.frontier["dev-a"].hlc
764 && next_b.hlc > ckpt_all.frontier["dev-b"].hlc,
765 "lamport advanced past everything the checkpoint covers"
766 );
767
768 let mut composed = tail.clone();
770 composed.push(next_a);
771 verify_anchored(&ckpt, &composed).unwrap();
772 }
773}