1use crate::error::{Error, Result};
15use crate::repo::VaultRepo;
16use git2::Oid;
17use tracing::instrument;
18
19const DEFAULT_MAX_RETRIES: u32 = 8;
22
23impl VaultRepo {
24 #[instrument(
31 skip(self),
32 fields(refname = %refname, expected = ?expected_old, new = %new),
33 name = "git_cas_ref"
34 )]
35 pub fn cas_ref(&self, refname: &str, expected_old: Option<Oid>, new: Oid) -> Result<()> {
36 let repo = self.git();
37 let mut tx = repo.transaction()?;
38 tx.lock_ref(refname)?;
39 let current = match repo.refname_to_id(refname) {
51 Ok(oid) => Some(oid),
52 Err(e) if e.code() == git2::ErrorCode::NotFound => None,
53 Err(e) => return Err(Error::Git(e)),
54 };
55 if current != expected_old {
56 return Err(Error::CasConflict {
58 refname: refname.to_string(),
59 expected: expected_old,
60 found: current,
61 });
62 }
63 tx.set_target(refname, new, None, "turbovault-git: cas advance")?;
64 tx.commit()?;
65 Ok(())
66 }
67
68 pub fn commit_with_retry<F>(&self, refname: &str, build: F) -> Result<Option<Oid>>
71 where
72 F: FnMut(Option<Oid>) -> Result<Option<Oid>>,
73 {
74 self.commit_with_retry_n(refname, DEFAULT_MAX_RETRIES, build)
75 }
76
77 #[instrument(
90 skip(self, build),
91 fields(refname = %refname, max_retries),
92 name = "git_commit_with_retry"
93 )]
94 pub fn commit_with_retry_n<F>(
95 &self,
96 refname: &str,
97 max_retries: u32,
98 mut build: F,
99 ) -> Result<Option<Oid>>
100 where
101 F: FnMut(Option<Oid>) -> Result<Option<Oid>>,
102 {
103 for _ in 0..=max_retries {
104 let tip = match self.git().refname_to_id(refname) {
108 Ok(oid) => Some(oid),
109 Err(e) if e.code() == git2::ErrorCode::NotFound => None,
110 Err(e) => return Err(Error::Git(e)),
111 };
112 let new = match build(tip)? {
115 Some(oid) => oid,
116 None => return Ok(None),
117 };
118 match self.cas_ref(refname, tip, new) {
119 Ok(()) => return Ok(Some(new)),
120 Err(Error::CasConflict { .. }) => continue,
123 Err(e) => return Err(e),
124 }
125 }
126 Err(Error::Other(format!(
127 "ref CAS exhausted {max_retries} retries on {refname} (excessive contention)"
128 )))
129 }
130}
131
132#[cfg(test)]
133mod tests {
134 use super::*;
135 use crate::plumbing::TreeChange;
136 use git2::Repository;
137 use std::cell::Cell;
138 use tempfile::TempDir;
139
140 const MAIN: &str = "refs/heads/main";
141
142 fn open_unborn() -> (TempDir, VaultRepo) {
143 let tmp = TempDir::new().unwrap();
144 let mut opts = git2::RepositoryInitOptions::new();
145 opts.initial_head("main");
146 Repository::init_opts(tmp.path(), &opts).unwrap();
147 let vr = VaultRepo::open(tmp.path()).unwrap();
148 (tmp, vr)
149 }
150
151 fn upsert(path: &str, content: &str) -> TreeChange {
152 TreeChange::Upsert {
153 path: path.to_string(),
154 content: content.as_bytes().to_vec(),
155 }
156 }
157
158 fn build_on(vr: &VaultRepo, parent: Option<Oid>, path: &str, content: &str) -> Oid {
160 let base = parent.map(|p| vr.git().find_commit(p).unwrap().tree_id());
161 let tree = vr.build_tree(base, &[upsert(path, content)]).unwrap();
162 let parents: Vec<Oid> = parent.into_iter().collect();
163 vr.commit_tree(tree, &parents, "c").unwrap()
164 }
165
166 #[test]
167 fn cas_ref_initial_then_advance() {
168 let (_tmp, vr) = open_unborn();
169 let c0 = build_on(&vr, None, "a.md", "a");
170 vr.cas_ref(MAIN, None, c0)
171 .expect("initial CAS (None -> c0)");
172 assert_eq!(vr.head_oid(), Some(c0));
173
174 let c1 = build_on(&vr, Some(c0), "b.md", "b");
175 vr.cas_ref(MAIN, Some(c0), c1).expect("advance c0 -> c1");
176 assert_eq!(vr.head_oid(), Some(c1));
177 }
178
179 #[test]
185 fn corrupt_ref_surfaces_error_instead_of_silent_absent() {
186 let (tmp, vr) = open_unborn();
187 let c0 = build_on(&vr, None, "a.md", "a");
188 vr.cas_ref(MAIN, None, c0).unwrap();
189 drop(vr); std::fs::write(tmp.path().join(".git/refs/heads/main"), "not-a-valid-oid\n").unwrap();
193 let vr = VaultRepo::open(tmp.path()).unwrap();
194
195 let code = vr.git().refname_to_id(MAIN).unwrap_err().code();
198 assert_ne!(
199 code,
200 git2::ErrorCode::NotFound,
201 "corruption must produce a non-NotFound error; got {code:?}"
202 );
203
204 let res = vr.commit_with_retry_n(MAIN, 0, |_tip| Ok(None));
207 assert!(
208 res.is_err(),
209 "commit_with_retry must surface the ref-read error, not swallow to None"
210 );
211
212 let some = Oid::from_str("0000000000000000000000000000000000000001").unwrap();
214 assert!(
215 vr.cas_ref(MAIN, None, some).is_err(),
216 "cas_ref must surface the ref-read error, not blind-write a 'new' ref"
217 );
218 }
219
220 #[test]
221 fn cas_ref_rejects_stale_and_leaves_ref() {
222 let (_tmp, vr) = open_unborn();
223 let c0 = build_on(&vr, None, "a.md", "a");
224 vr.cas_ref(MAIN, None, c0).unwrap();
225
226 let bogus = Oid::from_str("0000000000000000000000000000000000000001").unwrap();
227 let c1 = build_on(&vr, Some(c0), "b.md", "b");
228 match vr.cas_ref(MAIN, Some(bogus), c1) {
229 Err(Error::CasConflict { found, .. }) => assert_eq!(found, Some(c0)),
230 other => panic!("expected CasConflict, got {other:?}"),
231 }
232 assert_eq!(vr.head_oid(), Some(c0), "ref unchanged on reject");
233 }
234
235 #[test]
236 fn cas_ref_initial_rejects_when_ref_exists() {
237 let (_tmp, vr) = open_unborn();
238 let c0 = build_on(&vr, None, "a.md", "a");
239 vr.cas_ref(MAIN, None, c0).unwrap();
240 let c1 = build_on(&vr, Some(c0), "b.md", "b");
242 assert!(matches!(
243 vr.cas_ref(MAIN, None, c1),
244 Err(Error::CasConflict { .. })
245 ));
246 }
247
248 #[test]
255 fn reused_handle_detects_external_ref_advance_no_lost_update() {
256 let (tmp, vr_a) = open_unborn();
257 let c0 = build_on(&vr_a, None, "a.md", "v1");
259 vr_a.cas_ref(MAIN, None, c0).unwrap();
260 assert_eq!(vr_a.head_oid(), Some(c0));
261
262 let vr_b = VaultRepo::open(tmp.path()).unwrap();
264 let c1 = build_on(&vr_b, Some(c0), "b.md", "from-B");
265 vr_b.cas_ref(MAIN, Some(c0), c1).unwrap();
266
267 let got = vr_a
271 .commit_with_retry(MAIN, |tip| Ok(Some(build_on(&vr_a, tip, "c.md", "from-A"))))
272 .unwrap()
273 .expect("a commit was produced");
274 let parent = vr_a.git().find_commit(got).unwrap().parent_id(0).unwrap();
275 assert_eq!(
276 parent, c1,
277 "reused handle committed atop B's external advance (saw the ref change; no lost update)"
278 );
279 assert!(
280 vr_a.git().find_commit(c1).is_ok(),
281 "B's commit is still reachable, not clobbered"
282 );
283 }
284
285 #[test]
286 fn commit_with_retry_no_contention() {
287 let (_tmp, vr) = open_unborn();
288 let c0 = build_on(&vr, None, "a.md", "a");
289 vr.cas_ref(MAIN, None, c0).unwrap();
290
291 let got = vr
292 .commit_with_retry(MAIN, |tip| Ok(Some(build_on(&vr, tip, "b.md", "b"))))
293 .unwrap()
294 .expect("a commit was produced");
295 assert_eq!(vr.head_oid(), Some(got));
296 }
297
298 #[test]
299 fn commit_with_retry_rebuilds_on_conflict() {
300 let (_tmp, vr) = open_unborn();
301 let c0 = build_on(&vr, None, "a.md", "a");
302 vr.cas_ref(MAIN, None, c0).unwrap();
303
304 let calls = Cell::new(0u32);
305 let got = vr
306 .commit_with_retry(MAIN, |tip| {
307 calls.set(calls.get() + 1);
308 let tip = tip.unwrap();
309 if calls.get() == 1 {
312 let concurrent = build_on(&vr, Some(tip), "concurrent.md", "x");
313 vr.cas_ref(MAIN, Some(tip), concurrent).unwrap();
314 }
315 Ok(Some(build_on(&vr, Some(tip), "mine.md", "m")))
316 })
317 .unwrap()
318 .expect("a commit was produced");
319
320 assert_eq!(calls.get(), 2, "exactly one rebuild after the conflict");
321 assert_eq!(vr.head_oid(), Some(got));
322 let head_tree = vr.git().find_commit(got).unwrap().tree_id();
324 assert!(
325 vr.blob_oid_at(head_tree, "concurrent.md")
326 .unwrap()
327 .is_some()
328 );
329 assert!(vr.blob_oid_at(head_tree, "mine.md").unwrap().is_some());
330 }
331
332 #[test]
337 fn commit_with_retry_exhausts_under_relentless_contention() {
338 let (_tmp, vr) = open_unborn();
339 let c0 = build_on(&vr, None, "a.md", "a");
340 vr.cas_ref(MAIN, None, c0).unwrap();
341
342 let calls = Cell::new(0u32);
343 let err = vr
344 .commit_with_retry_n(MAIN, 2, |tip| {
345 calls.set(calls.get() + 1);
346 let tip = tip.unwrap();
347 let concurrent = build_on(&vr, Some(tip), &format!("c{}.md", calls.get()), "x");
349 vr.cas_ref(MAIN, Some(tip), concurrent).unwrap();
350 Ok(Some(build_on(&vr, Some(tip), "mine.md", "m")))
351 })
352 .unwrap_err();
353
354 assert_eq!(
356 calls.get(),
357 3,
358 "builder runs max_retries+1 times then gives up"
359 );
360 assert!(
361 err.to_string().contains("exhausted") && err.to_string().contains("contention"),
362 "loud exhaustion error: {err}"
363 );
364 }
365
366 #[test]
373 fn parallel_commit_changeset_lands_every_commit() {
374 let (tmp, vr0) = open_unborn();
375 vr0.commit_changeset(&crate::Changeset::new("seed").create("seed.md", "0"))
376 .unwrap();
377 let path = tmp.path().to_path_buf();
378 let locks = vr0.commit_locks();
379 drop(vr0);
380
381 let n = 8u32;
382 let handles: Vec<_> = (0..n)
383 .map(|i| {
384 let p = path.clone();
385 let l = std::sync::Arc::clone(&locks);
386 std::thread::spawn(move || {
387 let vr = crate::VaultRepo::open_with_locks(&p, l).unwrap();
388 vr.commit_changeset(
389 &crate::Changeset::new("c").create(format!("f{i}.md"), "x"),
390 )
391 .unwrap();
392 })
393 })
394 .collect();
395 for h in handles {
396 h.join().unwrap();
397 }
398
399 let vr = crate::VaultRepo::open_with_locks(&path, locks).unwrap();
401 let tree = vr
402 .git()
403 .find_commit(vr.head_oid().unwrap())
404 .unwrap()
405 .tree_id();
406 assert!(vr.blob_oid_at(tree, "seed.md").unwrap().is_some());
407 for i in 0..n {
408 assert!(
409 vr.blob_oid_at(tree, &format!("f{i}.md")).unwrap().is_some(),
410 "f{i}.md must have landed"
411 );
412 }
413 }
414}