znippy_plugin_git/
refs.rs

1//! `__gunnar_refs__` — the ref namespace as an append-only Arrow push log.
2//!
3//! One `RecordBatch` per push, written through [`crate::pushlog`]. A push that
4//! updates three branches is three rows in one batch, and those three either all
5//! land or none do, because the frame either completes or it does not. That is
6//! the whole transaction mechanism — there is no lock file, no journal and (D18)
7//! no database.
8//!
9//! ## Columns
10//!
11//! | column | meaning |
12//! |---|---|
13//! | `name` | full ref name, e.g. `refs/heads/main` |
14//! | `target` | the oid it now points at; **null means the ref was deleted** |
15//! | `peeled` | for an annotated tag, the commit it peels to |
16//! | `symref_target` | for a symbolic ref (`HEAD` → `refs/heads/main`), its target |
17//! | `push_seq` | monotonic push counter — the ordering authority |
18//! | `updated_ms` | wall clock, unix ms; for humans, never for ordering |
19//!
20//! `updated_ms` is deliberately not the ordering key: two pushes inside the same
21//! millisecond, or a clock that steps backwards, would silently reorder the ref
22//! namespace. `push_seq` is assigned by the writer and is the only thing
23//! [`fold`] compares.
24//!
25//! ## Current state
26//!
27//! The log is the history; [`fold`] replays it into the current namespace, last
28//! writer wins by `push_seq`, and a null `target` removes the ref. Reading the
29//! current refs is therefore a scan of a structure sized by *pushes*, not by
30//! repository size.
31
32use std::collections::BTreeMap;
33use std::path::Path;
34use std::sync::Arc;
35
36use anyhow::{Result, anyhow};
37use znippy_common::GUNNAR_REFS_MODULE;
38use znippy_common::arrow::array::{Array, StringArray, StringBuilder, UInt64Array, UInt64Builder};
39use znippy_common::arrow::datatypes::{DataType, Field, Schema};
40use znippy_common::arrow::record_batch::RecordBatch;
41
42use crate::pushlog::{PushLog, PushLogScan, read_sealed};
43
44/// One ref update inside a push.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct RefUpdate {
47    pub name: String,
48    /// `None` deletes the ref.
49    pub target: Option<String>,
50    pub peeled: Option<String>,
51    pub symref_target: Option<String>,
52}
53
54impl RefUpdate {
55    pub fn set(name: impl Into<String>, target: impl Into<String>) -> Self {
56        Self {
57            name: name.into(),
58            target: Some(target.into()),
59            peeled: None,
60            symref_target: None,
61        }
62    }
63
64    pub fn delete(name: impl Into<String>) -> Self {
65        Self { name: name.into(), target: None, peeled: None, symref_target: None }
66    }
67
68    pub fn symbolic(name: impl Into<String>, points_to: impl Into<String>) -> Self {
69        Self {
70            name: name.into(),
71            target: None,
72            peeled: None,
73            symref_target: Some(points_to.into()),
74        }
75    }
76
77    pub fn with_peeled(mut self, peeled: impl Into<String>) -> Self {
78        self.peeled = Some(peeled.into());
79        self
80    }
81}
82
83/// The state of one ref after replaying the log.
84#[derive(Debug, Clone, PartialEq, Eq)]
85pub struct RefState {
86    pub target: Option<String>,
87    pub peeled: Option<String>,
88    pub symref_target: Option<String>,
89    pub push_seq: u64,
90    pub updated_ms: u64,
91}
92
93pub fn refs_schema() -> Arc<Schema> {
94    Arc::new(Schema::new(vec![
95        Field::new("name", DataType::Utf8, false),
96        Field::new("target", DataType::Utf8, true),
97        Field::new("peeled", DataType::Utf8, true),
98        Field::new("symref_target", DataType::Utf8, true),
99        Field::new("push_seq", DataType::UInt64, false),
100        Field::new("updated_ms", DataType::UInt64, false),
101    ]))
102}
103
104/// Build the single `RecordBatch` that *is* one push.
105pub fn build_push_batch(updates: &[RefUpdate], push_seq: u64, updated_ms: u64) -> Result<RecordBatch> {
106    let n = updates.len();
107    let mut name = StringBuilder::with_capacity(n, n * 32);
108    let mut target = StringBuilder::with_capacity(n, n * 64);
109    let mut peeled = StringBuilder::with_capacity(n, n * 64);
110    let mut symref = StringBuilder::with_capacity(n, n * 32);
111    let mut seq = UInt64Builder::with_capacity(n);
112    let mut ms = UInt64Builder::with_capacity(n);
113
114    for u in updates {
115        name.append_value(&u.name);
116        match &u.target {
117            Some(t) => target.append_value(t),
118            None => target.append_null(),
119        }
120        match &u.peeled {
121            Some(t) => peeled.append_value(t),
122            None => peeled.append_null(),
123        }
124        match &u.symref_target {
125            Some(t) => symref.append_value(t),
126            None => symref.append_null(),
127        }
128        seq.append_value(push_seq);
129        ms.append_value(updated_ms);
130    }
131
132    RecordBatch::try_new(
133        refs_schema(),
134        vec![
135            Arc::new(name.finish()),
136            Arc::new(target.finish()),
137            Arc::new(peeled.finish()),
138            Arc::new(symref.finish()),
139            Arc::new(seq.finish()),
140            Arc::new(ms.finish()),
141        ],
142    )
143    .map_err(|e| anyhow!("refs push batch: {e}"))
144}
145
146/// The ref log of one repository.
147pub struct RefLog {
148    log: PushLog,
149}
150
151impl RefLog {
152    pub fn new(path: impl Into<std::path::PathBuf>) -> Self {
153        Self { log: PushLog::new(path, refs_schema()) }
154    }
155
156    /// The next `push_seq` this log should use: one past the highest already
157    /// recorded. Derived from the log itself, so a crashed writer that lost its
158    /// counter cannot reuse a sequence number and silently reorder history.
159    pub fn next_push_seq(&self) -> Result<u64> {
160        let scan = self.log.scan()?;
161        Ok(max_push_seq(&scan.pushes).map_or(0, |m| m + 1))
162    }
163
164    /// Append one push atomically. Returns the `push_seq` it was given.
165    pub fn push(&self, updates: &[RefUpdate]) -> Result<u64> {
166        let seq = self.next_push_seq()?;
167        let ms = std::time::SystemTime::now()
168            .duration_since(std::time::UNIX_EPOCH)
169            .map(|d| d.as_millis() as u64)
170            .unwrap_or(0);
171        let batch = build_push_batch(updates, seq, ms)?;
172        self.log.append(&batch)?;
173        Ok(seq)
174    }
175
176    /// Append an already-built push batch, whose `push_seq` the caller assigned.
177    ///
178    /// For a server that holds its own counter: [`push`](Self::push) re-derives
179    /// the sequence by rescanning the log, which is the safe default but is
180    /// O(log) per push. Returns the byte offset the frame starts at.
181    pub fn append_batch(&self, batch: &RecordBatch) -> Result<u64> {
182        self.log.append(batch)
183    }
184
185    pub fn scan(&self) -> Result<PushLogScan> {
186        self.log.scan()
187    }
188
189    /// Fold every frame into one. See [`PushLog::compact`] — rows and their
190    /// order are preserved, so [`fold`] answers identically before and after.
191    pub fn compact(&self) -> Result<crate::pushlog::CompactionReport> {
192        self.log.compact()
193    }
194
195    /// Compact if the log has grown past `policy`.
196    pub fn maybe_compact(
197        &self,
198        policy: crate::pushlog::CompactionPolicy,
199    ) -> Result<Option<crate::pushlog::CompactionReport>> {
200        self.log.maybe_compact(policy)
201    }
202
203    /// The current ref namespace.
204    pub fn current(&self) -> Result<BTreeMap<String, RefState>> {
205        Ok(fold(&self.log.scan()?.pushes)?)
206    }
207
208    /// The reserved section to seal into the archive.
209    pub fn seal_section(&self) -> Result<znippy_common::ReservedSection> {
210        self.log.seal_section(GUNNAR_REFS_MODULE)
211    }
212}
213
214fn max_push_seq(batches: &[RecordBatch]) -> Option<u64> {
215    let mut max = None;
216    for b in batches {
217        let seq = b.column_by_name("push_seq")?.as_any().downcast_ref::<UInt64Array>()?;
218        for i in 0..seq.len() {
219            max = Some(max.map_or(seq.value(i), |m: u64| m.max(seq.value(i))));
220        }
221    }
222    max
223}
224
225/// Replay pushes into the current namespace. Last writer wins by `push_seq`; a
226/// null `target` and no `symref_target` deletes the ref.
227///
228/// Ordering is by `push_seq` and never by position in the vector, so a caller
229/// that hands the batches over out of order still gets the right answer.
230pub fn fold(batches: &[RecordBatch]) -> Result<BTreeMap<String, RefState>> {
231    let mut rows: Vec<(u64, usize, RefState, String)> = Vec::new();
232
233    for (bi, b) in batches.iter().enumerate() {
234        let name = col::<StringArray>(b, "name")?;
235        let target = col::<StringArray>(b, "target")?;
236        let peeled = col::<StringArray>(b, "peeled")?;
237        let symref = col::<StringArray>(b, "symref_target")?;
238        let seq = col::<UInt64Array>(b, "push_seq")?;
239        let ms = col::<UInt64Array>(b, "updated_ms")?;
240
241        for i in 0..b.num_rows() {
242            rows.push((
243                seq.value(i),
244                bi,
245                RefState {
246                    target: (!target.is_null(i)).then(|| target.value(i).to_string()),
247                    peeled: (!peeled.is_null(i)).then(|| peeled.value(i).to_string()),
248                    symref_target: (!symref.is_null(i)).then(|| symref.value(i).to_string()),
249                    push_seq: seq.value(i),
250                    updated_ms: ms.value(i),
251                },
252                name.value(i).to_string(),
253            ));
254        }
255    }
256
257    rows.sort_by_key(|(seq, bi, _, _)| (*seq, *bi));
258
259    let mut out: BTreeMap<String, RefState> = BTreeMap::new();
260    for (_, _, state, name) in rows {
261        if state.target.is_none() && state.symref_target.is_none() {
262            out.remove(&name);
263        } else {
264            out.insert(name, state);
265        }
266    }
267    Ok(out)
268}
269
270/// Read the sealed `__gunnar_refs__` section out of an archive. `Ok(None)` when
271/// the archive carries none — distinct from an archive whose refs are empty.
272pub fn read_refs(archive: &Path) -> Result<Option<BTreeMap<String, RefState>>> {
273    match read_sealed(archive, GUNNAR_REFS_MODULE)? {
274        Some(batches) => Ok(Some(fold(&batches)?)),
275        None => Ok(None),
276    }
277}
278
279fn col<'a, T: Array + 'static>(b: &'a RecordBatch, name: &str) -> Result<&'a T> {
280    b.column_by_name(name)
281        .ok_or_else(|| anyhow!("refs: no `{name}` column"))?
282        .as_any()
283        .downcast_ref::<T>()
284        .ok_or_else(|| anyhow!("refs: `{name}` has an unexpected type"))
285}
286
287#[cfg(test)]
288mod tests {
289    use super::*;
290    use crate::pushlog::truncate_for_test;
291
292    fn tmpdir(tag: &str) -> std::path::PathBuf {
293        let ns = std::time::SystemTime::now()
294            .duration_since(std::time::UNIX_EPOCH)
295            .unwrap()
296            .as_nanos();
297        let d = std::env::temp_dir().join(format!("znippy_refs_{tag}_{ns}"));
298        std::fs::create_dir_all(&d).unwrap();
299        d
300    }
301
302    fn oid(c: char) -> String {
303        std::iter::repeat_n(c, 40).collect()
304    }
305
306    #[test]
307    fn last_writer_wins_and_a_null_target_deletes() {
308        let dir = tmpdir("fold");
309        let log = RefLog::new(dir.join("refs.log"));
310        log.push(&[
311            RefUpdate::set("refs/heads/main", oid('a')),
312            RefUpdate::set("refs/heads/topic", oid('b')),
313        ])
314        .unwrap();
315        log.push(&[RefUpdate::set("refs/heads/main", oid('c'))]).unwrap();
316        log.push(&[RefUpdate::delete("refs/heads/topic")]).unwrap();
317
318        let refs = log.current().unwrap();
319        assert_eq!(refs["refs/heads/main"].target, Some(oid('c')), "second push must win");
320        assert!(!refs.contains_key("refs/heads/topic"), "a null target deletes the ref");
321        assert_eq!(refs.len(), 1);
322
323        std::fs::remove_dir_all(&dir).ok();
324    }
325
326    /// A multi-ref push is one frame, so a crash during it must leave the ref
327    /// namespace exactly as it was before — never with some of its refs updated.
328    /// This is the property that makes the format usable without a lock file,
329    /// and the one a bolt-on ref file could not offer.
330    #[test]
331    fn a_crash_mid_push_leaves_no_partial_ref_update() {
332        let dir = tmpdir("atomic");
333        let path = dir.join("refs.log");
334        let log = RefLog::new(&path);
335        log.push(&[RefUpdate::set("refs/heads/main", oid('a'))]).unwrap();
336        let before = std::fs::metadata(&path).unwrap().len();
337
338        // A push touching three refs at once.
339        log.push(&[
340            RefUpdate::set("refs/heads/main", oid('9')),
341            RefUpdate::set("refs/heads/a", oid('1')),
342            RefUpdate::set("refs/heads/b", oid('2')),
343        ])
344        .unwrap();
345        let after = std::fs::metadata(&path).unwrap().len();
346        let intact = std::fs::read(&path).unwrap();
347
348        for cut in (before + 1)..after {
349            std::fs::write(&path, &intact).unwrap();
350            truncate_for_test(&path, cut).unwrap();
351            let refs = log.current().unwrap();
352            assert_eq!(
353                refs.len(),
354                1,
355                "cut at {cut}: a torn push must not publish ANY of its refs (got {refs:?})"
356            );
357            assert_eq!(
358                refs["refs/heads/main"].target,
359                Some(oid('a')),
360                "cut at {cut}: main must still be the pre-push value"
361            );
362            assert!(!refs.contains_key("refs/heads/a"), "cut at {cut}: leaked a partial ref");
363            assert!(!refs.contains_key("refs/heads/b"), "cut at {cut}: leaked a partial ref");
364        }
365
366        // Intact again: all three land together.
367        std::fs::write(&path, &intact).unwrap();
368        let refs = log.current().unwrap();
369        assert_eq!(refs.len(), 3, "the complete push publishes all three refs");
370        assert_eq!(refs["refs/heads/main"].target, Some(oid('9')));
371
372        std::fs::remove_dir_all(&dir).ok();
373    }
374
375    /// `push_seq` is re-derived from the log, so a writer that crashed and lost
376    /// its in-memory counter cannot reuse a sequence number.
377    #[test]
378    fn push_seq_is_recovered_from_the_log_not_from_memory() {
379        let dir = tmpdir("seq");
380        let path = dir.join("refs.log");
381        let a = RefLog::new(&path);
382        assert_eq!(a.push(&[RefUpdate::set("refs/heads/main", oid('a'))]).unwrap(), 0);
383        assert_eq!(a.push(&[RefUpdate::set("refs/heads/main", oid('b'))]).unwrap(), 1);
384
385        // A brand new writer over the same file — the "process restarted" case.
386        let b = RefLog::new(&path);
387        assert_eq!(
388            b.next_push_seq().unwrap(),
389            2,
390            "a restarted writer must continue the sequence, not restart it"
391        );
392        assert_eq!(b.push(&[RefUpdate::set("refs/heads/main", oid('c'))]).unwrap(), 2);
393        assert_eq!(b.current().unwrap()["refs/heads/main"].target, Some(oid('c')));
394
395        std::fs::remove_dir_all(&dir).ok();
396    }
397
398    /// Ordering is by `push_seq`, never by arrival order — a wall clock that
399    /// steps backwards, or two pushes in the same millisecond, must not reorder
400    /// the namespace.
401    #[test]
402    fn ordering_is_by_push_seq_not_by_timestamp_or_position() {
403        // Same `updated_ms` for both, and the LATER push handed over FIRST.
404        let newer = build_push_batch(&[RefUpdate::set("refs/heads/main", oid('c'))], 7, 1000).unwrap();
405        let older = build_push_batch(&[RefUpdate::set("refs/heads/main", oid('a'))], 3, 9999).unwrap();
406        let refs = fold(&[newer, older]).unwrap();
407        assert_eq!(
408            refs["refs/heads/main"].target,
409            Some(oid('c')),
410            "push_seq 7 must beat push_seq 3 regardless of order or clock"
411        );
412        assert_eq!(refs["refs/heads/main"].push_seq, 7);
413    }
414
415    #[test]
416    fn symbolic_and_peeled_refs_round_trip() {
417        let dir = tmpdir("sym");
418        let log = RefLog::new(dir.join("refs.log"));
419        log.push(&[
420            RefUpdate::symbolic("HEAD", "refs/heads/main"),
421            RefUpdate::set("refs/tags/v1", oid('t')).with_peeled(oid('e')),
422        ])
423        .unwrap();
424        let refs = log.current().unwrap();
425        assert_eq!(refs["HEAD"].symref_target.as_deref(), Some("refs/heads/main"));
426        assert!(refs["HEAD"].target.is_none(), "a symref has no direct target");
427        assert_eq!(refs["refs/tags/v1"].peeled, Some(oid('e')));
428        std::fs::remove_dir_all(&dir).ok();
429    }
430}
znippy_plugin_git/refs.rs

znippy_plugin_git/
refs.rs
