OXYGENHUB
# b2ef93bb2366 oxygen / crates / oxy-core / src / repo / checkpoint.rs
6.8 KiB · Rust · last changed Shallow clone: clone --depth, checkpoint and shallow endpoints, boundary-aware import and negotiation, oxy deepen; log from state metadata (spec 0.6g) 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
    /// Point `bookmark` at an already stored checkpoint (a shallow clone's base).
70
    pub(crate) fn install_checkpoint(&self, bookmark: &str, r: &ObjectRef) -> Result<()> {
71
        Checkpoint::decode(&self.store.get_typed(r, ObjectType::Checkpoint)?)?;
72
        fs::create_dir_all(self.oxy.join("checkpoints"))?;
73
        write_atomic(&self.checkpoint_pointer(bookmark), r.to_hex().as_bytes())
74
    }
75
76
    /// A bundle holding a checkpoint of exactly `heads` with its payload
77
    /// (spec §16.1 `checkpoint`); the current bookmark checkpoint is reused
78
    /// when it matches, otherwise one is made and, when `heads` are the
79
    /// bookmark's, kept.
80
    pub fn checkpoint_bundle_for(&self, heads: &[ChangeHash]) -> Result<Bundle> {
81
        let wanted = StateObject::new(heads.to_vec()).heads;
82
        let bookmark = self.config.bookmark.clone();
83
        let current = self.bookmark_heads()?;
84
        let r = match self.latest_checkpoint(&bookmark)? {
85
            Some((r, cp)) if cp.heads == wanted => r,
86
            _ => {
87
                let state = self.load_state(&wanted)?;
88
                if current == wanted {
89
                    self.write_checkpoint(&bookmark, &state, &wanted)?.0
90
                } else {
91
                    let payload = state.checkpoint_payload();
92
                    let blob = self.store_blob(&payload)?;
93
                    let cp = Checkpoint { heads: wanted.clone(), count: state.len() as u64, payload: blob };
94
                    self.store.put(ObjectType::Checkpoint, &cp.encode())?
95
                }
96
            }
97
        };
98
        let cp = Checkpoint::decode(&self.store.get_typed(&r, ObjectType::Checkpoint)?)?;
99
        let mut bundle = Bundle::default();
100
        for c in Blob::decode(&self.store.get_typed(&cp.payload, ObjectType::Blob)?)?.chunks {
101
            bundle.objects.push(self.store.get(&c)?);
102
        }
103
        bundle.objects.push(self.store.get(&cp.payload)?);
104
        bundle.objects.push((ObjectType::Checkpoint, cp.encode()));
105
        Ok(bundle)
106
    }
107
108
    /// Load the state of `heads` using a checkpoint when one applies.
109
    pub(crate) fn load_state_checkpointed(&self, heads: &[ChangeHash]) -> Result<Option<State>> {
110
        let mut bookmarks: Vec<String> = vec![self.config.bookmark.clone()];
111
        if let Ok(rd) = fs::read_dir(self.oxy.join("checkpoints")) {
112
            for e in rd.flatten() {
113
                let name = e.file_name().to_string_lossy().to_string();
114
                if !bookmarks.contains(&name) {
115
                    bookmarks.push(name);
116
                }
117
            }
118
        }
119
        for b in bookmarks {
120
            let Some((_, cp)) = self.latest_checkpoint(&b)? else { continue };
121
            let payload = self.read_blob(&cp.payload)?;
122
            let base = State::from_checkpoint_payload(&payload)?;
123
            let cp_heads: BTreeSet<ChangeHash> = cp.heads.iter().copied().collect();
124
            let mut found: BTreeSet<ChangeHash> = BTreeSet::new();
125
            let mut delta: Vec<Change> = Vec::new();
126
            let mut seen: BTreeSet<ChangeHash> = BTreeSet::new();
127
            let mut stack: Vec<ChangeHash> = heads.to_vec();
128
            while let Some(h) = stack.pop() {
129
                if !seen.insert(h) {
130
                    continue;
131
                }
132
                if base.contains(&h) {
133
                    if cp_heads.contains(&h) {
134
                        found.insert(h);
135
                    }
136
                    continue;
137
                }
138
                let c = self.store.get_change(&h)?;
139
                stack.extend(c.deps.iter().copied());
140
                delta.push(c);
141
            }
142
            if found.len() != cp_heads.len() {
143
                continue;
144
            }
145
            let mut state = base;
146
            for c in delta {
147
                state.add_unchecked(c);
148
            }
149
            state.validate()?;
150
            return Ok(Some(state));
151
        }
152
        Ok(None)
153
    }
154
}