choir_spike/
main.rs

1//! Phase-0 integrated gate spike (DECISIONS.md).
2//!
3//! End to end on one machine: provision N workspaces from a base tree via
4//! CoW clone (macOS `clonefile(2)` here; btrfs/ZFS on the Linux target),
5//! have every workspace submit ops through the single-writer sequencer, then
6//! push all concurrent edits of a shared file through the merge pipeline.
7//! Prints the gate report: provisioning percentiles, decision-latency
8//! percentiles, and merge outcomes (clean vs first-class conflict).
9//!
10//! # Where this sits
11//!
12//! `docs/architecture.md` is the map of the whole workspace.
13//! This crate is the Phase-0 gate binary: provisioning and decision-latency percentiles, and merge outcomes.
14//!
15//! It builds on [`choir_merge`], [`choir_oplog`] and [`choir_sequencer`].
16
17use choir_merge::{MergeOutcome, Pipeline};
18use choir_oplog::MemLog;
19use choir_sequencer::Sequencer;
20use std::path::Path;
21use std::time::{Duration, Instant};
22
23const WORKSPACES: usize = 16;
24const FILES_IN_BASE: usize = 500;
25
26#[cfg(target_os = "macos")]
27fn cow_clone(src: &Path, dst: &Path) -> std::io::Result<()> {
28    use std::os::raw::{c_char, c_int, c_uint};
29    extern "C" {
30        fn clonefile(src: *const c_char, dst: *const c_char, flags: c_uint) -> c_int;
31    }
32    let s = std::ffi::CString::new(src.as_os_str().as_encoded_bytes()).unwrap();
33    let d = std::ffi::CString::new(dst.as_os_str().as_encoded_bytes()).unwrap();
34    if unsafe { clonefile(s.as_ptr(), d.as_ptr(), 0) } == 0 {
35        Ok(())
36    } else {
37        Err(std::io::Error::last_os_error())
38    }
39}
40
41#[cfg(not(target_os = "macos"))]
42fn cow_clone(src: &Path, dst: &Path) -> std::io::Result<()> {
43    // Portability fallback only; the Linux target uses btrfs/ZFS reflinks.
44    fn copy_dir(src: &Path, dst: &Path) -> std::io::Result<()> {
45        std::fs::create_dir_all(dst)?;
46        for e in std::fs::read_dir(src)? {
47            let e = e?;
48            let to = dst.join(e.file_name());
49            if e.file_type()?.is_dir() {
50                copy_dir(&e.path(), &to)?;
51            } else {
52                std::fs::copy(e.path(), to)?;
53            }
54        }
55        Ok(())
56    }
57    copy_dir(src, dst)
58}
59
60fn percentile(sorted: &[Duration], q: f64) -> Duration {
61    sorted[((sorted.len() - 1) as f64 * q) as usize]
62}
63
64fn main() {
65    let work = std::env::temp_dir().join(format!("choir-spike-{}", std::process::id()));
66    let base = work.join("base");
67    std::fs::create_dir_all(&base).unwrap();
68
69    // shared.txt: widely spaced distinct regions, one per workspace (edits
70    // must fold cleanly). hot.txt: a single line every workspace edits
71    // (merges must refuse to resolve silently).
72    let mut shared = Vec::new();
73    for i in 0..(WORKSPACES * 4) {
74        shared.push(format!("region {i}"));
75    }
76    let shared_base = shared.join("\n") + "\n";
77    std::fs::write(base.join("shared.txt"), &shared_base).unwrap();
78    let hot_base = "hot line\n".to_string();
79    std::fs::write(base.join("hot.txt"), &hot_base).unwrap();
80    for i in 0..FILES_IN_BASE {
81        let d = base.join(format!("dir{}", i % 50));
82        std::fs::create_dir_all(&d).unwrap();
83        std::fs::write(d.join(format!("f{i}.txt")), format!("file {i}\n")).unwrap();
84    }
85
86    // 1) Provision N workspaces via CoW clone.
87    let mut provision_times = Vec::new();
88    let mut ws_paths = Vec::new();
89    for w in 0..WORKSPACES {
90        let dst = work.join(format!("ws{w}"));
91        let t = Instant::now();
92        cow_clone(&base, &dst).expect("workspace provisioning");
93        provision_times.push(t.elapsed());
94        ws_paths.push(dst);
95    }
96    provision_times.sort();
97
98    // 2) Every workspace edits its own region of shared.txt and the single
99    //    line of hot.txt, then submits an op through the sequencer,
100    //    concurrently.
101    let sequencer = Sequencer::spawn(Box::new(MemLog::new()));
102    let mut threads = Vec::new();
103    for (w, path) in ws_paths.iter().enumerate() {
104        let handle = sequencer.handle();
105        let path = path.clone();
106        let base_text = shared_base.clone();
107        threads.push(std::thread::spawn(move || {
108            let mut lines: Vec<String> = base_text.lines().map(String::from).collect();
109            lines[w * 4 + 2] = format!("region {} edited by ws{w}", w * 4 + 2);
110            let shared_edit = lines.join("\n") + "\n";
111            let hot_edit = format!("hot line edited by ws{w}\n");
112            std::fs::write(path.join("shared.txt"), &shared_edit).unwrap();
113            std::fs::write(path.join("hot.txt"), &hot_edit).unwrap();
114            let accepted = handle.submit(&format!("ws{w}"), shared_edit.clone().into_bytes());
115            (shared_edit, hot_edit, accepted.decision_latency)
116        }));
117    }
118    let mut shared_edits = Vec::new();
119    let mut hot_edits = Vec::new();
120    let mut latencies = Vec::new();
121    for t in threads {
122        let (shared_edit, hot_edit, lat) = t.join().unwrap();
123        shared_edits.push(shared_edit);
124        hot_edits.push(hot_edit);
125        latencies.push(lat);
126    }
127    latencies.sort();
128    let log = sequencer.shutdown();
129    assert_eq!(log.len(), WORKSPACES as u64, "zero op loss");
130
131    // 3a) Disjoint-region edits must fold cleanly through the pipeline.
132    let pipeline = Pipeline::default_v1();
133    let mut acc = shared_edits[0].clone();
134    let mut clean = 0usize;
135    let mut wrong_conflicts = 0usize;
136    for edit in &shared_edits[1..] {
137        match pipeline.merge(&shared_base, &acc, edit).outcome {
138            MergeOutcome::Resolved(merged) => {
139                clean += 1;
140                acc = merged;
141            }
142            MergeOutcome::Conflict { .. } => wrong_conflicts += 1,
143            MergeOutcome::Unavailable(_) => unreachable!(),
144        }
145    }
146    let regions_merged = (0..WORKSPACES)
147        .filter(|w| acc.contains(&format!("edited by ws{w}")))
148        .count();
149
150    // 3b) Same-line edits must surface as first-class conflicts, never a
151    //     silent pick.
152    let mut hot_conflicts = 0usize;
153    let mut silent_hot_merges = 0usize;
154    for pair in hot_edits.windows(2) {
155        match pipeline.merge(&hot_base, &pair[0], &pair[1]).outcome {
156            MergeOutcome::Conflict { .. } => hot_conflicts += 1,
157            MergeOutcome::Resolved(_) => silent_hot_merges += 1,
158            MergeOutcome::Unavailable(_) => unreachable!(),
159        }
160    }
161
162    std::fs::remove_dir_all(&work).ok();
163
164    println!("== Phase-0 integrated gate report ==");
165    println!(
166        "workspaces: {WORKSPACES} (gate: >10) | base tree: {FILES_IN_BASE} files + shared.txt"
167    );
168    println!(
169        "provisioning (CoW clone): p50={:?} p90={:?} max={:?} (target p50 < 50 ms warm)",
170        percentile(&provision_times, 0.5),
171        percentile(&provision_times, 0.9),
172        provision_times.last().unwrap()
173    );
174    println!(
175        "sequencer decision latency: p50={:?} p99={:?} (gate: < 100 ms, CI excluded)",
176        percentile(&latencies, 0.5),
177        percentile(&latencies, 0.99)
178    );
179    println!(
180        "disjoint-region fold: {clean}/{} clean merges, {regions_merged}/{WORKSPACES} regions present, {wrong_conflicts} spurious conflicts",
181        WORKSPACES - 1
182    );
183    println!(
184        "same-line (hot) merges: {hot_conflicts}/{} first-class conflicts, {silent_hot_merges} silent picks (must be 0)",
185        WORKSPACES - 1
186    );
187
188    let pass = WORKSPACES > 10
189        && percentile(&provision_times, 0.5) < Duration::from_millis(50)
190        && percentile(&latencies, 0.99) < Duration::from_millis(100)
191        && wrong_conflicts == 0
192        && regions_merged == WORKSPACES
193        && silent_hot_merges == 0;
194    println!(
195        "GATE ({} dev-machine): {}",
196        std::env::consts::OS,
197        if pass { "PASS" } else { "FAIL" }
198    );
199    std::process::exit(if pass { 0 } else { 1 });
200}
choir_spike/main.rs

choir_spike/
main.rs
