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