1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
//! Conformance tests for #262 multi-writer CAS on branch advance.
//!
//! Pre-#262, two writers calling `apply_operation` against the same
//! branch concurrently could either lose one another's update (last
//! write wins on the branch file) or land in an inconsistent state.
//! With #262, branch advance is a CAS guarded by `fs2` advisory
//! locking on a per-branch lockfile, and `apply_operation` retries
//! up to 8 times on contention before surfacing
//! `StoreError::Contention`.
use lex_store::{Operation, OperationKind, StageTransition, Store, DEFAULT_BRANCH};
use std::collections::BTreeSet;
use std::sync::Arc;
use std::thread;
fn fresh() -> (Arc<Store>, tempfile::TempDir) {
let tmp = tempfile::tempdir().unwrap();
let s = Store::open(tmp.path()).unwrap();
(Arc::new(s), tmp)
}
#[test]
fn n_concurrent_writers_all_land() {
// Spawn N threads, each calling `apply_operation` with a unique
// signature. After all threads finish, the head_state must
// contain every signature — none lost to races.
const N: usize = 12;
let (s, _tmp) = fresh();
let handles: Vec<_> = (0..N)
.map(|i| {
let s = Arc::clone(&s);
thread::spawn(move || {
let sig = format!("sig-{}", i);
let stage = format!("stg-{}", i);
let op = Operation::new(
OperationKind::AddFunction {
sig_id: sig.clone(),
stage_id: stage.clone(),
effects: BTreeSet::new(),
budget_cost: None,
},
[],
);
let t = StageTransition::Create {
sig_id: sig.clone(),
stage_id: stage.clone(),
};
s.apply_operation(DEFAULT_BRANCH, op, t)
.expect("apply should succeed under contention")
})
})
.collect();
for h in handles {
h.join().unwrap();
}
let head = s.branch_head(DEFAULT_BRANCH).unwrap();
assert_eq!(head.len(), N, "every writer's signature must be present");
for i in 0..N {
assert_eq!(head.get(&format!("sig-{}", i)), Some(&format!("stg-{}", i)));
}
}
#[test]
fn n_concurrent_writers_chain_into_a_single_history() {
// After N concurrent writers, the op-log must form a single
// chain of length N rooted at the genesis op (parents=[]).
// Each non-root op has exactly one parent, and the chain
// terminates at the branch head.
const N: usize = 8;
let (s, _tmp) = fresh();
let handles: Vec<_> = (0..N)
.map(|i| {
let s = Arc::clone(&s);
thread::spawn(move || {
let sig = format!("sig-{}", i);
let stage = format!("stg-{}", i);
let op = Operation::new(
OperationKind::AddFunction {
sig_id: sig.clone(),
stage_id: stage.clone(),
effects: BTreeSet::new(),
budget_cost: None,
},
[],
);
let t = StageTransition::Create {
sig_id: sig.clone(),
stage_id: stage.clone(),
};
s.apply_operation(DEFAULT_BRANCH, op, t).unwrap()
})
})
.collect();
for h in handles {
h.join().unwrap();
}
// Walk back from head and count nodes; verify single linear chain.
let log = lex_vcs::OpLog::open(s.root()).unwrap();
let head_op = s.get_branch(DEFAULT_BRANCH).unwrap().unwrap().head_op.unwrap();
let mut cursor = Some(head_op);
let mut count = 0;
while let Some(id) = cursor {
let rec = log.get(&id).unwrap().unwrap();
count += 1;
cursor = match rec.op.parents.len() {
0 => None,
1 => Some(rec.op.parents[0].clone()),
n => panic!("unexpected fan-in of {} at op {}", n, id),
};
}
assert_eq!(count, N, "history should be a linear chain of N ops");
}
#[test]
fn concurrent_writers_do_not_lose_op_records() {
// Even on contention paths where the final op_id changes after
// a CAS-mismatch retry (because the rebuilt op has a new
// parent), the *intermediate* persisted op record is still on
// disk. This test ensures we don't silently drop op records on
// retry — we just don't reference them from the branch head.
const N: usize = 6;
let (s, tmp) = fresh();
let handles: Vec<_> = (0..N)
.map(|i| {
let s = Arc::clone(&s);
thread::spawn(move || {
let sig = format!("sig-{}", i);
let stage = format!("stg-{}", i);
let op = Operation::new(
OperationKind::AddFunction {
sig_id: sig.clone(),
stage_id: stage.clone(),
effects: BTreeSet::new(),
budget_cost: None,
},
[],
);
let t = StageTransition::Create {
sig_id: sig.clone(),
stage_id: stage.clone(),
};
s.apply_operation(DEFAULT_BRANCH, op, t).unwrap()
})
})
.collect();
for h in handles {
h.join().unwrap();
}
// The ops/ directory should contain at least N op records (one
// per logical write). It may contain *more* if any writer was
// forced to retry — those orphan records are intentional under
// the append-only contract.
let ops_dir = tmp.path().join("ops");
let n_ops = std::fs::read_dir(&ops_dir)
.unwrap()
.filter_map(|e| e.ok())
.filter(|e| e.path().extension().is_some_and(|x| x == "json"))
.count();
assert!(
n_ops >= N,
"ops/ should hold at least {} records (got {})",
N,
n_ops
);
}
#[test]
fn empty_parents_op_chains_off_existing_head() {
// Deterministic regression for the cas_retry_advance race that
// surfaced in CI for #262: a caller submitting an op with
// `parents = []` against a branch that already has a head
// should chain off the existing head rather than fail
// StaleParent. This is the single-threaded analogue of the
// sibling-writer race in `n_concurrent_writers_all_land`.
let tmp = tempfile::tempdir().unwrap();
let s = Store::open(tmp.path()).unwrap();
fn add(sig: &str, stage: &str) -> (Operation, StageTransition) {
let op = Operation::new(
OperationKind::AddFunction {
sig_id: sig.into(),
stage_id: stage.into(),
effects: BTreeSet::new(),
budget_cost: None,
},
[],
);
let t = StageTransition::Create {
sig_id: sig.into(),
stage_id: stage.into(),
};
(op, t)
}
// First op lands cleanly.
let (op1, t1) = add("first", "stg-first");
let head1 = s.apply_operation(DEFAULT_BRANCH, op1, t1).unwrap();
// Second op is built with parents = [] (same shape as the
// concurrent-writer test). Pre-fix this would have failed
// StaleParent because attempt 1 doesn't rebuild.
let (op2, t2) = add("second", "stg-second");
assert!(op2.parents.is_empty(), "op2 must have empty parents to exercise the race");
let head2 = s.apply_operation(DEFAULT_BRANCH, op2, t2).unwrap();
assert_ne!(head1, head2, "second op should produce a fresh op_id");
// History walks back from head2 → head1, confirming the
// empty-parents op was rebuilt with head1 as parent.
let log = lex_vcs::OpLog::open(s.root()).unwrap();
let walked = log.walk_back(&head2, None).unwrap();
assert_eq!(walked.len(), 2);
assert_eq!(walked[0].op_id, head2);
assert_eq!(walked[1].op_id, head1);
assert_eq!(walked[0].op.parents, vec![head1.clone()]);
}