use crate::fold::{fold, state_hash, SyncState};
use crate::oplog::{canonical_json, verify_log, ChainError, DeviceLog, Hlc, OpRecord};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::fs::{self, File};
use std::io::Write;
use std::path::{Path, PathBuf};
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FrontierEntry {
pub seq: u64,
pub hlc: Hlc,
pub head: String,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Checkpoint {
pub frontier: BTreeMap<String, FrontierEntry>,
pub scopes: Vec<String>,
pub state_hash: String,
pub checkpoint_hash: String,
pub state: SyncState,
}
impl Checkpoint {
pub fn assemble(
frontier: BTreeMap<String, FrontierEntry>,
scopes: Vec<String>,
state: SyncState,
) -> Self {
let hash = state_hash(&state);
let mut checkpoint = Self {
frontier,
scopes,
state_hash: hash,
checkpoint_hash: String::new(),
state,
};
checkpoint.checkpoint_hash = checkpoint.content_hash();
checkpoint
}
pub fn from_ops(ops: &[OpRecord]) -> Result<Self, ChainError> {
verify_log(ops)?;
let state = fold(ops);
let mut frontier: BTreeMap<String, FrontierEntry> = BTreeMap::new();
let mut scopes: BTreeSet<String> = BTreeSet::new();
for op in ops {
scopes.insert(op.scope.tag());
let replace = match frontier.get(&op.device_id) {
Some(existing) => op.seq > existing.seq,
None => true,
};
if replace {
frontier.insert(
op.device_id.clone(),
FrontierEntry {
seq: op.seq,
hlc: op.hlc.clone(),
head: op.op_id.clone(),
},
);
}
}
Ok(Self::assemble(
frontier,
scopes.into_iter().collect(),
state,
))
}
pub fn content_hash(&self) -> String {
let mut value = serde_json::to_value(self).expect("Checkpoint serializes");
if let Some(obj) = value.as_object_mut() {
obj.remove("checkpoint_hash");
}
let mut hasher = Sha256::new();
hasher.update(canonical_json(&value).as_bytes());
let digest = hasher.finalize();
let hex: String = digest.iter().take(16).map(|b| format!("{b:02x}")).collect();
format!("ckpt-{hex}")
}
pub fn verify(&self) -> Result<(), CheckpointError> {
let actual_state = state_hash(&self.state);
if actual_state != self.state_hash {
return Err(CheckpointError::HashMismatch {
expected: self.state_hash.clone(),
actual: actual_state,
});
}
let actual_content = self.content_hash();
if actual_content != self.checkpoint_hash {
return Err(CheckpointError::ContentMismatch {
expected: self.checkpoint_hash.clone(),
actual: actual_content,
});
}
Ok(())
}
pub fn file_name(&self) -> String {
format!("{}.checkpoint.json", self.checkpoint_hash)
}
pub fn save(&self, dir: &Path) -> std::io::Result<PathBuf> {
fs::create_dir_all(dir)?;
let final_path = dir.join(self.file_name());
let tmp_path = dir.join(format!("{}.tmp", self.file_name()));
{
let mut tmp = File::create(&tmp_path)?;
tmp.write_all(
serde_json::to_string(self)
.map_err(std::io::Error::other)?
.as_bytes(),
)?;
tmp.sync_all()?;
}
fs::rename(&tmp_path, &final_path)?;
#[cfg(unix)]
{
let _ = File::open(dir).and_then(|d| d.sync_all());
}
Ok(final_path)
}
pub fn load(path: &Path) -> Result<Self, CheckpointError> {
let raw = fs::read_to_string(path).map_err(CheckpointError::Io)?;
let checkpoint: Checkpoint = serde_json::from_str(&raw).map_err(CheckpointError::Parse)?;
checkpoint.verify()?;
let expected = checkpoint.file_name();
let found = path
.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_default();
if found != expected {
return Err(CheckpointError::AddressMismatch { expected, found });
}
Ok(checkpoint)
}
}
#[derive(Debug)]
pub enum CheckpointError {
Io(std::io::Error),
Parse(serde_json::Error),
HashMismatch {
expected: String,
actual: String,
},
ContentMismatch {
expected: String,
actual: String,
},
AddressMismatch {
expected: String,
found: String,
},
}
impl fmt::Display for CheckpointError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
CheckpointError::Io(e) => write!(f, "checkpoint io error: {e}"),
CheckpointError::Parse(e) => write!(f, "checkpoint parse error: {e}"),
CheckpointError::HashMismatch { expected, actual } => write!(
f,
"checkpoint state_hash mismatch (stored {expected}, recomputed {actual}) — \
tampered or corrupt snapshot, refusing to fold it"
),
CheckpointError::ContentMismatch { expected, actual } => write!(
f,
"checkpoint content-address mismatch (stored {expected}, recomputed {actual}) — \
frontier/scopes tampered or record corrupt, refusing to fold or anchor on it"
),
CheckpointError::AddressMismatch { expected, found } => write!(
f,
"checkpoint file name {found} is not its content address {expected} — \
renamed or wrongly-stored checkpoint, refusing to load it"
),
}
}
}
impl std::error::Error for CheckpointError {}
#[derive(Debug)]
pub enum AnchorError {
Checkpoint(CheckpointError),
Chain(ChainError),
BrokenAnchor { device_id: String, detail: String },
UnanchoredDevice { device_id: String, first_seq: u64 },
}
impl fmt::Display for AnchorError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
AnchorError::Checkpoint(e) => write!(f, "anchor checkpoint invalid: {e}"),
AnchorError::Chain(e) => write!(f, "anchored tail chain invalid: {e}"),
AnchorError::BrokenAnchor { device_id, detail } => {
write!(
f,
"device {device_id}: tail does not anchor on checkpoint ({detail})"
)
}
AnchorError::UnanchoredDevice {
device_id,
first_seq,
} => write!(
f,
"device {device_id}: first op has seq {first_seq} but the checkpoint \
records no frontier for it — unanchored truncation"
),
}
}
}
impl std::error::Error for AnchorError {}
pub fn verify_anchored(checkpoint: &Checkpoint, tail: &[OpRecord]) -> Result<(), AnchorError> {
checkpoint.verify().map_err(AnchorError::Checkpoint)?;
verify_log(tail).map_err(AnchorError::Chain)?;
let mut first: BTreeMap<&str, &OpRecord> = BTreeMap::new();
for op in tail {
let replace = match first.get(op.device_id.as_str()) {
Some(existing) => op.seq < existing.seq,
None => true,
};
if replace {
first.insert(&op.device_id, op);
}
}
for (device_id, op) in first {
match checkpoint.frontier.get(device_id) {
Some(anchor) => {
if op.seq != anchor.seq + 1 {
return Err(AnchorError::BrokenAnchor {
device_id: device_id.to_string(),
detail: format!(
"first retained seq {} does not continue frontier seq {}",
op.seq, anchor.seq
),
});
}
if op.prev.as_deref() != Some(anchor.head.as_str()) {
return Err(AnchorError::BrokenAnchor {
device_id: device_id.to_string(),
detail: format!(
"first retained op's prev does not link to frontier head {}",
anchor.head
),
});
}
if op.hlc <= anchor.hlc {
return Err(AnchorError::BrokenAnchor {
device_id: device_id.to_string(),
detail: "first retained op's hlc does not advance past the frontier"
.to_string(),
});
}
}
None => {
if op.seq != 0 {
return Err(AnchorError::UnanchoredDevice {
device_id: device_id.to_string(),
first_seq: op.seq,
});
}
}
}
}
Ok(())
}
pub fn resume_anchored(
device_id: impl Into<String>,
checkpoint: &Checkpoint,
tail: &[OpRecord],
) -> Result<DeviceLog, AnchorError> {
verify_anchored(checkpoint, tail)?;
let device_id = device_id.into();
let mut log = DeviceLog::new(device_id.clone());
for entry in checkpoint.frontier.values() {
log.clock.observe(&entry.hlc);
}
if let Some(anchor) = checkpoint.frontier.get(&device_id) {
log.next_seq = anchor.seq + 1;
log.prev = Some(anchor.head.clone());
}
for op in tail {
log.clock.observe(&op.hlc);
if op.device_id == device_id && op.seq >= log.next_seq {
log.next_seq = op.seq + 1;
log.prev = Some(op.op_id.clone());
}
}
Ok(log)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::fold::fold_onto;
use crate::oplog::{Scope, Surface};
use serde_json::json;
fn ops_with_cut() -> (Vec<OpRecord>, usize) {
let mut a = DeviceLog::new("dev-a");
let mut b = DeviceLog::new("dev-b");
let mut ops = vec![
a.append(
Scope::Personal,
Surface::Knowledge,
json!({"id": "f1", "v": 1}),
),
a.append(
Scope::Shared { org: "acme".into() },
Surface::Declagent,
json!({"id": "agent-1", "rev": "a"}),
),
];
for op in &ops {
b.observe(&op.hlc);
}
ops.push(b.append(
Scope::Personal,
Surface::Knowledge,
json!({"id": "f2", "v": 2}),
));
let split = ops.len();
ops.push(a.append(
Scope::Personal,
Surface::Conversation,
json!({"speaker": "u", "text": "hi", "timestamp": 10}),
));
for op in &ops[split..] {
b.observe(&op.hlc);
}
ops.push(b.append(
Scope::Shared { org: "acme".into() },
Surface::Declagent,
json!({"id": "agent-1", "rev": "b"}),
));
(ops, split)
}
#[test]
fn from_ops_records_frontier_heads_and_scopes() {
let (ops, split) = ops_with_cut();
let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
assert_eq!(
ckpt.scopes,
vec!["personal".to_string(), "shared:acme".to_string()]
);
assert_eq!(ckpt.frontier.len(), 2);
assert_eq!(ckpt.frontier["dev-a"].seq, 1);
assert_eq!(ckpt.frontier["dev-a"].head, ops[1].op_id);
assert_eq!(ckpt.frontier["dev-b"].seq, 0);
assert_eq!(ckpt.frontier["dev-b"].head, ops[2].op_id);
assert_eq!(ckpt.state_hash, state_hash(&fold(&ops[..split])));
}
#[test]
fn from_ops_refuses_an_invalid_log() {
let (mut ops, _) = ops_with_cut();
ops[0].payload = json!({"forged": true}); assert!(matches!(
Checkpoint::from_ops(&ops),
Err(ChainError::IdMismatch { .. })
));
}
#[test]
fn checkpoint_save_load_round_trips_content_addressed() {
let dir = tempfile::tempdir().unwrap();
let (ops, split) = ops_with_cut();
let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
let path = ckpt.save(dir.path()).unwrap();
assert_eq!(
path.file_name().unwrap().to_str().unwrap(),
format!("{}.checkpoint.json", ckpt.checkpoint_hash),
"file name is the WHOLE-RECORD content address"
);
let loaded = Checkpoint::load(&path).unwrap();
assert_eq!(loaded, ckpt);
let again = ckpt.save(dir.path()).unwrap();
assert_eq!(again, path);
let names: Vec<String> = fs::read_dir(dir.path())
.unwrap()
.map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
.collect();
assert_eq!(
names.len(),
1,
"only the final checkpoint file exists: {names:?}"
);
}
#[test]
fn tampered_checkpoint_is_rejected_on_load() {
let dir = tempfile::tempdir().unwrap();
let (ops, split) = ops_with_cut();
let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
let path = ckpt.save(dir.path()).unwrap();
let raw = fs::read_to_string(&path).unwrap();
let tampered = raw.replace("\"v\":1", "\"v\":999");
assert_ne!(
raw, tampered,
"tamper target must exist in the serialized state"
);
fs::write(&path, tampered).unwrap();
assert!(matches!(
Checkpoint::load(&path),
Err(CheckpointError::HashMismatch { .. })
));
let mut forged = ckpt.clone();
forged.state.logs.clear();
assert!(matches!(
forged.verify(),
Err(CheckpointError::HashMismatch { .. })
));
}
#[test]
fn tampered_frontier_is_rejected_on_load_and_verify() {
let dir = tempfile::tempdir().unwrap();
let (ops, split) = ops_with_cut();
let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
let path = ckpt.save(dir.path()).unwrap();
let mut value: serde_json::Value =
serde_json::from_str(&fs::read_to_string(&path).unwrap()).unwrap();
value["frontier"]["dev-a"]["seq"] = json!(7);
fs::write(&path, serde_json::to_string(&value).unwrap()).unwrap();
assert!(matches!(
Checkpoint::load(&path),
Err(CheckpointError::ContentMismatch { .. })
));
let mut forged = ckpt.clone();
forged.frontier.get_mut("dev-a").unwrap().seq = 7;
assert!(matches!(
forged.verify(),
Err(CheckpointError::ContentMismatch { .. })
));
assert!(matches!(
verify_anchored(&forged, &ops[split..]),
Err(AnchorError::Checkpoint(
CheckpointError::ContentMismatch { .. }
))
));
let good = ckpt.save(dir.path()).unwrap();
let renamed = dir.path().join("latest.checkpoint.json");
fs::rename(&good, &renamed).unwrap();
assert!(matches!(
Checkpoint::load(&renamed),
Err(CheckpointError::AddressMismatch { .. })
));
}
#[test]
fn different_frontiers_never_share_a_content_address() {
let mut a = DeviceLog::new("a");
let mut b = DeviceLog::new("b");
let fact = json!({"id": "f1", "body": "hi"});
let oa = a.append(Scope::Personal, Surface::Knowledge, fact.clone());
let ob = b.append(Scope::Personal, Surface::Knowledge, fact);
let just_a = Checkpoint::from_ops(std::slice::from_ref(&oa)).unwrap();
let both = Checkpoint::from_ops(&[oa, ob]).unwrap();
assert_eq!(
just_a.state, both.state,
"cross-device dedup: identical folded state"
);
assert_eq!(
just_a.state_hash, both.state_hash,
"state hash agrees (divergence invariant)"
);
assert_ne!(just_a.frontier, both.frontier, "but the frontiers differ");
assert_ne!(
just_a.checkpoint_hash, both.checkpoint_hash,
"so the content addresses MUST differ"
);
assert_ne!(
just_a.file_name(),
both.file_name(),
"…and so must the file names"
);
let dir = tempfile::tempdir().unwrap();
let p1 = just_a.save(dir.path()).unwrap();
let p2 = both.save(dir.path()).unwrap();
assert_ne!(p1, p2);
assert_eq!(Checkpoint::load(&p1).unwrap(), just_a);
assert_eq!(Checkpoint::load(&p2).unwrap(), both);
}
#[test]
fn verify_anchored_accepts_the_truncated_composition() {
let (ops, split) = ops_with_cut();
let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
let tail = &ops[split..];
verify_log(tail).expect("verify_log alone accepts a non-zero-seq chain");
verify_anchored(&ckpt, tail).expect("checkpoint anchors the truncated tail");
assert_eq!(fold_onto(&ckpt.state, tail), fold(&ops));
}
#[test]
fn verify_anchored_rejects_breaks() {
let (ops, split) = ops_with_cut();
let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
let tail: Vec<OpRecord> = ops[split..].to_vec();
let mut a_extra = resume_anchored("dev-a", &ckpt, &tail).unwrap();
let extra = a_extra.append(Scope::Personal, Surface::Knowledge, json!({"id": "f9"}));
let mut with_extra = tail.clone();
with_extra.push(extra.clone());
verify_anchored(&ckpt, &with_extra).unwrap();
let holed: Vec<OpRecord> = with_extra
.iter()
.filter(|o| o.op_id != tail[0].op_id)
.cloned()
.collect();
assert!(matches!(
verify_anchored(&ckpt, &holed),
Err(AnchorError::BrokenAnchor { .. })
));
let mut stranger = DeviceLog::new("dev-c");
stranger.append(Scope::Personal, Surface::Knowledge, json!({"id": "s0"}));
let s1 = stranger.append(Scope::Personal, Surface::Knowledge, json!({"id": "s1"}));
let mut with_stranger = tail.clone();
with_stranger.push(s1); assert!(matches!(
verify_anchored(&ckpt, &with_stranger),
Err(AnchorError::UnanchoredDevice { first_seq: 1, .. })
));
let mut forged = ckpt.clone();
forged.state.logs.clear();
assert!(matches!(
verify_anchored(&forged, &tail),
Err(AnchorError::Checkpoint(
CheckpointError::HashMismatch { .. }
))
));
}
#[test]
fn resume_anchored_continues_chains_after_truncation() {
let (ops, split) = ops_with_cut();
let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
let tail: Vec<OpRecord> = ops[split..].to_vec();
let mut a = resume_anchored("dev-a", &ckpt, &tail).unwrap();
let next_a = a.append(Scope::Personal, Surface::Knowledge, json!({"id": "na"}));
assert_eq!(next_a.seq, 3);
assert_eq!(next_a.prev.as_deref(), Some(tail[0].op_id.as_str()));
let ckpt_all = Checkpoint::from_ops(&ops).unwrap();
let mut b = resume_anchored("dev-b", &ckpt_all, &[]).unwrap();
let next_b = b.append(Scope::Personal, Surface::Knowledge, json!({"id": "nb"}));
assert_eq!(next_b.seq, 2, "continues past the checkpointed chain");
assert_eq!(
next_b.prev.as_deref(),
Some(ckpt_all.frontier["dev-b"].head.as_str())
);
assert!(
next_b.hlc > ckpt_all.frontier["dev-a"].hlc
&& next_b.hlc > ckpt_all.frontier["dev-b"].hlc,
"lamport advanced past everything the checkpoint covers"
);
let mut composed = tail.clone();
composed.push(next_a);
verify_anchored(&ckpt, &composed).unwrap();
}
}