alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1mod common;
2
3use std::path::Path;
4use std::process::{Command, Stdio};
5use std::time::{Duration, Instant, SystemTime};
6
7use common::{backdate, incompressible, object_path, oid_of, pointer_blob};
8use knot_git::{EntryKind, Identity, Layout, NewCommit, RefUpdate, StagedAction, StagedChange};
9use knot_lfs::{ClaimedSize, DiskStore, LfsOid, LfsSize, LfsStore, LfsStorePath, collect_repo};
10use knot_types::{AuthorName, BranchName, Email, Oid, RefName, RepoDid, UnixSeconds};
11
12const DID: &str = "did:plc:squid";
13const EMPTY_TREE: &str = "4b825dc642cb6eb9a060e54bf8d69288fbee4904";
14const PUT_BYTES: usize = 48 * 1024 * 1024;
15const REFERENCED: usize = 24;
16const UNREFERENCED: usize = 320;
17const GRACE: Duration = Duration::from_secs(86_400);
18const BACKDATE: Duration = Duration::from_secs(60 * 86_400);
19
20fn did() -> RepoDid {
21 RepoDid::new(DID).unwrap()
22}
23
24fn spawn_worker(role: &str, envs: &[(&str, &Path)]) -> std::process::Child {
25 let mut command = Command::new(std::env::current_exe().unwrap());
26 command
27 .args(["--exact", &format!("chaos_{role}_worker"), "--nocapture"])
28 .env("KNOT_CHAOS_ROLE", role)
29 .stdout(Stdio::null())
30 .stderr(Stdio::null());
31 envs.iter().for_each(|(key, value)| {
32 command.env(key, value);
33 });
34 command.spawn().expect("spawn chaos worker")
35}
36
37fn kill_after(mut child: std::process::Child, delay: Duration) {
38 std::thread::sleep(delay);
39 let _ = child.kill();
40 child.wait().unwrap();
41}
42
43fn delays(full: Duration) -> Vec<Duration> {
44 let fractions = [0.20, 0.35, 0.45, 0.55, 0.65, 0.75, 0.85, 0.95];
45 std::iter::once(Duration::from_millis(1))
46 .chain(std::iter::once(Duration::from_millis(3)))
47 .chain(fractions.iter().map(|fraction| full.mul_f64(*fraction)))
48 .chain(std::iter::once(full.mul_f64(2.0)))
49 .collect()
50}
51
52#[test]
53fn chaos_put_worker() {
54 if std::env::var("KNOT_CHAOS_ROLE").as_deref() != Ok("put") {
55 return;
56 }
57 let store_dir = std::env::var("KNOT_CHAOS_STORE").unwrap();
58 let store = DiskStore::open(LfsStorePath::new(&store_dir)).unwrap();
59 let body = incompressible(PUT_BYTES, 0x2545_f491_4f6c_dd1d);
60 let oid = oid_of(&body);
61 let _ = store.put(
62 &did(),
63 &oid,
64 ClaimedSize::new(body.len() as u64),
65 &mut &body[..],
66 );
67}
68
69#[test]
70fn kill9_during_put_object_never_leaves_a_torn_object() {
71 let body = incompressible(PUT_BYTES, 0x2545_f491_4f6c_dd1d);
72 let oid = oid_of(&body);
73 let scratch = tempfile::tempdir().unwrap();
74
75 let warm = scratch.path().join("warm");
76 std::fs::create_dir_all(&warm).unwrap();
77 let started = Instant::now();
78 spawn_worker("put", &[("KNOT_CHAOS_STORE", &warm)])
79 .wait()
80 .unwrap();
81 let full = started.elapsed();
82 assert!(
83 object_path(&warm, &oid).is_file(),
84 "an uninterrupted put stores the object"
85 );
86
87 let landed: Vec<bool> = delays(full)
88 .iter()
89 .enumerate()
90 .map(|(trial, delay)| {
91 let store_dir = scratch.path().join(format!("store-{trial}"));
92 std::fs::create_dir_all(&store_dir).unwrap();
93 kill_after(
94 spawn_worker("put", &[("KNOT_CHAOS_STORE", &store_dir)]),
95 *delay,
96 );
97
98 let final_path = object_path(&store_dir, &oid);
99 let present = final_path.is_file();
100 if present {
101 let bytes = std::fs::read(&final_path).unwrap();
102 assert_eq!(
103 bytes.len(),
104 PUT_BYTES,
105 "trial {trial}: a visible object is never truncated"
106 );
107 assert_eq!(
108 oid_of(&bytes),
109 oid,
110 "trial {trial}: a visible object always hashes to its oid"
111 );
112 }
113
114 let store = DiskStore::open(LfsStorePath::new(&store_dir)).unwrap();
115 let incoming: Vec<_> = std::fs::read_dir(store_dir.join(".incoming"))
116 .unwrap()
117 .collect();
118 assert!(
119 incoming.is_empty(),
120 "trial {trial}: boot sweep clears abandoned uploads, found {incoming:?}"
121 );
122 assert_eq!(
123 store.probe(&did(), &oid).unwrap().is_some(),
124 present,
125 "trial {trial}: the boot sweep never deletes a stored object"
126 );
127 present
128 })
129 .collect();
130
131 assert!(
132 landed.iter().any(|present| !present),
133 "some trial must be killed before the rename, or the kill delays are all too long"
134 );
135 assert!(
136 landed.iter().any(|present| *present),
137 "some trial must complete, or the kill delays are all too short"
138 );
139}
140
141fn identity() -> Identity {
142 Identity {
143 name: AuthorName::new("nel"),
144 email: Email::new("nel@oyster.cafe"),
145 time: UnixSeconds::new(1_700_000_000),
146 offset_seconds: 0,
147 }
148}
149
150struct GcFixture {
151 referenced: Vec<(LfsOid, Vec<u8>)>,
152 unreferenced: Vec<LfsOid>,
153}
154
155fn build_gc_fixture(scan: &Path, store_dir: &Path) -> GcFixture {
156 let layout = Layout::new(scan).with_default_branch(BranchName::new("main").unwrap());
157 let repo = layout.create(&did()).unwrap();
158 let store = DiskStore::open(LfsStorePath::new(store_dir)).unwrap();
159
160 let referenced: Vec<(LfsOid, Vec<u8>)> = (0..REFERENCED)
161 .map(|index| {
162 let body = incompressible(2048, 0x9e37_79b9_7f4a_7c15 ^ index as u64);
163 let oid = oid_of(&body);
164 store
165 .put(
166 &did(),
167 &oid,
168 ClaimedSize::new(body.len() as u64),
169 &mut &body[..],
170 )
171 .unwrap();
172 backdate(&object_path(store_dir, &oid), BACKDATE);
173 (oid, body)
174 })
175 .collect();
176
177 let unreferenced: Vec<LfsOid> = (0..UNREFERENCED)
178 .map(|index| {
179 let body = incompressible(512, 0xdead_beef_cafe_f00d ^ index as u64);
180 let oid = oid_of(&body);
181 store
182 .put(
183 &did(),
184 &oid,
185 ClaimedSize::new(body.len() as u64),
186 &mut &body[..],
187 )
188 .unwrap();
189 backdate(&object_path(store_dir, &oid), BACKDATE);
190 oid
191 })
192 .collect();
193
194 let changes: Vec<StagedChange> = referenced
195 .iter()
196 .enumerate()
197 .map(|(index, (oid, body))| StagedChange {
198 path: knot_types::RepoPath::new(format!("media/clip{index}.bin")).unwrap(),
199 action: StagedAction::Put {
200 content: pointer_blob(oid, LfsSize::new(body.len() as u64)),
201 kind: EntryKind::Blob,
202 },
203 })
204 .collect();
205 let empty = Oid::from_hex(EMPTY_TREE).unwrap();
206 let tree = repo.write_staged_tree(empty, &changes).unwrap();
207 let tip = repo
208 .write_commit(&NewCommit {
209 tree,
210 parents: Vec::new(),
211 author: identity(),
212 committer: identity(),
213 message: "add media".to_string(),
214 extra_headers: Vec::new(),
215 })
216 .unwrap();
217 repo.update_ref(&RefUpdate::Create {
218 name: RefName::new("refs/heads/main").unwrap(),
219 new: tip,
220 })
221 .unwrap();
222
223 GcFixture {
224 referenced,
225 unreferenced,
226 }
227}
228
229fn run_gc(scan: &Path, store_dir: &Path) {
230 let layout = Layout::new(scan);
231 let repo = layout.open(&did()).unwrap();
232 let store = DiskStore::open(LfsStorePath::new(store_dir)).unwrap();
233 let _ = collect_repo(&store, &repo, &did(), GRACE, SystemTime::now());
234}
235
236#[test]
237fn chaos_gc_worker() {
238 if std::env::var("KNOT_CHAOS_ROLE").as_deref() != Ok("gc") {
239 return;
240 }
241 let scan = std::env::var("KNOT_CHAOS_SCAN").unwrap();
242 let store_dir = std::env::var("KNOT_CHAOS_STORE").unwrap();
243 run_gc(Path::new(&scan), Path::new(&store_dir));
244}
245
246#[test]
247fn kill9_during_gc_never_loses_a_referenced_object() {
248 let scratch = tempfile::tempdir().unwrap();
249
250 let warm_scan = scratch.path().join("warm-scan");
251 let warm_store = scratch.path().join("warm-store");
252 let warm_fixture = build_gc_fixture(&warm_scan, &warm_store);
253 let started = Instant::now();
254 spawn_worker(
255 "gc",
256 &[
257 ("KNOT_CHAOS_SCAN", &warm_scan),
258 ("KNOT_CHAOS_STORE", &warm_store),
259 ],
260 )
261 .wait()
262 .unwrap();
263 let full = started.elapsed();
264 let warm_disk = DiskStore::open(LfsStorePath::new(&warm_store)).unwrap();
265 warm_fixture.referenced.iter().for_each(|(oid, _)| {
266 assert!(
267 warm_disk.probe(&did(), oid).unwrap().is_some(),
268 "an uninterrupted gc keeps every referenced object"
269 );
270 });
271 warm_fixture.unreferenced.iter().for_each(|oid| {
272 assert_eq!(
273 warm_disk.probe(&did(), oid).unwrap(),
274 None,
275 "an uninterrupted gc reclaims every expired orphan"
276 );
277 });
278
279 delays(full).iter().enumerate().for_each(|(trial, delay)| {
280 let scan = scratch.path().join(format!("scan-{trial}"));
281 let store_dir = scratch.path().join(format!("store-{trial}"));
282 let fixture = build_gc_fixture(&scan, &store_dir);
283 kill_after(
284 spawn_worker(
285 "gc",
286 &[("KNOT_CHAOS_SCAN", &scan), ("KNOT_CHAOS_STORE", &store_dir)],
287 ),
288 *delay,
289 );
290
291 let store = DiskStore::open(LfsStorePath::new(&store_dir)).unwrap();
292 fixture.referenced.iter().for_each(|(oid, body)| {
293 assert_eq!(
294 store.probe(&did(), oid).unwrap(),
295 Some(LfsSize::new(body.len() as u64)),
296 "trial {trial}: a referenced object remains stored after a killed sweep"
297 );
298 });
299
300 run_gc(&scan, &store_dir);
301 fixture.referenced.iter().for_each(|(oid, _)| {
302 assert!(
303 store.probe(&did(), oid).unwrap().is_some(),
304 "trial {trial}: a referenced object remains stored after the self-heal pass"
305 );
306 });
307 fixture.unreferenced.iter().for_each(|oid| {
308 assert_eq!(
309 store.probe(&did(), oid).unwrap(),
310 None,
311 "trial {trial}: the self-heal pass finishes the interrupted reclaim"
312 );
313 });
314 });
315}