OXYGENHUB
# ec2cfb379e70 oxygen / crates / oxy-core / src / repo / checkpoint.rs
5.0 KiB · Rust · last changed Checkpoints: serialized states per bookmark, automatic every core.checkpoint-interval changes, oxy checkpoint, checkpoint-accelerated load_state; states keep change metadata only 1 month ago source blame history raw
1
//! Checkpoints (spec §11): a serialized state kept per bookmark so loading
2
//! a state costs the changes since the checkpoint rather than all of them.
3
//!
4
//! The newest checkpoint of each bookmark is pointed at by
5
//! `.oxy/checkpoints/<bookmark>`; older ones become unreachable and are
6
//! garbage collected. A checkpoint applies to a requested head set when all
7
//! of its heads lie in that set's closure, which the loader establishes while
8
//! walking from the requested heads and stopping at checkpoint members.
9
10
use super::*;
11
use crate::objects::Checkpoint;
12
13
impl Repo {
14
    fn checkpoint_pointer(&self, bookmark: &str) -> PathBuf {
15
        self.oxy.join("checkpoints").join(bookmark)
16
    }
17
18
    /// The newest checkpoint recorded for a bookmark.
19
    pub fn latest_checkpoint(&self, bookmark: &str) -> Result<Option<(ObjectRef, Checkpoint)>> {
20
        match fs::read_to_string(self.checkpoint_pointer(bookmark)) {
21
            Ok(h) => {
22
                let r = ObjectRef::from_hex(h.trim()).ok_or_else(|| RepoError::Msg("corrupt checkpoint pointer".into()))?;
23
                match self.store.get_typed(&r, ObjectType::Checkpoint) {
24
                    Ok(bytes) => Ok(Some((r, Checkpoint::decode(&bytes)?))),
25
                    Err(StoreError::NotFound(_)) => Ok(None),
26
                    Err(e) => Err(e.into()),
27
                }
28
            }
29
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
30
            Err(e) => Err(e.into()),
31
        }
32
    }
33
34
    /// Every checkpoint pointer, for garbage collection roots.
35
    pub(crate) fn checkpoint_refs(&self) -> Result<Vec<ObjectRef>> {
36
        let dir = self.oxy.join("checkpoints");
37
        let mut out = Vec::new();
38
        if let Ok(rd) = fs::read_dir(&dir) {
39
            for e in rd.flatten() {
40
                if let Ok(h) = fs::read_to_string(e.path()) {
41
                    if let Some(r) = ObjectRef::from_hex(h.trim()) {
42
                        out.push(r);
43
                    }
44
                }
45
            }
46
        }
47
        Ok(out)
48
    }
49
50
    /// Write a checkpoint of `state` (whose heads are `heads`) for `bookmark`.
51
    pub(crate) fn write_checkpoint(&self, bookmark: &str, state: &State, heads: &[ChangeHash]) -> Result<(ObjectRef, usize)> {
52
        let payload = state.checkpoint_payload();
53
        let blob = self.store_blob(&payload)?;
54
        let cp = Checkpoint { heads: StateObject::new(heads.to_vec()).heads, count: state.len() as u64, payload: blob };
55
        let r = self.store.put(ObjectType::Checkpoint, &cp.encode())?;
56
        fs::create_dir_all(self.oxy.join("checkpoints"))?;
57
        write_atomic(&self.checkpoint_pointer(bookmark), r.to_hex().as_bytes())?;
58
        Ok((r, payload.len()))
59
    }
60
61
    /// Checkpoint the current bookmark's state now.
62
    pub fn checkpoint(&self) -> Result<(ObjectRef, usize, usize)> {
63
        let heads = self.bookmark_heads()?;
64
        let state = self.load_state_full(&heads)?;
65
        let (r, bytes) = self.write_checkpoint(&self.config.bookmark, &state, &heads)?;
66
        Ok((r, state.len(), bytes))
67
    }
68
69
    /// Load the state of `heads` using a checkpoint when one applies.
70
    pub(crate) fn load_state_checkpointed(&self, heads: &[ChangeHash]) -> Result<Option<State>> {
71
        let mut bookmarks: Vec<String> = vec![self.config.bookmark.clone()];
72
        if let Ok(rd) = fs::read_dir(self.oxy.join("checkpoints")) {
73
            for e in rd.flatten() {
74
                let name = e.file_name().to_string_lossy().to_string();
75
                if !bookmarks.contains(&name) {
76
                    bookmarks.push(name);
77
                }
78
            }
79
        }
80
        for b in bookmarks {
81
            let Some((_, cp)) = self.latest_checkpoint(&b)? else { continue };
82
            // Cheap pre-check: every checkpoint head must be stored.
83
            if cp.heads.iter().any(|h| !self.store.has(h)) {
84
                continue;
85
            }
86
            let payload = self.read_blob(&cp.payload)?;
87
            let base = State::from_checkpoint_payload(&payload)?;
88
            let cp_heads: BTreeSet<ChangeHash> = cp.heads.iter().copied().collect();
89
            let mut found: BTreeSet<ChangeHash> = BTreeSet::new();
90
            let mut delta: Vec<Change> = Vec::new();
91
            let mut seen: BTreeSet<ChangeHash> = BTreeSet::new();
92
            let mut stack: Vec<ChangeHash> = heads.to_vec();
93
            while let Some(h) = stack.pop() {
94
                if !seen.insert(h) {
95
                    continue;
96
                }
97
                if base.contains(&h) {
98
                    if cp_heads.contains(&h) {
99
                        found.insert(h);
100
                    }
101
                    continue;
102
                }
103
                let c = self.store.get_change(&h)?;
104
                stack.extend(c.deps.iter().copied());
105
                delta.push(c);
106
            }
107
            if found.len() != cp_heads.len() {
108
                continue;
109
            }
110
            let mut state = base;
111
            for c in delta {
112
                state.add_unchecked(c);
113
            }
114
            state.validate()?;
115
            return Ok(Some(state));
116
        }
117
        Ok(None)
118
    }
119
}