alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::path::Path;
2use std::process::{Command, Stdio};
3use std::time::{Duration, Instant};
4
5use knot_git::Layout;
6use knot_types::{Oid, RefName, RepoDid};
7
8mod common;
9use common::{must, pack_objects, receive_request};
10
11const DID: &str = "did:plc:squid";
12const BLOB_BYTES: usize = 32 * 1024 * 1024;
13
14fn incompressible(len: usize) -> Vec<u8> {
15 let mut state = 0x2545_f491_4f6c_dd1du64;
16 (0..len)
17 .map(|_| {
18 state ^= state << 13;
19 state ^= state >> 7;
20 state ^= state << 17;
21 (state & 0xff) as u8
22 })
23 .collect()
24}
25
26struct Seed {
27 c1: String,
28 c2: String,
29 request: std::path::PathBuf,
30}
31
32fn build_seed(scratch: &Path) -> Seed {
33 let work = scratch.join("work");
34 std::fs::create_dir_all(&work).unwrap();
35 must(&work, &["init", "-q", "-b", "main"]);
36 std::fs::write(work.join("base.txt"), "baseline\n").unwrap();
37 must(&work, &["add", "-A"]);
38 must(&work, &["commit", "-q", "-m", "c1"]);
39 let c1 = must(&work, &["rev-parse", "HEAD"]);
40 std::fs::write(work.join("big.bin"), incompressible(BLOB_BYTES)).unwrap();
41 must(&work, &["add", "-A"]);
42 must(&work, &["commit", "-q", "-m", "c2"]);
43 let c2 = must(&work, &["rev-parse", "HEAD"]);
44
45 let oids: Vec<String> = must(&work, &["rev-list", "--objects", &c2, "--not", &c1])
46 .lines()
47 .map(|line| line.split_whitespace().next().unwrap().to_string())
48 .collect();
49 let pack = pack_objects(&work, &oids);
50 let request = scratch.join("c2.request");
51 std::fs::write(
52 &request,
53 receive_request("refs/heads/main", &c1, &c2, &pack),
54 )
55 .unwrap();
56 Seed { c1, c2, request }
57}
58
59fn fresh_repo(scratch: &Path, trial: usize, seed: &Seed) -> std::path::PathBuf {
60 let scan = scratch.join(format!("scan-{trial}"));
61 let layout = Layout::new(&scan);
62 let did = RepoDid::new(DID).unwrap();
63 layout.create(&did).unwrap();
64 let bare = layout.repo_path(&did).unwrap();
65 must(
66 scratch.join("work").as_path(),
67 &[
68 "push",
69 "-q",
70 bare.to_str().unwrap(),
71 &format!("{}:refs/heads/main", seed.c1),
72 ],
73 );
74 scan
75}
76
77fn spawn_worker(scan: &Path, request: &Path) -> std::process::Child {
78 Command::new(std::env::current_exe().unwrap())
79 .args(["--exact", "chaos_receive_worker", "--nocapture"])
80 .env("KNOT_CHAOS_ROLE", "worker")
81 .env("KNOT_CHAOS_SCAN", scan)
82 .env("KNOT_CHAOS_REQUEST", request)
83 .stdout(Stdio::null())
84 .stderr(Stdio::null())
85 .spawn()
86 .expect("spawn chaos worker")
87}
88
89fn fsck_clean(bare: &Path) -> Result<(), String> {
90 knot_fixtures::fsck(bare)
91}
92
93fn main_tip(scan: &Path) -> Option<Oid> {
94 let layout = Layout::new(scan);
95 let repo = layout
96 .open(&RepoDid::new(DID).unwrap())
97 .expect("repo must reopen cleanly after a kill");
98 repo.find_ref(&RefName::new("refs/heads/main").unwrap())
99 .expect("references must be readable after a kill")
100}
101
102#[test]
103fn chaos_receive_worker() {
104 if std::env::var("KNOT_CHAOS_ROLE").as_deref() != Ok("worker") {
105 return;
106 }
107 let scan = std::env::var("KNOT_CHAOS_SCAN").unwrap();
108 let request = std::env::var("KNOT_CHAOS_REQUEST").unwrap();
109 let layout = Layout::new(&scan);
110 let repo = layout.open(&RepoDid::new(DID).unwrap()).unwrap();
111 let body = std::fs::read(&request).unwrap();
112 let _ = knot_pack::receive_pack(&repo, &body);
113}
114
115#[test]
116fn kill9_during_receive_pack_leaves_a_consistent_repo() {
117 let scratch = tempfile::tempdir().unwrap();
118 let seed = build_seed(scratch.path());
119 let c1 = Oid::from_hex(&seed.c1).unwrap();
120 let c2 = Oid::from_hex(&seed.c2).unwrap();
121 let did = RepoDid::new(DID).unwrap();
122
123 let warm_scan = fresh_repo(scratch.path(), 9000, &seed);
124 let started = Instant::now();
125 let mut warm = spawn_worker(&warm_scan, &seed.request);
126 warm.wait().unwrap();
127 let full = started.elapsed();
128 assert_eq!(
129 main_tip(&warm_scan),
130 Some(c2),
131 "uninterrupted receive must fast-forward main to new tip"
132 );
133
134 let fractions = [0.30, 0.45, 0.55, 0.62, 0.70, 0.78, 0.85, 0.92, 1.05, 1.25];
135 let delays: Vec<Duration> = std::iter::once(Duration::from_millis(2))
136 .chain(std::iter::once(Duration::from_millis(5)))
137 .chain(fractions.iter().map(|fraction| full.mul_f64(*fraction)))
138 .chain(std::iter::once(full.mul_f64(2.0)))
139 .chain(std::iter::once(full.mul_f64(2.0)))
140 .collect();
141
142 let outcomes: Vec<Oid> = delays
143 .iter()
144 .enumerate()
145 .map(|(trial, delay)| {
146 let scan = fresh_repo(scratch.path(), trial, &seed);
147 let mut child = spawn_worker(&scan, &seed.request);
148 std::thread::sleep(*delay);
149 let _ = child.kill();
150 child.wait().unwrap();
151
152 let bare = Layout::new(&scan).repo_path(&did).unwrap();
153 fsck_clean(&bare).unwrap_or_else(|errors| {
154 panic!("trial {trial}: killed receive left a corrupt repo:\n{errors}")
155 });
156 let tip = main_tip(&scan).unwrap_or_else(|| {
157 panic!("trial {trial}: main vanished after a kill, acknowledged ref was lost")
158 });
159 assert!(
160 tip == c1 || tip == c2,
161 "trial {trial}: main must hold either acknowledged baseline or completed tip, never a torn value, got {tip}"
162 );
163 tip
164 })
165 .collect();
166
167 assert!(
168 outcomes.contains(&c1),
169 "no trial was interrupted before ref update; chaos window never opened"
170 );
171 assert!(
172 outcomes.contains(&c2),
173 "no trial ran to completion; receive never finished under chosen delays"
174 );
175}