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(frontier, scopes.into_iter().collect(), state))
142 }
143
144 pub fn content_hash(&self) -> String {
149 let mut value = serde_json::to_value(self).expect("Checkpoint serializes");
150 if let Some(obj) = value.as_object_mut() {
151 obj.remove("checkpoint_hash");
152 }
153 let mut hasher = Sha256::new();
154 hasher.update(canonical_json(&value).as_bytes());
155 let digest = hasher.finalize();
156 let hex: String = digest.iter().take(16).map(|b| format!("{b:02x}")).collect();
157 format!("ckpt-{hex}")
158 }
159
160 pub fn verify(&self) -> Result<(), CheckpointError> {
166 let actual_state = state_hash(&self.state);
167 if actual_state != self.state_hash {
168 return Err(CheckpointError::HashMismatch {
169 expected: self.state_hash.clone(),
170 actual: actual_state,
171 });
172 }
173 let actual_content = self.content_hash();
174 if actual_content != self.checkpoint_hash {
175 return Err(CheckpointError::ContentMismatch {
176 expected: self.checkpoint_hash.clone(),
177 actual: actual_content,
178 });
179 }
180 Ok(())
181 }
182
183 pub fn file_name(&self) -> String {
188 format!("{}.checkpoint.json", self.checkpoint_hash)
189 }
190
191 pub fn save(&self, dir: &Path) -> std::io::Result<PathBuf> {
196 fs::create_dir_all(dir)?;
197 let final_path = dir.join(self.file_name());
198 let tmp_path = dir.join(format!("{}.tmp", self.file_name()));
199 {
200 let mut tmp = File::create(&tmp_path)?;
201 tmp.write_all(
202 serde_json::to_string(self)
203 .map_err(std::io::Error::other)?
204 .as_bytes(),
205 )?;
206 tmp.sync_all()?;
207 }
208 fs::rename(&tmp_path, &final_path)?;
209 #[cfg(unix)]
211 {
212 let _ = File::open(dir).and_then(|d| d.sync_all());
213 }
214 Ok(final_path)
215 }
216
217 pub fn load(path: &Path) -> Result<Self, CheckpointError> {
224 let raw = fs::read_to_string(path).map_err(CheckpointError::Io)?;
225 let checkpoint: Checkpoint =
226 serde_json::from_str(&raw).map_err(CheckpointError::Parse)?;
227 checkpoint.verify()?;
228 let expected = checkpoint.file_name();
229 let found = path
230 .file_name()
231 .map(|n| n.to_string_lossy().into_owned())
232 .unwrap_or_default();
233 if found != expected {
234 return Err(CheckpointError::AddressMismatch { expected, found });
235 }
236 Ok(checkpoint)
237 }
238}
239
240#[derive(Debug)]
242pub enum CheckpointError {
243 Io(std::io::Error),
244 Parse(serde_json::Error),
245 HashMismatch { expected: String, actual: String },
248 ContentMismatch { expected: String, actual: String },
252 AddressMismatch { expected: String, found: String },
255}
256
257impl fmt::Display for CheckpointError {
258 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
259 match self {
260 CheckpointError::Io(e) => write!(f, "checkpoint io error: {e}"),
261 CheckpointError::Parse(e) => write!(f, "checkpoint parse error: {e}"),
262 CheckpointError::HashMismatch { expected, actual } => write!(
263 f,
264 "checkpoint state_hash mismatch (stored {expected}, recomputed {actual}) — \
265 tampered or corrupt snapshot, refusing to fold it"
266 ),
267 CheckpointError::ContentMismatch { expected, actual } => write!(
268 f,
269 "checkpoint content-address mismatch (stored {expected}, recomputed {actual}) — \
270 frontier/scopes tampered or record corrupt, refusing to fold or anchor on it"
271 ),
272 CheckpointError::AddressMismatch { expected, found } => write!(
273 f,
274 "checkpoint file name {found} is not its content address {expected} — \
275 renamed or wrongly-stored checkpoint, refusing to load it"
276 ),
277 }
278 }
279}
280
281impl std::error::Error for CheckpointError {}
282
283#[derive(Debug)]
285pub enum AnchorError {
286 Checkpoint(CheckpointError),
288 Chain(ChainError),
290 BrokenAnchor { device_id: String, detail: String },
293 UnanchoredDevice { device_id: String, first_seq: u64 },
296}
297
298impl fmt::Display for AnchorError {
299 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
300 match self {
301 AnchorError::Checkpoint(e) => write!(f, "anchor checkpoint invalid: {e}"),
302 AnchorError::Chain(e) => write!(f, "anchored tail chain invalid: {e}"),
303 AnchorError::BrokenAnchor { device_id, detail } => {
304 write!(f, "device {device_id}: tail does not anchor on checkpoint ({detail})")
305 }
306 AnchorError::UnanchoredDevice { device_id, first_seq } => write!(
307 f,
308 "device {device_id}: first op has seq {first_seq} but the checkpoint \
309 records no frontier for it — unanchored truncation"
310 ),
311 }
312 }
313}
314
315impl std::error::Error for AnchorError {}
316
317pub fn verify_anchored(checkpoint: &Checkpoint, tail: &[OpRecord]) -> Result<(), AnchorError> {
333 checkpoint.verify().map_err(AnchorError::Checkpoint)?;
334 verify_log(tail).map_err(AnchorError::Chain)?;
335
336 let mut first: BTreeMap<&str, &OpRecord> = BTreeMap::new();
338 for op in tail {
339 let replace = match first.get(op.device_id.as_str()) {
340 Some(existing) => op.seq < existing.seq,
341 None => true,
342 };
343 if replace {
344 first.insert(&op.device_id, op);
345 }
346 }
347
348 for (device_id, op) in first {
349 match checkpoint.frontier.get(device_id) {
350 Some(anchor) => {
351 if op.seq != anchor.seq + 1 {
352 return Err(AnchorError::BrokenAnchor {
353 device_id: device_id.to_string(),
354 detail: format!(
355 "first retained seq {} does not continue frontier seq {}",
356 op.seq, anchor.seq
357 ),
358 });
359 }
360 if op.prev.as_deref() != Some(anchor.head.as_str()) {
361 return Err(AnchorError::BrokenAnchor {
362 device_id: device_id.to_string(),
363 detail: format!(
364 "first retained op's prev does not link to frontier head {}",
365 anchor.head
366 ),
367 });
368 }
369 if op.hlc <= anchor.hlc {
370 return Err(AnchorError::BrokenAnchor {
371 device_id: device_id.to_string(),
372 detail: "first retained op's hlc does not advance past the frontier"
373 .to_string(),
374 });
375 }
376 }
377 None => {
378 if op.seq != 0 {
379 return Err(AnchorError::UnanchoredDevice {
380 device_id: device_id.to_string(),
381 first_seq: op.seq,
382 });
383 }
384 }
385 }
386 }
387 Ok(())
388}
389
390pub fn resume_anchored(
402 device_id: impl Into<String>,
403 checkpoint: &Checkpoint,
404 tail: &[OpRecord],
405) -> Result<DeviceLog, AnchorError> {
406 verify_anchored(checkpoint, tail)?;
407 let device_id = device_id.into();
408 let mut log = DeviceLog::new(device_id.clone());
409 for entry in checkpoint.frontier.values() {
410 log.clock.observe(&entry.hlc);
411 }
412 if let Some(anchor) = checkpoint.frontier.get(&device_id) {
413 log.next_seq = anchor.seq + 1;
414 log.prev = Some(anchor.head.clone());
415 }
416 for op in tail {
417 log.clock.observe(&op.hlc);
418 if op.device_id == device_id && op.seq >= log.next_seq {
419 log.next_seq = op.seq + 1;
420 log.prev = Some(op.op_id.clone());
421 }
422 }
423 Ok(log)
424}
425
426#[cfg(test)]
427mod tests {
428 use super::*;
429 use crate::fold::fold_onto;
430 use crate::oplog::{Scope, Surface};
431 use serde_json::json;
432
433 fn ops_with_cut() -> (Vec<OpRecord>, usize) {
436 let mut a = DeviceLog::new("dev-a");
437 let mut b = DeviceLog::new("dev-b");
438 let mut ops = vec![
439 a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f1", "v": 1})),
440 a.append(
441 Scope::Shared { org: "acme".into() },
442 Surface::Declagent,
443 json!({"id": "agent-1", "rev": "a"}),
444 ),
445 ];
446 for op in &ops {
447 b.observe(&op.hlc);
448 }
449 ops.push(b.append(Scope::Personal, Surface::Knowledge, json!({"id": "f2", "v": 2})));
450 let split = ops.len();
451 ops.push(a.append(Scope::Personal, Surface::Conversation, json!({"speaker": "u", "text": "hi", "timestamp": 10})));
452 for op in &ops[split..] {
453 b.observe(&op.hlc);
454 }
455 ops.push(b.append(
456 Scope::Shared { org: "acme".into() },
457 Surface::Declagent,
458 json!({"id": "agent-1", "rev": "b"}),
459 ));
460 (ops, split)
461 }
462
463 #[test]
464 fn from_ops_records_frontier_heads_and_scopes() {
465 let (ops, split) = ops_with_cut();
466 let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
467 assert_eq!(ckpt.scopes, vec!["personal".to_string(), "shared:acme".to_string()]);
468 assert_eq!(ckpt.frontier.len(), 2);
469 assert_eq!(ckpt.frontier["dev-a"].seq, 1);
470 assert_eq!(ckpt.frontier["dev-a"].head, ops[1].op_id);
471 assert_eq!(ckpt.frontier["dev-b"].seq, 0);
472 assert_eq!(ckpt.frontier["dev-b"].head, ops[2].op_id);
473 assert_eq!(ckpt.state_hash, state_hash(&fold(&ops[..split])));
474 }
475
476 #[test]
477 fn from_ops_refuses_an_invalid_log() {
478 let (mut ops, _) = ops_with_cut();
479 ops[0].payload = json!({"forged": true}); assert!(matches!(Checkpoint::from_ops(&ops), Err(ChainError::IdMismatch { .. })));
481 }
482
483 #[test]
484 fn checkpoint_save_load_round_trips_content_addressed() {
485 let dir = tempfile::tempdir().unwrap();
486 let (ops, split) = ops_with_cut();
487 let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
488
489 let path = ckpt.save(dir.path()).unwrap();
490 assert_eq!(
491 path.file_name().unwrap().to_str().unwrap(),
492 format!("{}.checkpoint.json", ckpt.checkpoint_hash),
493 "file name is the WHOLE-RECORD content address"
494 );
495 let loaded = Checkpoint::load(&path).unwrap();
496 assert_eq!(loaded, ckpt);
497
498 let again = ckpt.save(dir.path()).unwrap();
500 assert_eq!(again, path);
501 let names: Vec<String> = fs::read_dir(dir.path())
503 .unwrap()
504 .map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
505 .collect();
506 assert_eq!(names.len(), 1, "only the final checkpoint file exists: {names:?}");
507 }
508
509 #[test]
510 fn tampered_checkpoint_is_rejected_on_load() {
511 let dir = tempfile::tempdir().unwrap();
512 let (ops, split) = ops_with_cut();
513 let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
514 let path = ckpt.save(dir.path()).unwrap();
515
516 let raw = fs::read_to_string(&path).unwrap();
519 let tampered = raw.replace("\"v\":1", "\"v\":999");
520 assert_ne!(raw, tampered, "tamper target must exist in the serialized state");
521 fs::write(&path, tampered).unwrap();
522 assert!(matches!(
523 Checkpoint::load(&path),
524 Err(CheckpointError::HashMismatch { .. })
525 ));
526
527 let mut forged = ckpt.clone();
529 forged.state.logs.clear();
530 assert!(matches!(forged.verify(), Err(CheckpointError::HashMismatch { .. })));
531 }
532
533 #[test]
534 fn tampered_frontier_is_rejected_on_load_and_verify() {
535 let dir = tempfile::tempdir().unwrap();
539 let (ops, split) = ops_with_cut();
540 let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
541 let path = ckpt.save(dir.path()).unwrap();
542
543 let mut value: serde_json::Value =
546 serde_json::from_str(&fs::read_to_string(&path).unwrap()).unwrap();
547 value["frontier"]["dev-a"]["seq"] = json!(7);
548 fs::write(&path, serde_json::to_string(&value).unwrap()).unwrap();
549 assert!(matches!(
550 Checkpoint::load(&path),
551 Err(CheckpointError::ContentMismatch { .. })
552 ));
553
554 let mut forged = ckpt.clone();
557 forged.frontier.get_mut("dev-a").unwrap().seq = 7;
558 assert!(matches!(forged.verify(), Err(CheckpointError::ContentMismatch { .. })));
559 assert!(matches!(
560 verify_anchored(&forged, &ops[split..]),
561 Err(AnchorError::Checkpoint(CheckpointError::ContentMismatch { .. }))
562 ));
563
564 let good = ckpt.save(dir.path()).unwrap();
566 let renamed = dir.path().join("latest.checkpoint.json");
567 fs::rename(&good, &renamed).unwrap();
568 assert!(matches!(
569 Checkpoint::load(&renamed),
570 Err(CheckpointError::AddressMismatch { .. })
571 ));
572 }
573
574 #[test]
575 fn different_frontiers_never_share_a_content_address() {
576 let mut a = DeviceLog::new("a");
588 let mut b = DeviceLog::new("b");
589 let fact = json!({"id": "f1", "body": "hi"});
590 let oa = a.append(Scope::Personal, Surface::Knowledge, fact.clone());
591 let ob = b.append(Scope::Personal, Surface::Knowledge, fact);
592
593 let just_a = Checkpoint::from_ops(std::slice::from_ref(&oa)).unwrap();
594 let both = Checkpoint::from_ops(&[oa, ob]).unwrap();
595
596 assert_eq!(just_a.state, both.state, "cross-device dedup: identical folded state");
597 assert_eq!(just_a.state_hash, both.state_hash, "state hash agrees (divergence invariant)");
598 assert_ne!(just_a.frontier, both.frontier, "but the frontiers differ");
599 assert_ne!(
600 just_a.checkpoint_hash, both.checkpoint_hash,
601 "so the content addresses MUST differ"
602 );
603 assert_ne!(just_a.file_name(), both.file_name(), "…and so must the file names");
604
605 let dir = tempfile::tempdir().unwrap();
607 let p1 = just_a.save(dir.path()).unwrap();
608 let p2 = both.save(dir.path()).unwrap();
609 assert_ne!(p1, p2);
610 assert_eq!(Checkpoint::load(&p1).unwrap(), just_a);
611 assert_eq!(Checkpoint::load(&p2).unwrap(), both);
612 }
613
614 #[test]
615 fn verify_anchored_accepts_the_truncated_composition() {
616 let (ops, split) = ops_with_cut();
617 let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
618 let tail = &ops[split..];
619 verify_log(tail).expect("verify_log alone accepts a non-zero-seq chain");
620 verify_anchored(&ckpt, tail).expect("checkpoint anchors the truncated tail");
621 assert_eq!(fold_onto(&ckpt.state, tail), fold(&ops));
623 }
624
625 #[test]
626 fn verify_anchored_rejects_breaks() {
627 let (ops, split) = ops_with_cut();
628 let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
629 let tail: Vec<OpRecord> = ops[split..].to_vec();
630
631 let mut a_extra = resume_anchored("dev-a", &ckpt, &tail).unwrap();
638 let extra = a_extra.append(Scope::Personal, Surface::Knowledge, json!({"id": "f9"}));
639 let mut with_extra = tail.clone();
640 with_extra.push(extra.clone());
641 verify_anchored(&ckpt, &with_extra).unwrap();
642
643 let holed: Vec<OpRecord> = with_extra
648 .iter()
649 .filter(|o| o.op_id != tail[0].op_id)
650 .cloned()
651 .collect();
652 assert!(matches!(
653 verify_anchored(&ckpt, &holed),
654 Err(AnchorError::BrokenAnchor { .. })
655 ));
656
657 let mut stranger = DeviceLog::new("dev-c");
659 stranger.append(Scope::Personal, Surface::Knowledge, json!({"id": "s0"}));
660 let s1 = stranger.append(Scope::Personal, Surface::Knowledge, json!({"id": "s1"}));
661 let mut with_stranger = tail.clone();
662 with_stranger.push(s1); assert!(matches!(
664 verify_anchored(&ckpt, &with_stranger),
665 Err(AnchorError::UnanchoredDevice { first_seq: 1, .. })
666 ));
667
668 let mut forged = ckpt.clone();
670 forged.state.logs.clear();
671 assert!(matches!(
672 verify_anchored(&forged, &tail),
673 Err(AnchorError::Checkpoint(CheckpointError::HashMismatch { .. }))
674 ));
675 }
676
677 #[test]
678 fn resume_anchored_continues_chains_after_truncation() {
679 let (ops, split) = ops_with_cut();
680 let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
681 let tail: Vec<OpRecord> = ops[split..].to_vec();
682
683 let mut a = resume_anchored("dev-a", &ckpt, &tail).unwrap();
685 let next_a = a.append(Scope::Personal, Surface::Knowledge, json!({"id": "na"}));
686 assert_eq!(next_a.seq, 3);
687 assert_eq!(next_a.prev.as_deref(), Some(tail[0].op_id.as_str()));
688
689 let ckpt_all = Checkpoint::from_ops(&ops).unwrap();
692 let mut b = resume_anchored("dev-b", &ckpt_all, &[]).unwrap();
693 let next_b = b.append(Scope::Personal, Surface::Knowledge, json!({"id": "nb"}));
694 assert_eq!(next_b.seq, 2, "continues past the checkpointed chain");
695 assert_eq!(next_b.prev.as_deref(), Some(ckpt_all.frontier["dev-b"].head.as_str()));
696 assert!(
697 next_b.hlc > ckpt_all.frontier["dev-a"].hlc && next_b.hlc > ckpt_all.frontier["dev-b"].hlc,
698 "lamport advanced past everything the checkpoint covers"
699 );
700
701 let mut composed = tail.clone();
703 composed.push(next_a);
704 verify_anchored(&ckpt, &composed).unwrap();
705 }
706}