OXYGENHUB
# 9054759cfbd1 oxygen / crates / oxy-core / src / repo / checkpoint.rs
7.1 KiB · Rust · last changed Unbound hosted repositories adopt the first pusher's id 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())
23
                    .ok_or_else(|| RepoError::Msg("corrupt checkpoint pointer".into()))?;
24
                match self.store.get_typed(&r, ObjectType::Checkpoint) {
25
                    Ok(bytes) => Ok(Some((r, Checkpoint::decode(&bytes)?))),
26
                    Err(StoreError::NotFound(_)) => Ok(None),
27
                    Err(e) => Err(e.into()),
28
                }
29
            }
30
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
31
            Err(e) => Err(e.into()),
32
        }
33
    }
34
35
    /// Every checkpoint pointer, for garbage collection roots.
36
    pub(crate) fn checkpoint_refs(&self) -> Result<Vec<ObjectRef>> {
37
        let dir = self.oxy.join("checkpoints");
38
        let mut out = Vec::new();
39
        if let Ok(rd) = fs::read_dir(&dir) {
40
            for e in rd.flatten() {
41
                if let Ok(h) = fs::read_to_string(e.path()) {
42
                    if let Some(r) = ObjectRef::from_hex(h.trim()) {
43
                        out.push(r);
44
                    }
45
                }
46
            }
47
        }
48
        Ok(out)
49
    }
50
51
    /// Write a checkpoint of `state` (whose heads are `heads`) for `bookmark`.
52
    pub(crate) fn write_checkpoint(
53
        &self,
54
        bookmark: &str,
55
        state: &State,
56
        heads: &[ChangeHash],
57
    ) -> Result<(ObjectRef, usize)> {
58
        let payload = state.checkpoint_payload();
59
        let batch = self.store.begin_batch()?;
60
        let blob = self.store_blob(&payload)?;
61
        let cp = Checkpoint {
62
            heads: StateObject::new(heads.to_vec()).heads,
63
            count: state.len() as u64,
64
            payload: blob,
65
        };
66
        let r = self.store.put(ObjectType::Checkpoint, &cp.encode())?;
67
        batch.finish()?;
68
        fs::create_dir_all(self.oxy.join("checkpoints"))?;
69
        write_atomic(&self.checkpoint_pointer(bookmark), r.to_hex().as_bytes())?;
70
        Ok((r, payload.len()))
71
    }
72
73
    /// Checkpoint the current bookmark's state now.
74
    pub fn checkpoint(&self) -> Result<(ObjectRef, usize, usize)> {
75
        let heads = self.bookmark_heads()?;
76
        let state = self.load_state_full(&heads)?;
77
        let (r, bytes) = self.write_checkpoint(&self.config.bookmark, &state, &heads)?;
78
        Ok((r, state.len(), bytes))
79
    }
80
81
    /// Point `bookmark` at an already stored checkpoint (a shallow clone's base).
82
    pub(crate) fn install_checkpoint(&self, bookmark: &str, r: &ObjectRef) -> Result<()> {
83
        Checkpoint::decode(&self.store.get_typed(r, ObjectType::Checkpoint)?)?;
84
        fs::create_dir_all(self.oxy.join("checkpoints"))?;
85
        write_atomic(&self.checkpoint_pointer(bookmark), r.to_hex().as_bytes())
86
    }
87
88
    /// A bundle holding a checkpoint of exactly `heads` with its payload
89
    /// (spec §16.1 `checkpoint`); the current bookmark checkpoint is reused
90
    /// when it matches, otherwise one is made and, when `heads` are the
91
    /// bookmark's, kept.
92
    pub fn checkpoint_bundle_for(&self, heads: &[ChangeHash]) -> Result<Bundle> {
93
        let wanted = StateObject::new(heads.to_vec()).heads;
94
        let bookmark = self.config.bookmark.clone();
95
        let current = self.bookmark_heads()?;
96
        let r = match self.latest_checkpoint(&bookmark)? {
97
            Some((r, cp)) if cp.heads == wanted => r,
98
            _ => {
99
                let state = self.load_state(&wanted)?;
100
                if current == wanted {
101
                    self.write_checkpoint(&bookmark, &state, &wanted)?.0
102
                } else {
103
                    let payload = state.checkpoint_payload();
104
                    let blob = self.store_blob(&payload)?;
105
                    let cp = Checkpoint {
106
                        heads: wanted.clone(),
107
                        count: state.len() as u64,
108
                        payload: blob,
109
                    };
110
                    self.store.put(ObjectType::Checkpoint, &cp.encode())?
111
                }
112
            }
113
        };
114
        let cp = Checkpoint::decode(&self.store.get_typed(&r, ObjectType::Checkpoint)?)?;
115
        let mut bundle = Bundle::default();
116
        for c in Blob::decode(&self.store.get_typed(&cp.payload, ObjectType::Blob)?)?.chunks {
117
            bundle.objects.push(self.store.get(&c)?);
118
        }
119
        bundle.objects.push(self.store.get(&cp.payload)?);
120
        bundle.objects.push((ObjectType::Checkpoint, cp.encode()));
121
        Ok(bundle)
122
    }
123
124
    /// Load the state of `heads` using a checkpoint when one applies.
125
    pub(crate) fn load_state_checkpointed(&self, heads: &[ChangeHash]) -> Result<Option<State>> {
126
        let mut bookmarks: Vec<String> = vec![self.config.bookmark.clone()];
127
        if let Ok(rd) = fs::read_dir(self.oxy.join("checkpoints")) {
128
            for e in rd.flatten() {
129
                let name = e.file_name().to_string_lossy().to_string();
130
                if !bookmarks.contains(&name) {
131
                    bookmarks.push(name);
132
                }
133
            }
134
        }
135
        for b in bookmarks {
136
            let Some((_, cp)) = self.latest_checkpoint(&b)? else {
137
                continue;
138
            };
139
            let payload = self.read_blob(&cp.payload)?;
140
            let base = State::from_checkpoint_payload(&payload)?;
141
            let cp_heads: BTreeSet<ChangeHash> = cp.heads.iter().copied().collect();
142
            let mut found: BTreeSet<ChangeHash> = BTreeSet::new();
143
            let mut delta: Vec<Change> = Vec::new();
144
            let mut seen: BTreeSet<ChangeHash> = BTreeSet::new();
145
            let mut stack: Vec<ChangeHash> = heads.to_vec();
146
            while let Some(h) = stack.pop() {
147
                if !seen.insert(h) {
148
                    continue;
149
                }
150
                if base.contains(&h) {
151
                    if cp_heads.contains(&h) {
152
                        found.insert(h);
153
                    }
154
                    continue;
155
                }
156
                let c = self.store.get_change(&h)?;
157
                stack.extend(c.deps.iter().copied());
158
                delta.push(c);
159
            }
160
            if found.len() != cp_heads.len() {
161
                continue;
162
            }
163
            let mut state = base;
164
            for c in delta {
165
                state.add_unchecked(c);
166
            }
167
            state.validate()?;
168
            return Ok(Some(state));
169
        }
170
        Ok(None)
171
    }
172
}