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
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
//! Branches: each branch is identified by a name and a head OpId.
//! The SigId → StageId map every consumer reads is computed by
//! replaying the op log from the head back. No materialized cache.
//!
//! `lifecycle.json` (Draft/Active/Deprecated/Tombstone per stage)
//! survives as orthogonal stage-status metadata; it no longer drives
//! branch resolution.
use crate::store::{Store, StoreError};
use lex_vcs::{OpId, OpLog, StageTransition};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::fs;
use std::path::PathBuf;
/// Why a CAS branch advance failed (#262). Surfaced through the
/// retry loop in `Store::apply_operation` and friends; callers
/// either retry (on `Mismatch`, after re-reading the head and
/// rebuilding the candidate op) or propagate (on `Io` /
/// `UnknownBranch`).
#[derive(Debug)]
pub enum CasFailed {
/// Read-time `head_op` didn't match the supplied `expected`.
/// Another writer advanced the branch between this caller's
/// read and write. `actual` is the head we found instead.
/// Public field so callers (e.g. an HTTP layer) can surface
/// the actual head in a structured error envelope; today's
/// retry loop just discards it and re-reads on the next
/// iteration.
#[allow(dead_code)] // populated for callers that inspect the variant
Mismatch { actual: Option<OpId> },
/// Branch doesn't exist (and isn't the default branch).
UnknownBranch(String),
/// Disk I/O failure (lock acquisition, file read, atomic
/// write). Stringified at the boundary because `io::Error`
/// isn't `Clone`/`PartialEq` and the variant is consumed by
/// the retry loop, not pattern-matched on.
Io(String),
}
pub const DEFAULT_BRANCH: &str = "main";
/// Persisted, best-effort cache of `branch_head`'s computed view,
/// keyed on the head it was computed for. See `Store::branch_head`'s
/// doc comment for the incremental-replay design this backs.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct HeadSnapshot {
head_op: OpId,
map: BTreeMap<String, String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct Branch {
pub name: String,
pub parent: Option<String>,
/// Op DAG head. `None` means the branch has never had an op
/// applied (empty branch) *or* it's a predicate-defined branch
/// where the head is computed lazily from `predicate`.
#[serde(default)]
pub head_op: Option<OpId>,
/// Predicate over the op log (#133). When `Some`, the branch is
/// a saved query rather than a snapshot — `head_op` is the
/// optional materialization cache. The predicate's JSON shape
/// matches `lex_vcs::Predicate::to_value()`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub predicate: Option<serde_json::Value>,
/// Append-only journal of merges committed *into* this branch.
#[serde(default)]
pub merges: Vec<MergeRecord>,
pub created_at: u64,
/// Last op_id through which the producer-block gate (#248) has
/// verified the branch's history (#256). When advancing from
/// `head_op` to a new tip, the gate walks ops in
/// `(last_gate_checkpoint .. head_op]` and runs the
/// producer-block check on each ancestor's attestable stages —
/// not just the new op. Once that walk passes, the checkpoint
/// advances.
///
/// Invalidated (set to `None`) when `lex attest retro-block`
/// lands a new `ProducerBlock` attestation, forcing the next
/// advance to re-walk from genesis once. Steady-state advances
/// are `O(new ops)` because the previous advance already
/// covered everything up through `last_gate_checkpoint`.
///
/// Pre-#256 branch files have no `last_gate_checkpoint` field;
/// serde defaults to `None`, which forces a one-time full walk
/// on next advance. Same backward-compat trick `intent_id`
/// (#131) used.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_gate_checkpoint: Option<OpId>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct MergeRecord {
pub src: String,
pub at: u64,
pub merged: usize,
pub conflicts: usize,
}
#[derive(Debug, Clone, Serialize)]
pub struct MergeReport {
pub summary: MergeSummary,
pub merged: Vec<MergeEntry>,
pub conflicts: Vec<MergeConflict>,
}
#[derive(Debug, Clone, Serialize, Default)]
pub struct MergeSummary {
pub total_sigs: usize,
pub clean: usize,
pub conflicts: usize,
pub base: Option<String>,
#[serde(default)]
pub src: String,
#[serde(default)]
pub dst: String,
}
#[derive(Debug, Clone, Serialize)]
pub struct MergeEntry {
pub sig_id: String,
pub stage_id: String,
pub from: &'static str, // "src" | "dst" | "both"
}
#[derive(Debug, Clone, Serialize)]
pub struct MergeConflict {
pub sig_id: String,
pub kind: &'static str,
pub base: Option<String>,
pub src: Option<String>,
pub dst: Option<String>,
}
impl Store {
fn branches_dir(&self) -> PathBuf { self.root().join("branches") }
fn branch_path(&self, name: &str) -> PathBuf {
self.branches_dir().join(format!("{name}.json"))
}
fn current_branch_path(&self) -> PathBuf {
self.root().join("current_branch")
}
pub fn current_branch(&self) -> String {
match fs::read_to_string(self.current_branch_path()) {
Ok(s) => s.trim().to_string(),
Err(_) => DEFAULT_BRANCH.to_string(),
}
}
pub fn set_current_branch(&self, name: &str) -> Result<(), StoreError> {
if name != DEFAULT_BRANCH && self.get_branch(name)?.is_none() {
return Err(StoreError::UnknownBranch(name.into()));
}
fs::write(self.current_branch_path(), name)?;
Ok(())
}
pub fn list_branches(&self) -> Result<Vec<String>, StoreError> {
let mut out: Vec<String> = vec![DEFAULT_BRANCH.into()];
let dir = self.branches_dir();
if !dir.exists() { return Ok(out); }
for entry in fs::read_dir(&dir)? {
let entry = entry?;
let path = entry.path();
if path.extension().is_some_and(|e| e == "json") {
if let Some(name) = path.file_stem().and_then(|s| s.to_str()) {
if name != DEFAULT_BRANCH { out.push(name.to_string()); }
}
}
}
out.sort();
Ok(out)
}
pub fn get_branch(&self, name: &str) -> Result<Option<Branch>, StoreError> {
let path = self.branch_path(name);
if !path.exists() { return Ok(None); }
let raw = fs::read_to_string(&path)?;
let b: Branch = serde_json::from_str(&raw)?;
Ok(Some(b))
}
fn head_snapshot_path(&self, name: &str) -> PathBuf {
self.branches_dir().join(format!("{name}.head_snapshot.json"))
}
/// Best-effort read of the persisted snapshot for `name`. Any
/// failure (missing file, corrupt/partial JSON from an unclean
/// shutdown) is treated as "no snapshot" rather than an error —
/// this is a pure performance optimization, so losing it must
/// never break correctness, only fall back to a full walk.
fn load_head_snapshot(&self, name: &str) -> Option<HeadSnapshot> {
let raw = fs::read_to_string(self.head_snapshot_path(name)).ok()?;
serde_json::from_str(&raw).ok()
}
/// Best-effort write; a failure here (e.g. read-only filesystem)
/// only costs a future full walk, so it's swallowed rather than
/// propagated. Not atomic against a concurrent writer or a crash
/// mid-write — same tradeoff `set_branch_head_op`'s own
/// `fs::write` already makes for `branch_path`, and a torn write
/// just fails `load_head_snapshot`'s parse on next read.
fn save_head_snapshot(&self, name: &str, head_op: &OpId, map: &BTreeMap<String, String>) {
let snap = HeadSnapshot { head_op: head_op.clone(), map: map.clone() };
if let Ok(s) = serde_json::to_string(&snap) {
let _ = fs::write(self.head_snapshot_path(name), s);
}
}
/// Computed view: walk the op log from the branch head and
/// replay each transition into a SigId → StageId map.
///
/// Backed by a persisted snapshot (`<branch>.head_snapshot.json`)
/// keyed on the head it was computed for. Steady state — this
/// call's head_op matches the last call's — replays only the ops
/// since the snapshot instead of the whole history: O(ops since
/// the last call) instead of O(total branch history). Falls back
/// to a full walk (and refreshes the snapshot) whenever there's no
/// snapshot yet, or the snapshot's op isn't actually an ancestor
/// of the new head (a branch reset, or history reordered by a
/// merge) — see `OpLog::walk_forward_since`'s own doc comment.
///
/// This existed as a genuine, measured bottleneck before the
/// snapshot: a single call over a tenant with 110k+ accumulated
/// ops took on the order of an hour, dominated by one disk read
/// per ancestor op in the full BFS walk (alpibrusl/lex-lang#813's
/// follow-up). Every consumer that used to call this once per
/// file in a multi-file publish (fixed separately, also #813) now
/// calls it once per publish request — but "once" was still a full
/// walk over the *entire* history every time, since nothing
/// persisted the result between calls.
pub fn branch_head(&self, name: &str) -> Result<BTreeMap<String, String>, StoreError> {
let b = match self.get_branch(name)? {
Some(b) => b,
None if name == DEFAULT_BRANCH => return Ok(BTreeMap::new()),
None => return Err(StoreError::UnknownBranch(name.into())),
};
let Some(head) = b.head_op else { return Ok(BTreeMap::new()); };
let log = OpLog::open(self.root())?;
if let Some(snap) = self.load_head_snapshot(name) {
if snap.head_op == head {
return Ok(snap.map);
}
if let Some(new_records) = log.walk_forward_since(&head, &snap.head_op)? {
let mut map = snap.map;
for rec in &new_records {
apply_transition(&mut map, &rec.produces);
}
self.save_head_snapshot(name, &head, &map);
return Ok(map);
}
// Snapshot's op isn't an ancestor of the new head — fall
// through to a full walk below, which also refreshes it.
}
let mut map = BTreeMap::new();
for rec in log.walk_forward(&head, None)? {
apply_transition(&mut map, &rec.produces);
}
self.save_head_snapshot(name, &head, &map);
Ok(map)
}
pub fn branch_log(&self, name: &str) -> Result<Vec<MergeRecord>, StoreError> {
match self.get_branch(name)? {
Some(b) => Ok(b.merges),
None if name == DEFAULT_BRANCH => Ok(Vec::new()),
None => Err(StoreError::UnknownBranch(name.into())),
}
}
/// Snapshot the source branch's head_op into a new named branch.
pub fn create_branch(&self, name: &str, from: &str) -> Result<(), StoreError> {
if name.is_empty() || name.contains('/') || name.contains('\\') {
return Err(StoreError::InvalidTransition(
format!("branch name `{name}` rejected (empty or path-like)")));
}
if self.branch_path(name).exists() {
return Err(StoreError::InvalidTransition(
format!("branch `{name}` already exists")));
}
let head_op = self.get_branch(from)?.and_then(|b| b.head_op);
fs::create_dir_all(self.branches_dir())?;
let b = Branch {
name: name.into(),
parent: Some(from.into()),
head_op,
predicate: None,
merges: Vec::new(),
created_at: now(),
last_gate_checkpoint: None,
};
fs::write(self.branch_path(name), serde_json::to_string_pretty(&b)?)?;
Ok(())
}
/// Create a predicate-defined branch (#133). The branch's
/// content is the set of ops matching `predicate`; `head_op`
/// stays `None` and is materialized lazily by callers when
/// they need a single point to apply ops against. Cheap to
/// create and discard — it's a saved query, not a snapshot.
pub fn create_predicate_branch(
&self,
name: &str,
predicate: serde_json::Value,
) -> Result<(), StoreError> {
if name.is_empty() || name.contains('/') || name.contains('\\') {
return Err(StoreError::InvalidTransition(
format!("branch name `{name}` rejected (empty or path-like)")));
}
if self.branch_path(name).exists() {
return Err(StoreError::InvalidTransition(
format!("branch `{name}` already exists")));
}
fs::create_dir_all(self.branches_dir())?;
let b = Branch {
name: name.into(),
parent: None,
head_op: None,
predicate: Some(predicate),
merges: Vec::new(),
created_at: now(),
last_gate_checkpoint: None,
};
fs::write(self.branch_path(name), serde_json::to_string_pretty(&b)?)?;
Ok(())
}
pub fn delete_branch(&self, name: &str) -> Result<(), StoreError> {
if name == DEFAULT_BRANCH {
return Err(StoreError::InvalidTransition(
"cannot delete the default branch".into()));
}
if self.current_branch() == name {
return Err(StoreError::InvalidTransition(format!(
"cannot delete `{name}`; check out another branch first")));
}
let path = self.branch_path(name);
if !path.exists() {
return Err(StoreError::UnknownBranch(name.into()));
}
fs::remove_file(path)?;
Ok(())
}
/// Atomically set a branch's `head_op`. Used by `apply_operation`
/// after a successful op apply. Materializes `main`'s branch file
/// on first call (creates `branches/main.json`).
///
/// Crash safety: the tempfile's data is fsync'd before rename
/// (see `write_branch_atomic`), so a successful return implies a
/// durable branch file at the final path. The containing directory
/// is not fsync'd; on a crash between rename and the directory's
/// metadata flush, the rename can be lost — the prior head (or
/// missing branch file for a fresh `main`) survives. The op record
/// itself is content-addressed and is independently durable in the
/// op log.
///
/// Concurrency: single-writer per store. Two writers calling this
/// for the same branch race on read-modify-write of the JSON file
/// (each reads, mutates `head_op`, renames its tempfile in). Last
/// writer wins; the loser's head update is silently dropped, even
/// though both their op records survive in the op log. Tier-1
/// merge / `lex publish` callers run sequentially; multi-writer
/// safety (file locking) is on the table once `lex serve` becomes
/// a real concurrent producer (#130 territory).
pub(crate) fn set_branch_head_op(
&self,
name: &str,
head_op: OpId,
) -> Result<(), StoreError> {
let mut b = match self.get_branch(name)? {
Some(b) => b,
None if name == DEFAULT_BRANCH => Branch {
name: DEFAULT_BRANCH.into(),
parent: None,
head_op: None,
predicate: None,
merges: Vec::new(),
created_at: now(),
last_gate_checkpoint: None,
},
None => return Err(StoreError::UnknownBranch(name.into())),
};
// #256: every successful advance also moves the gate
// checkpoint to the new head. The gate that ran before this
// call has already verified everything in
// `(last_gate_checkpoint .. new_head]`, so the new head is
// now the verified frontier.
b.head_op = Some(head_op.clone());
b.last_gate_checkpoint = Some(head_op);
fs::create_dir_all(self.branches_dir())?;
write_branch_atomic(&self.branch_path(name), &b)?;
Ok(())
}
/// Atomic compare-and-swap on `branch.head_op` (#262). Holds
/// an advisory `flock` on a per-branch lock file across the
/// read-compare-write sequence so two concurrent writers can't
/// both see the same `head_op`, both decide to advance, and
/// both succeed (silently dropping one's lineage).
///
/// Returns `Ok(())` when `expected == current head_op` and the
/// new value is durably written; `Err(CasFailed { actual })`
/// when the actual head doesn't match `expected`. Callers
/// should re-read the head, rebuild the candidate op against
/// the new parent, and retry. Op records are content-addressed
/// and idempotent, so re-persisting under a new parent is safe.
///
/// Crash safety: same tempfile + rename + fsync as the
/// non-CAS path. The lock is the only addition; on a crashed
/// process the OS releases the flock and the next writer can
/// proceed.
pub(crate) fn set_branch_head_op_cas(
&self,
name: &str,
expected: Option<OpId>,
new: OpId,
) -> Result<(), CasFailed> {
// Acquire the per-branch advisory lock. Path is
// `<branches_dir>/<name>.lock`; the file is created on
// first use and reused thereafter. We hold the lock for
// the entire RMW sequence.
fs::create_dir_all(self.branches_dir())
.map_err(|e| CasFailed::Io(e.to_string()))?;
let lock_path = self.branches_dir().join(format!("{name}.lock"));
let lock_file = fs::OpenOptions::new()
.create(true)
.truncate(false)
.read(true)
.write(true)
.open(&lock_path)
.map_err(|e| CasFailed::Io(e.to_string()))?;
use fs2::FileExt;
lock_file.lock_exclusive()
.map_err(|e| CasFailed::Io(e.to_string()))?;
// Critical section: read, compare, write. The lock is
// released when `lock_file` drops at end of scope (or on
// early return).
let result = (|| -> Result<(), CasFailed> {
let actual = self.get_branch(name)
.map_err(|e| CasFailed::Io(format!("{e}")))?
.and_then(|b| b.head_op);
if actual != expected {
return Err(CasFailed::Mismatch { actual });
}
let mut b = match self.get_branch(name)
.map_err(|e| CasFailed::Io(format!("{e}")))?
{
Some(b) => b,
None if name == DEFAULT_BRANCH => Branch {
name: DEFAULT_BRANCH.into(),
parent: None,
head_op: None,
predicate: None,
merges: Vec::new(),
created_at: now(),
last_gate_checkpoint: None,
},
None => return Err(CasFailed::UnknownBranch(name.into())),
};
b.head_op = Some(new.clone());
b.last_gate_checkpoint = Some(new);
write_branch_atomic(&self.branch_path(name), &b)
.map_err(|e| CasFailed::Io(format!("{e}")))?;
Ok(())
})();
// Best-effort unlock; OS releases on file close anyway.
let _ = fs2::FileExt::unlock(&lock_file);
result
}
/// Invalidate every branch's `last_gate_checkpoint` (#256). Run
/// when a new `ProducerBlock` attestation lands so the next
/// branch advance walks back from genesis once and re-verifies
/// the full chain. Returns the number of branches whose
/// checkpoint changed.
pub fn invalidate_gate_checkpoints(&self) -> Result<usize, StoreError> {
let dir = self.branches_dir();
if !dir.exists() {
return Ok(0);
}
let mut updated = 0usize;
for entry in fs::read_dir(&dir)? {
let entry = entry?;
let path = entry.path();
if path.extension().is_none_or(|e| e != "json") { continue; }
let bytes = fs::read(&path)?;
let mut b: Branch = match serde_json::from_slice(&bytes) {
Ok(b) => b,
// Corrupt branch file shouldn't take down the
// invalidation pass; the next gate run will surface
// the parse error on a real call path.
Err(_) => continue,
};
if b.last_gate_checkpoint.is_some() {
b.last_gate_checkpoint = None;
write_branch_atomic(&path, &b)?;
updated += 1;
}
}
Ok(updated)
}
}
/// Apply a single `StageTransition` to a sig-stage map. Used by
/// `branch_head` to replay an op log.
pub(crate) fn apply_transition(map: &mut BTreeMap<String, String>, t: &StageTransition) {
match t {
StageTransition::Create { sig_id, stage_id }
| StageTransition::Replace { sig_id, to: stage_id, .. } => {
map.insert(sig_id.clone(), stage_id.clone());
}
StageTransition::Remove { sig_id, .. } => {
map.remove(sig_id);
}
StageTransition::Rename { from, to, body_stage_id } => {
map.remove(from);
map.insert(to.clone(), body_stage_id.clone());
}
StageTransition::ImportOnly => {}
StageTransition::Merge { entries } => {
for (sig, stage) in entries {
match stage {
Some(s) => { map.insert(sig.clone(), s.clone()); }
None => { map.remove(sig); }
}
}
}
}
}
fn write_branch_atomic(path: &std::path::Path, b: &Branch) -> Result<(), StoreError> {
use std::io::Write;
let bytes = serde_json::to_vec_pretty(b)?;
let tmp = path.with_extension("json.tmp");
let mut f = fs::File::create(&tmp)?;
f.write_all(&bytes)?;
f.sync_all()?;
fs::rename(&tmp, path)?;
Ok(())
}
fn now() -> u64 {
use std::time::{SystemTime, UNIX_EPOCH};
SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0)
}
impl Store {
pub fn merge(&self, src: &str, dst: &str) -> Result<MergeReport, StoreError> {
let log = OpLog::open(self.root())?;
let src_head = self.get_branch(src)?.and_then(|b| b.head_op);
let dst_head = match self.get_branch(dst)? {
Some(b) => b.head_op,
None if dst == DEFAULT_BRANCH => None,
None => return Err(StoreError::UnknownBranch(dst.into())),
};
let out = lex_vcs::merge(&log, src_head.as_ref(), dst_head.as_ref())?;
let mut report = MergeReport {
summary: MergeSummary {
base: out.lca.clone(),
src: src.into(),
dst: dst.into(),
..Default::default()
},
merged: Vec::new(),
conflicts: Vec::new(),
};
for o in out.outcomes {
match o {
lex_vcs::MergeOutcome::Both { sig_id, stage_id } => {
if let Some(stage_id) = stage_id {
report.merged.push(MergeEntry { sig_id, stage_id, from: "both" });
}
}
lex_vcs::MergeOutcome::Src { sig_id, stage_id } => {
if let Some(stage_id) = stage_id {
report.merged.push(MergeEntry { sig_id, stage_id, from: "src" });
}
}
lex_vcs::MergeOutcome::Dst { sig_id, stage_id } => {
if let Some(stage_id) = stage_id {
report.merged.push(MergeEntry { sig_id, stage_id, from: "dst" });
}
}
lex_vcs::MergeOutcome::Conflict { sig_id, kind, base, src, dst } => {
let kind: &'static str = match kind {
lex_vcs::ConflictKind::ModifyModify => "modify-modify",
lex_vcs::ConflictKind::ModifyDelete => "modify-delete",
lex_vcs::ConflictKind::DeleteModify => "delete-modify",
lex_vcs::ConflictKind::AddAdd => "add-add",
};
report.conflicts.push(MergeConflict {
sig_id, kind, base, src, dst,
});
}
}
}
report.summary.clean = report.merged.len();
report.summary.conflicts = report.conflicts.len();
report.summary.total_sigs = report.merged.len() + report.conflicts.len();
Ok(report)
}
pub fn commit_merge(&self, dst: &str, report: &MergeReport) -> Result<(), StoreError> {
if !report.conflicts.is_empty() {
return Err(StoreError::InvalidTransition(format!(
"{} conflicts; resolve before committing", report.conflicts.len())));
}
let dst_head_map = self.branch_head(dst)?;
let mut entries: BTreeMap<String, Option<String>> = BTreeMap::new();
for m in &report.merged {
let cur = dst_head_map.get(&m.sig_id);
if cur != Some(&m.stage_id) {
entries.insert(m.sig_id.clone(), Some(m.stage_id.clone()));
}
}
let src_head = self.get_branch(&report.summary.src)?.and_then(|b| b.head_op);
let dst_head_op = self.get_branch(dst)?.and_then(|b| b.head_op);
match (src_head.clone(), dst_head_op.clone()) {
// Fast-forward: dst is empty, just adopt src's head.
(Some(s), None) => {
self.set_branch_head_op(dst, s)?;
}
// Both sides have heads at the same op: nothing structural
// to merge. Skip apply but still journal below.
(Some(s), Some(d)) if s == d => { /* no-op */ }
(Some(s), Some(d)) => {
let op = lex_vcs::Operation::new(
lex_vcs::OperationKind::Merge { resolved: entries.len() },
[s, d],
);
let t = lex_vcs::StageTransition::Merge { entries };
// Gated (#833): land the merge op, type-check the real
// post-merge head, and roll back if it doesn't compose.
let _ = self.apply_merge_op_gated(dst, op, t)?;
}
// src empty: nothing to merge in. Treat as no-op.
(None, _) => { /* no-op */ }
}
// Atomicity note: the merge op is durable after apply_operation
// returns; the journal entry below is a separate write. A
// crash between leaves the merge in the op DAG but no journal
// row — `lex log` will be missing this merge. The branch is
// still functionally correct (head_op points at the merge op,
// which carries `entries`), so the gap is recoverable by
// re-running commit_merge once (which will journal but skip
// the apply on the same-head match arm above). Tier-1 single-
// writer assumption applies; multi-writer locking is on the
// table for #130.
// Journal the merge so `lex log` can show it.
let mut b = self.get_branch(dst)?
.ok_or_else(|| StoreError::UnknownBranch(dst.into()))?;
if !report.summary.src.is_empty() {
b.merges.push(MergeRecord {
src: report.summary.src.clone(),
at: now(),
merged: report.merged.len(),
conflicts: 0,
});
write_branch_atomic(&self.branch_path(dst), &b)?;
}
Ok(())
}
}
#[cfg(test)]
mod branch_head_snapshot_tests {
use super::*;
use lex_vcs::{Operation, OperationKind};
use std::collections::BTreeSet;
fn add(store: &Store, sig: &str, stg: &str) -> OpId {
let parent = store.get_branch(DEFAULT_BRANCH).unwrap().and_then(|b| b.head_op);
let op = Operation::new(
OperationKind::AddFunction {
sig_id: sig.into(),
stage_id: stg.into(),
effects: BTreeSet::new(),
budget_cost: None,
},
parent.into_iter().collect::<Vec<_>>(),
);
let transition = StageTransition::Create { sig_id: sig.into(), stage_id: stg.into() };
store.apply_operation(DEFAULT_BRANCH, op, transition).unwrap()
}
/// The fallback this exercises can't be reached through the public
/// API alone: `apply_operation`'s CAS retry always rebuilds a
/// single-parent op's `parents` to match the *current* head, so
/// there is no ordinary way to advance a branch to an op that
/// doesn't descend from its own history. `set_branch_head_op`
/// (crate-internal) is what a real reset/rebase operation would
/// eventually call, so this directly forces that same shape: a
/// head whose ancestry does NOT include the op the persisted
/// snapshot was computed for.
#[test]
fn branch_head_falls_back_to_full_walk_when_snapshot_predates_a_reset() {
let tmp = tempfile::tempdir().unwrap();
let store = Store::open(tmp.path()).unwrap();
add(&store, "fn::a", "stage_a");
add(&store, "fn::b", "stage_b");
let snapshotted = store.branch_head(DEFAULT_BRANCH).unwrap();
assert_eq!(snapshotted.len(), 2, "sanity: snapshot covers both ops");
// Force the branch onto a disconnected, single-op history —
// the snapshot's op is not among its ancestors.
let reset_op = Operation::new(
OperationKind::AddFunction {
sig_id: "fn::reset_only".into(),
stage_id: "stage_reset".into(),
effects: BTreeSet::new(),
budget_cost: None,
},
Vec::new(), // no parents: a fresh root, unrelated to fn::a/fn::b
);
let reset_op_id = reset_op.op_id();
let reset_record = lex_vcs::OperationRecord::new(
reset_op,
StageTransition::Create {
sig_id: "fn::reset_only".into(),
stage_id: "stage_reset".into(),
},
);
let log = OpLog::open(store.root()).unwrap();
log.put(&reset_record).unwrap();
store.set_branch_head_op(DEFAULT_BRANCH, reset_op_id).unwrap();
let after_reset = store.branch_head(DEFAULT_BRANCH).unwrap();
assert_eq!(
after_reset.len(), 1,
"stale snapshot must not be reused across a non-ancestor head change: {after_reset:?}"
);
assert_eq!(after_reset.get("fn::reset_only"), Some(&"stage_reset".to_string()));
assert!(!after_reset.contains_key("fn::a"));
assert!(!after_reset.contains_key("fn::b"));
// A repeat call must now hit the (correctly refreshed) snapshot
// and still agree.
let again = store.branch_head(DEFAULT_BRANCH).unwrap();
assert_eq!(after_reset, again);
}
}