1use std::collections::BTreeMap;
33use std::path::Path;
34use std::sync::{Arc, Mutex, MutexGuard};
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
44pub use git_storage_trait::RefUpdate;
47
48#[derive(Debug, Clone, PartialEq, Eq)]
50pub struct RefState {
51 pub target: Option<String>,
52 pub peeled: Option<String>,
53 pub symref_target: Option<String>,
54 pub push_seq: u64,
55 pub updated_ms: u64,
56}
57
58pub fn refs_schema() -> Arc<Schema> {
59 Arc::new(Schema::new(vec![
60 Field::new("name", DataType::Utf8, false),
61 Field::new("target", DataType::Utf8, true),
62 Field::new("peeled", DataType::Utf8, true),
63 Field::new("symref_target", DataType::Utf8, true),
64 Field::new("push_seq", DataType::UInt64, false),
65 Field::new("updated_ms", DataType::UInt64, false),
66 ]))
67}
68
69pub fn build_push_batch(updates: &[RefUpdate], push_seq: u64, updated_ms: u64) -> Result<RecordBatch> {
71 let n = updates.len();
72 let mut name = StringBuilder::with_capacity(n, n * 32);
73 let mut target = StringBuilder::with_capacity(n, n * 64);
74 let mut peeled = StringBuilder::with_capacity(n, n * 64);
75 let mut symref = StringBuilder::with_capacity(n, n * 32);
76 let mut seq = UInt64Builder::with_capacity(n);
77 let mut ms = UInt64Builder::with_capacity(n);
78
79 for u in updates {
80 name.append_value(&u.name);
81 match &u.target {
82 Some(t) => target.append_value(t),
83 None => target.append_null(),
84 }
85 match &u.peeled {
86 Some(t) => peeled.append_value(t),
87 None => peeled.append_null(),
88 }
89 match &u.symref_target {
90 Some(t) => symref.append_value(t),
91 None => symref.append_null(),
92 }
93 seq.append_value(push_seq);
94 ms.append_value(updated_ms);
95 }
96
97 RecordBatch::try_new(
98 refs_schema(),
99 vec![
100 Arc::new(name.finish()),
101 Arc::new(target.finish()),
102 Arc::new(peeled.finish()),
103 Arc::new(symref.finish()),
104 Arc::new(seq.finish()),
105 Arc::new(ms.finish()),
106 ],
107 )
108 .map_err(|e| anyhow!("refs push batch: {e}"))
109}
110
111struct RefCache {
117 state: BTreeMap<String, RefState>,
118 next_seq: u64,
119}
120
121impl RefCache {
122 fn apply(&mut self, updates: &[RefUpdate], seq: u64, ms: u64) {
127 for u in updates {
128 if u.target.is_none() && u.symref_target.is_none() {
129 self.state.remove(&u.name);
130 } else {
131 self.state.insert(
132 u.name.clone(),
133 RefState {
134 target: u.target.clone(),
135 peeled: u.peeled.clone(),
136 symref_target: u.symref_target.clone(),
137 push_seq: seq,
138 updated_ms: ms,
139 },
140 );
141 }
142 }
143 self.next_seq = self.next_seq.max(seq + 1);
144 }
145}
146
147pub struct RefLog {
180 log: PushLog,
181 cache: Mutex<Option<RefCache>>,
182}
183
184impl RefLog {
185 pub fn new(path: impl Into<std::path::PathBuf>) -> Self {
186 Self { log: PushLog::new(path, refs_schema()), cache: Mutex::new(None) }
187 }
188
189 fn cached(&self) -> Result<MutexGuard<'_, Option<RefCache>>> {
191 let mut guard = self
192 .cache
193 .lock()
194 .map_err(|_| anyhow!("the ref log cache is poisoned"))?;
195 if guard.is_none() {
196 let scan = self.log.scan()?;
197 *guard = Some(RefCache {
198 state: fold(&scan.pushes)?,
199 next_seq: max_push_seq(&scan.pushes).map_or(0, |m| m + 1),
200 });
201 }
202 Ok(guard)
203 }
204
205 pub fn invalidate(&self) {
213 if let Ok(mut g) = self.cache.lock() {
214 *g = None;
215 }
216 }
217
218 pub fn next_push_seq(&self) -> Result<u64> {
223 Ok(self.cached()?.as_ref().expect("loaded above").next_seq)
224 }
225
226 pub fn push(&self, updates: &[RefUpdate]) -> Result<u64> {
229 self.push_with(updates, true)
230 }
231
232 pub fn push_rebuildable(&self, updates: &[RefUpdate]) -> Result<u64> {
242 self.push_with(updates, false)
243 }
244
245 fn push_with(&self, updates: &[RefUpdate], durable: bool) -> Result<u64> {
246 let mut guard = self.cached()?;
247 let cache = guard.as_mut().expect("loaded above");
248 let seq = cache.next_seq;
249 let ms = std::time::SystemTime::now()
250 .duration_since(std::time::UNIX_EPOCH)
251 .map(|d| d.as_millis() as u64)
252 .unwrap_or(0);
253 let batch = build_push_batch(updates, seq, ms)?;
254 self.log.append_with(&batch, durable)?;
258 cache.apply(updates, seq, ms);
259 Ok(seq)
260 }
261
262 pub fn append_batch(&self, batch: &RecordBatch) -> Result<u64> {
269 let offset = self.log.append(batch)?;
270 self.invalidate();
271 Ok(offset)
272 }
273
274 pub fn scan(&self) -> Result<PushLogScan> {
275 self.log.scan()
276 }
277
278 pub fn scans(&self) -> u64 {
281 self.log.scans()
282 }
283
284 pub fn compact(&self) -> Result<crate::pushlog::CompactionReport> {
287 self.log.compact()
288 }
289
290 pub fn maybe_compact(
292 &self,
293 policy: crate::pushlog::CompactionPolicy,
294 ) -> Result<Option<crate::pushlog::CompactionReport>> {
295 self.log.maybe_compact(policy)
296 }
297
298 pub fn current(&self) -> Result<BTreeMap<String, RefState>> {
303 Ok(self.cached()?.as_ref().expect("loaded above").state.clone())
304 }
305
306 pub fn get(&self, name: &str) -> Result<Option<RefState>> {
315 Ok(self
316 .cached()?
317 .as_ref()
318 .expect("loaded above")
319 .state
320 .get(name)
321 .cloned())
322 }
323
324 pub fn seal_section(&self) -> Result<znippy_common::ReservedSection> {
326 self.log.seal_section(GUNNAR_REFS_MODULE)
327 }
328}
329
330fn max_push_seq(batches: &[RecordBatch]) -> Option<u64> {
331 let mut max = None;
332 for b in batches {
333 let seq = b.column_by_name("push_seq")?.as_any().downcast_ref::<UInt64Array>()?;
334 for i in 0..seq.len() {
335 max = Some(max.map_or(seq.value(i), |m: u64| m.max(seq.value(i))));
336 }
337 }
338 max
339}
340
341pub fn fold(batches: &[RecordBatch]) -> Result<BTreeMap<String, RefState>> {
347 let mut rows: Vec<(u64, usize, RefState, String)> = Vec::new();
348
349 for (bi, b) in batches.iter().enumerate() {
350 let name = col::<StringArray>(b, "name")?;
351 let target = col::<StringArray>(b, "target")?;
352 let peeled = col::<StringArray>(b, "peeled")?;
353 let symref = col::<StringArray>(b, "symref_target")?;
354 let seq = col::<UInt64Array>(b, "push_seq")?;
355 let ms = col::<UInt64Array>(b, "updated_ms")?;
356
357 for i in 0..b.num_rows() {
358 rows.push((
359 seq.value(i),
360 bi,
361 RefState {
362 target: (!target.is_null(i)).then(|| target.value(i).to_string()),
363 peeled: (!peeled.is_null(i)).then(|| peeled.value(i).to_string()),
364 symref_target: (!symref.is_null(i)).then(|| symref.value(i).to_string()),
365 push_seq: seq.value(i),
366 updated_ms: ms.value(i),
367 },
368 name.value(i).to_string(),
369 ));
370 }
371 }
372
373 rows.sort_by_key(|(seq, bi, _, _)| (*seq, *bi));
374
375 let mut out: BTreeMap<String, RefState> = BTreeMap::new();
376 for (_, _, state, name) in rows {
377 if state.target.is_none() && state.symref_target.is_none() {
378 out.remove(&name);
379 } else {
380 out.insert(name, state);
381 }
382 }
383 Ok(out)
384}
385
386pub fn read_refs(archive: &Path) -> Result<Option<BTreeMap<String, RefState>>> {
389 match read_sealed(archive, GUNNAR_REFS_MODULE)? {
390 Some(batches) => Ok(Some(fold(&batches)?)),
391 None => Ok(None),
392 }
393}
394
395fn col<'a, T: Array + 'static>(b: &'a RecordBatch, name: &str) -> Result<&'a T> {
396 b.column_by_name(name)
397 .ok_or_else(|| anyhow!("refs: no `{name}` column"))?
398 .as_any()
399 .downcast_ref::<T>()
400 .ok_or_else(|| anyhow!("refs: `{name}` has an unexpected type"))
401}
402
403#[cfg(test)]
404mod tests {
405 use super::*;
406 use crate::pushlog::truncate_for_test;
407
408 fn tmpdir(tag: &str) -> std::path::PathBuf {
409 let ns = std::time::SystemTime::now()
410 .duration_since(std::time::UNIX_EPOCH)
411 .unwrap()
412 .as_nanos();
413 let d = std::env::temp_dir().join(format!("znippy_refs_{tag}_{ns}"));
414 std::fs::create_dir_all(&d).unwrap();
415 d
416 }
417
418 fn oid(c: char) -> String {
419 std::iter::repeat_n(c, 40).collect()
420 }
421
422 #[test]
432 fn get_answers_exactly_what_current_holds() {
433 let dir = tmpdir("keyed-get");
434 let log = RefLog::new(dir.join("refs.log"));
435 log.push(&[
436 RefUpdate::set("refs/heads/main", oid('a')),
437 RefUpdate::set("refs/heads/topic", oid('b')),
438 RefUpdate::set("refs/tags/v1", oid('c')),
439 ])
440 .unwrap();
441 log.push(&[RefUpdate::set("refs/heads/main", oid('d'))]).unwrap();
442 log.push(&[RefUpdate::delete("refs/heads/topic")]).unwrap();
443
444 let whole = log.current().unwrap();
445 for (name, state) in &whole {
448 assert_eq!(
449 log.get(name).unwrap().as_ref(),
450 Some(state),
451 "get({name}) != current[{name}]"
452 );
453 }
454 assert_eq!(log.get("refs/heads/topic").unwrap(), None);
457 assert!(!whole.contains_key("refs/heads/topic"));
458 assert_eq!(log.get("refs/heads/nope").unwrap(), None);
460 assert_eq!(log.get("refs/heads/mai").unwrap(), None);
461 assert_eq!(log.get("refs/heads/main").unwrap().unwrap().target, Some(oid('d')));
463
464 std::fs::remove_dir_all(&dir).ok();
465 }
466
467 #[test]
468 fn last_writer_wins_and_a_null_target_deletes() {
469 let dir = tmpdir("fold");
470 let log = RefLog::new(dir.join("refs.log"));
471 log.push(&[
472 RefUpdate::set("refs/heads/main", oid('a')),
473 RefUpdate::set("refs/heads/topic", oid('b')),
474 ])
475 .unwrap();
476 log.push(&[RefUpdate::set("refs/heads/main", oid('c'))]).unwrap();
477 log.push(&[RefUpdate::delete("refs/heads/topic")]).unwrap();
478
479 let refs = log.current().unwrap();
480 assert_eq!(refs["refs/heads/main"].target, Some(oid('c')), "second push must win");
481 assert!(!refs.contains_key("refs/heads/topic"), "a null target deletes the ref");
482 assert_eq!(refs.len(), 1);
483
484 std::fs::remove_dir_all(&dir).ok();
485 }
486
487 #[test]
492 fn a_crash_mid_push_leaves_no_partial_ref_update() {
493 let dir = tmpdir("atomic");
494 let path = dir.join("refs.log");
495 let log = RefLog::new(&path);
496 log.push(&[RefUpdate::set("refs/heads/main", oid('a'))]).unwrap();
497 let before = std::fs::metadata(&path).unwrap().len();
498
499 log.push(&[
501 RefUpdate::set("refs/heads/main", oid('9')),
502 RefUpdate::set("refs/heads/a", oid('1')),
503 RefUpdate::set("refs/heads/b", oid('2')),
504 ])
505 .unwrap();
506 let after = std::fs::metadata(&path).unwrap().len();
507 let intact = std::fs::read(&path).unwrap();
508
509 for cut in (before + 1)..after {
510 std::fs::write(&path, &intact).unwrap();
511 truncate_for_test(&path, cut).unwrap();
512 let refs = RefLog::new(&path).current().unwrap();
518 assert_eq!(
519 refs.len(),
520 1,
521 "cut at {cut}: a torn push must not publish ANY of its refs (got {refs:?})"
522 );
523 assert_eq!(
524 refs["refs/heads/main"].target,
525 Some(oid('a')),
526 "cut at {cut}: main must still be the pre-push value"
527 );
528 assert!(!refs.contains_key("refs/heads/a"), "cut at {cut}: leaked a partial ref");
529 assert!(!refs.contains_key("refs/heads/b"), "cut at {cut}: leaked a partial ref");
530 }
531
532 std::fs::write(&path, &intact).unwrap();
534 let refs = RefLog::new(&path).current().unwrap();
535 assert_eq!(refs.len(), 3, "the complete push publishes all three refs");
536 assert_eq!(refs["refs/heads/main"].target, Some(oid('9')));
537
538 std::fs::remove_dir_all(&dir).ok();
539 }
540
541 #[test]
544 fn push_seq_is_recovered_from_the_log_not_from_memory() {
545 let dir = tmpdir("seq");
546 let path = dir.join("refs.log");
547 let a = RefLog::new(&path);
548 assert_eq!(a.push(&[RefUpdate::set("refs/heads/main", oid('a'))]).unwrap(), 0);
549 assert_eq!(a.push(&[RefUpdate::set("refs/heads/main", oid('b'))]).unwrap(), 1);
550
551 let b = RefLog::new(&path);
553 assert_eq!(
554 b.next_push_seq().unwrap(),
555 2,
556 "a restarted writer must continue the sequence, not restart it"
557 );
558 assert_eq!(b.push(&[RefUpdate::set("refs/heads/main", oid('c'))]).unwrap(), 2);
559 assert_eq!(b.current().unwrap()["refs/heads/main"].target, Some(oid('c')));
560
561 std::fs::remove_dir_all(&dir).ok();
562 }
563
564 #[test]
568 fn ordering_is_by_push_seq_not_by_timestamp_or_position() {
569 let newer = build_push_batch(&[RefUpdate::set("refs/heads/main", oid('c'))], 7, 1000).unwrap();
571 let older = build_push_batch(&[RefUpdate::set("refs/heads/main", oid('a'))], 3, 9999).unwrap();
572 let refs = fold(&[newer, older]).unwrap();
573 assert_eq!(
574 refs["refs/heads/main"].target,
575 Some(oid('c')),
576 "push_seq 7 must beat push_seq 3 regardless of order or clock"
577 );
578 assert_eq!(refs["refs/heads/main"].push_seq, 7);
579 }
580
581 #[test]
594 fn the_cost_of_a_push_does_not_grow_with_the_log() {
595 for prior in [1usize, 100, 1000] {
596 let dir = tmpdir(&format!("flat{prior}"));
597 let path = dir.join("refs.log");
598 let log = RefLog::new(&path);
599 for i in 0..prior {
600 log.push(&[RefUpdate::set(
601 format!("refs/heads/b{i:05}"),
602 format!("{i:040x}"),
603 )])
604 .unwrap();
605 }
606 let after_writes = log.scans();
607 let _ = log.current().unwrap();
609 log.push(&[RefUpdate::set("refs/heads/main", oid('a'))]).unwrap();
610
611 assert_eq!(
612 after_writes, 1,
613 "{prior} pushes through one handle must cost ONE pass over the log, not one per push (saw {after_writes})"
614 );
615 assert_eq!(
616 log.scans(),
617 1,
618 "the compare-and-swap pair (fold, then append) must add no pass at all"
619 );
620 assert_eq!(log.next_push_seq().unwrap(), prior as u64 + 1);
621 assert_eq!(
622 log.current().unwrap()["refs/heads/main"].target,
623 Some(oid('a'))
624 );
625 assert_eq!(RefLog::new(&path).current().unwrap(), log.current().unwrap());
628 std::fs::remove_dir_all(&dir).ok();
629 }
630 }
631
632 #[test]
636 fn a_rebuildable_push_is_a_whole_frame_that_simply_was_not_fsynced() {
637 let dir = tmpdir("rebuildable");
638 let path = dir.join("refs.log");
639 let log = RefLog::new(&path);
640 log.push_rebuildable(&[RefUpdate::symbolic("HEAD", "refs/heads/main")])
641 .unwrap();
642 assert_eq!(
643 RefLog::new(&path).current().unwrap()["HEAD"].symref_target.as_deref(),
644 Some("refs/heads/main"),
645 "the frame is complete on disk; only the fsync was skipped"
646 );
647 assert_eq!(log.push(&[RefUpdate::set("refs/heads/main", oid('a'))]).unwrap(), 1);
648 let cold = RefLog::new(&path).current().unwrap();
649 assert_eq!(cold.len(), 2, "both frames fold, in one log");
650 assert_eq!(cold["refs/heads/main"].target, Some(oid('a')));
651 std::fs::remove_dir_all(&dir).ok();
652 }
653
654 #[test]
655 fn symbolic_and_peeled_refs_round_trip() {
656 let dir = tmpdir("sym");
657 let log = RefLog::new(dir.join("refs.log"));
658 log.push(&[
659 RefUpdate::symbolic("HEAD", "refs/heads/main"),
660 RefUpdate::set("refs/tags/v1", oid('t')).with_peeled(oid('e')),
661 ])
662 .unwrap();
663 let refs = log.current().unwrap();
664 assert_eq!(refs["HEAD"].symref_target.as_deref(), Some("refs/heads/main"));
665 assert!(refs["HEAD"].target.is_none(), "a symref has no direct target");
666 assert_eq!(refs["refs/tags/v1"].peeled, Some(oid('e')));
667 std::fs::remove_dir_all(&dir).ok();
668 }
669}