Oregami
Repositories/oxedyne/fe2o3

oxedyne/fe2o3/fe2o3_o3db_sync/tests/sweep.rs

23.5 KiB, 3 runs

created by r1870400018:38133, which is this file's identity for as long as the history lasts, whatever it is later renamed to

download · who wrote it · its history

1//! The online orphan sweep must reclaim orphaned chunk records, leave every live value untouched,
2//! and stay safe against a concurrent writer.
3//!
4//! A chunked value's chunk records are keyed by its geometry, so an overwrite that changes the
5//! geometry -- or, for a value written under a pre-fix random ticket, any overwrite -- leaves the old
6//! chunk records under keys no live bunch key names. Supersession-based collection never reaches
7//! them: they are orphans. `sweep_orphans` tombstones each orphaned chunk key so the running
8//! collector reclaims its bytes, guarding against concurrent writers with an epoch on the record
9//! timestamp.
10//!
11//! `sweep_reclaims_orphans` seeds a store with random-ticket orphans, geometry-change orphans, dead
12//! tombstones, a live chunked value and live unchunked values, shows the orphan bytes do not
13//! self-reclaim, sweeps, and insists the footprint drops sharply while every live value reads back
14//! byte-identical, the tombstones are left untouched, and the store is self-consistent across a
15//! restart. `concurrency_is_safe` runs a writer creating fresh chunked values during the sweep and
16//! insists the epoch guard engaged (`skipped_recent > 0`) and every concurrent value survived
17//! byte-identical -- the proof that a value written during the sweep is never mistaken for an orphan.
18//!
19//! These tests force chunking with an explicit, prod-like threshold (not the drifting dev default),
20//! exactly as `chunk_leak.rs` does.
21//!
22//! [Written with AI entirely](https://need2know.ai/entirely-ai/code)\
23//! Anthropic Claude
24
25use oxedyne_fe2o3_core::{
26 prelude::*,
27 alt::Override,
28 rand::Rand,
29};
30use oxedyne_fe2o3_crypto::enc::EncryptionScheme;
31use oxedyne_fe2o3_hash::{
32 csum::ChecksumScheme,
33 hash::HashScheme,
34};
35use oxedyne_fe2o3_iop_db::api::{
36 Database,
37 RestSchemesOverride,
38};
39use oxedyne_fe2o3_jdat::prelude::*;
40use oxedyne_fe2o3_o3db_sync::{
41 base::constant,
42 comm::response::Wait,
43 data::core::RestSchemesInput,
44 sweep,
45 test::setup,
46};
47
48use std::{
49 fs,
50 path::{
51 Path,
52 PathBuf,
53 },
54 sync::{
55 Arc,
56 atomic::{
57 AtomicBool,
58 AtomicUsize,
59 Ordering,
60 },
61 },
62 thread,
63 time::Duration,
64};
65
66const VALUE_BYTES: usize = 6_000; // comfortably over the threshold, several chunks
67const SMALL_BYTES: usize = 3_200; // still over the threshold, but fewer chunks (a geometry change)
68const TINY_BYTES: usize = 200; // under the threshold: an unchunked value
69
70/// A byte string of the given size, filled deterministically so each version is distinct on disk.
71fn value_of(seed: u8, len: usize) -> Dat {
72 let mut v = vec![0u8; len];
73 for (i, b) in v.iter_mut().enumerate() {
74 *b = seed.wrapping_add((i % 251) as u8);
75 }
76 Dat::BU32(v)
77}
78
79fn chunky_value(seed: u8) -> Dat { value_of(seed, VALUE_BYTES) }
80
81fn scan_wait() -> Wait {
82 Wait {
83 max_wait: Duration::from_secs(60),
84 check_interval: constant::CHECK_INTERVAL,
85 }
86}
87
88pub fn test_sweep(_filter: &'static str) -> Outcome<()> {
89
90 let db_root = res!(canonical_dir("./test_db_sweep"));
91 let db_root_c = res!(canonical_dir("./test_db_sweep_concurrent"));
92
93 let mut enckey = [0u8; 32];
94 Rand::fill_u8(&mut enckey);
95 let aes_gcm = res!(EncryptionScheme::new_aes_256_gcm_with_key(&enckey[..]));
96 let crc32 = ChecksumScheme::new_crc32();
97 let schms2: RestSchemesOverride<EncryptionScheme, HashScheme> =
98 RestSchemesOverride::default().set_encrypter(Override::Default(aes_gcm.clone()));
99 let schms2 = Some(&schms2);
100 let user = setup::Uid::default();
101 let schms_input = RestSchemesInput::new(
102 Some(aes_gcm.clone()),
103 None::<HashScheme>,
104 None::<HashScheme>,
105 Some(crc32.clone()),
106 );
107
108 // Force chunking deterministically, as chunk_leak.rs does: an explicit threshold the value
109 // exceeds, a small chunk size so a value is several chunks, and a data file small enough that
110 // files seal and collection runs.
111 let mut cfg = res!(setup::default_cfg());
112 cfg.data_file_max_bytes = 16_000;
113 cfg.rest_chunk_threshold = 3_000;
114 cfg.rest_chunk_bytes = 1_000;
115 cfg.sync_on_write = true;
116 cfg.zone_overrides = DaticleMap::new();
117
118 res!(sweep_reclaims_orphans(&db_root, &cfg, &schms_input, schms2, &aes_gcm, user));
119 res!(concurrency_is_safe(&db_root_c, &cfg, &schms_input, schms2, &aes_gcm, user));
120
121 Ok(())
122}
123
124/// The sweep must retire every orphaned chunk set, reclaim its bytes, leave every live value
125/// byte-identical and every dead tombstone untouched, and leave the store self-consistent across a
126/// restart. The teeth: garbage collection runs the whole time, yet the orphan bytes do not
127/// self-reclaim -- only the sweep reclaims them.
128fn sweep_reclaims_orphans(
129 db_root: &PathBuf,
130 cfg: &oxedyne_fe2o3_o3db_sync::base::cfg::OzoneConfig,
131 schms_input: &RestSchemesInput<EncryptionScheme, HashScheme, HashScheme, ChecksumScheme>,
132 schms2: Option<&RestSchemesOverride<EncryptionScheme, HashScheme>>,
133 _aes_gcm: &EncryptionScheme,
134 user: setup::Uid,
135)
136 -> Outcome<()>
137{
138 test!(sync_log::stream(), "+--- sweep: reclaims orphans, keeps live values ---");
139
140 let db = res!(setup::start_db(
141 db_root.clone(),
142 Some(cfg.clone()),
143 schms_input.clone(),
144 None,
145 true, // gc on -- the sweep relies on the running collector
146 true, // wipe
147 ));
148
149 let n_live_small = 6usize;
150 let n_geo = 16usize;
151 let n_rnd = 16usize;
152 let n_tomb = 6usize;
153
154 // --- Live unchunked values: must survive byte-identical. ---
155 for k in 0..n_live_small {
156 res!(db.insert(dat!(fmt!("live:small:{:03}", k)), value_of(k as u8, TINY_BYTES), user, schms2));
157 }
158
159 // --- A live chunked value under a stable key: its chunks must never be retired. ---
160 let (_, live_chunks) = res!(db.insert(dat!("live:chunked"), chunky_value(42), user, schms2));
161 if live_chunks < 2 {
162 return Err(err!("The live chunked value was not chunked ({} chunk(s)).", live_chunks;
163 Test, Invalid, Configuration));
164 }
165
166 // --- Dead tombstones: chunked values deleted and never reused. NOT the sweep's job. ---
167 for k in 0..n_tomb {
168 res!(db.insert(dat!(fmt!("tomb:{:03}", k)), chunky_value(k as u8), user, schms2));
169 }
170 thread::sleep(Duration::from_secs(2));
171 for k in 0..n_tomb {
172 if !res!(db.delete(&dat!(fmt!("tomb:{:03}", k)), user, schms2)) {
173 return Err(err!("Delete reported tombstone key {} absent.", k; Test, Missing, Data));
174 }
175 }
176
177 // --- Geometry-change orphans: a big chunked value overwritten by a smaller chunked one. The
178 // prior geometry's chunk set is orphaned; the current small value stays live. ---
179 for k in 0..n_geo {
180 let (_, big) = res!(db.insert(dat!(fmt!("geo:{:03}", k)), value_of(k as u8, VALUE_BYTES), user, schms2));
181 let (_, small) = res!(db.insert(dat!(fmt!("geo:{:03}", k)), value_of(k as u8, SMALL_BYTES), user, schms2));
182 if big < 2 || small < 2 || small >= big {
183 return Err(err!(
184 "Geometry-orphan case geometry wrong: big={} small={} chunks.", big, small;
185 Test, Invalid, Configuration));
186 }
187 }
188
189 // --- Random-ticket orphans: a chunked value written under a random (pre-fix) set_id, then
190 // overwritten by a tiny unchunked value. The whole random-keyed chunk set is orphaned and
191 // the current value is a tiny live record. ---
192 for k in 0..n_rnd {
193 let key = dat!(fmt!("rnd:{:03}", k));
194 let mut set_id_bytes = [0u8; 8];
195 Rand::fill_u8(&mut set_id_bytes);
196 let random_set_id = u64::from_be_bytes(set_id_bytes);
197 let resp = db.api().responder();
198 let nchunks = res!(db.api().store_dat_using_responder_forcing_set_id(
199 key.clone(),
200 value_of(k as u8, VALUE_BYTES),
201 user,
202 schms2,
203 resp.clone(),
204 random_set_id,
205 ));
206 if nchunks < 2 {
207 return Err(err!("The random-ticket value was not chunked ({} chunk(s)).", nchunks;
208 Test, Invalid, Configuration));
209 }
210 // The count, then every record written and durable. Counting answers instead took the
211 // count for one of them, and ignored an error.
212 res!(resp.recv_store_ack());
213 // Overwrite with a tiny unchunked value: supersedes the bunch key, orphans every chunk.
214 res!(db.insert(key.clone(), value_of((k as u8).wrapping_add(1), TINY_BYTES), user, schms2));
215 }
216
217 // Let the collector fully settle: it reclaims every legitimately superseded record (old bunch
218 // keys, same-geometry overwrites), then has nothing left to do, because an orphaned chunk set is
219 // exactly what supersession can never reach. The settled footprint therefore still holds the
220 // orphan bytes -- the teeth below confirm the sweep then finds them. Settling also ages the
221 // orphans well past the epoch skew.
222 let before = res!(settle_footprint(db_root));
223 test!(sync_log::stream(), "sweep: {} data bytes at the settled pre-sweep baseline.", before);
224
225 // Run the sweep against the live store.
226 let report = res!(sweep::sweep_orphans(
227 db.api(),
228 user,
229 schms2,
230 scan_wait(),
231 Duration::from_secs(2),
232 ));
233 test!(sync_log::stream(), "sweep: {}", report.summary().replace('\n', " | "));
234
235 if report.orphans_found == 0 {
236 return Err(err!(
237 "The sweep found no orphans, but random-ticket and geometry-change orphans were seeded: \
238 the orphan detection is not working.";
239 Test, Invalid, Data));
240 }
241 if report.orphans_retired != report.orphans_found {
242 return Err(err!(
243 "The sweep retired {} of {} orphans found: a tombstone write did not land.",
244 report.orphans_retired, report.orphans_found;
245 Test, Mismatch, Data));
246 }
247
248 // Let the collector settle again, then the footprint must have dropped sharply: the orphan
249 // chunk bytes, which GC could not reach before the sweep, have been reclaimed.
250 let after = res!(settle_footprint(db_root));
251 test!(sync_log::stream(),
252 "sweep: data bytes {} before, {} after the sweep (ratio {:.2}).",
253 before, after, after as f64 / before as f64);
254 if after.saturating_mul(3) >= before.saturating_mul(2) {
255 return Err(err!(
256 "Sweep did not reclaim: {} bytes before, {} after, but the orphan chunk bytes should \
257 have fallen away.", before, after;
258 Test, Mismatch, Data));
259 }
260
261 // Every live value reads back byte-identical -- chunked and unchunked.
262 res!(check_live_values(&db, n_live_small, n_geo, n_rnd, schms2));
263
264 // Dead tombstones are untouched: a deleted key still reads absent.
265 for k in 0..n_tomb {
266 if res!(db.get(&dat!(fmt!("tomb:{:03}", k)), schms2)).is_some() {
267 return Err(err!("A dead tombstone key {} reads back a value after the sweep.", k;
268 Test, Invalid, Data));
269 }
270 }
271
272 res!(db.shutdown());
273 thread::sleep(Duration::from_secs(1));
274
275 // RESTART over what the sweep left on disk: every live value must read back unchanged and the
276 // footprint must not have regrown, proving the retirement left the on-disk accounting exact.
277 let db = res!(setup::start_db(
278 db_root.clone(),
279 Some(cfg.clone()),
280 schms_input.clone(),
281 None,
282 true,
283 false, // do not wipe: read what the sweep left
284 ));
285 res!(check_live_values(&db, n_live_small, n_geo, n_rnd, schms2));
286 for k in 0..n_tomb {
287 if res!(db.get(&dat!(fmt!("tomb:{:03}", k)), schms2)).is_some() {
288 return Err(err!("After restart, dead tombstone key {} reads back a value.", k;
289 Test, Invalid, Data));
290 }
291 }
292 let after_restart = res!(zone_data_bytes(db_root));
293 test!(sync_log::stream(), "sweep: {} data bytes after restart.", after_restart);
294 if after_restart.saturating_mul(3) >= before.saturating_mul(2) {
295 return Err(err!(
296 "After a restart the footprint is {} bytes against {} before the sweep: the reclaim \
297 did not survive the restart.", after_restart, before;
298 Test, Mismatch, Data));
299 }
300
301 res!(db.shutdown());
302 thread::sleep(Duration::from_millis(200));
303 test!(sync_log::stream(), "+--- sweep: reclaims orphans, keeps live values : passed ---");
304 Ok(())
305}
306
307/// Reads back every live value and checks it is byte-identical to what was written.
308fn check_live_values(
309 db: &oxedyne_fe2o3_o3db_sync::db::O3db<{ setup::UID_LEN }, setup::Uid, EncryptionScheme, HashScheme, HashScheme, ChecksumScheme>,
310 n_live_small: usize,
311 n_geo: usize,
312 n_rnd: usize,
313 schms2: Option<&RestSchemesOverride<EncryptionScheme, HashScheme>>,
314)
315 -> Outcome<()>
316{
317 for k in 0..n_live_small {
318 let want = value_of(k as u8, TINY_BYTES);
319 match res!(db.get(&dat!(fmt!("live:small:{:03}", k)), schms2)) {
320 Some((got, _)) => if got != want {
321 return Err(err!("Live unchunked value {} came back changed.", k; Test, Invalid, Data));
322 },
323 None => return Err(err!("Live unchunked value {} is gone.", k; Test, Missing, Data)),
324 }
325 }
326 match res!(db.get(&dat!("live:chunked"), schms2)) {
327 Some((got, _)) => if got != chunky_value(42) {
328 return Err(err!("The live chunked value came back changed."; Test, Invalid, Data));
329 },
330 None => return Err(err!("The live chunked value is gone."; Test, Missing, Data)),
331 }
332 for k in 0..n_geo {
333 let want = value_of(k as u8, SMALL_BYTES);
334 match res!(db.get(&dat!(fmt!("geo:{:03}", k)), schms2)) {
335 Some((got, _)) => if got != want {
336 return Err(err!("Geometry-orphan current value {} came back changed.", k; Test, Invalid, Data));
337 },
338 None => return Err(err!("Geometry-orphan current value {} is gone.", k; Test, Missing, Data)),
339 }
340 }
341 for k in 0..n_rnd {
342 let want = value_of((k as u8).wrapping_add(1), TINY_BYTES);
343 match res!(db.get(&dat!(fmt!("rnd:{:03}", k)), schms2)) {
344 Some((got, _)) => if got != want {
345 return Err(err!("Random-ticket current value {} came back changed.", k; Test, Invalid, Data));
346 },
347 None => return Err(err!("Random-ticket current value {} is gone.", k; Test, Missing, Data)),
348 }
349 }
350 Ok(())
351}
352
353/// A writer creating fresh chunked values during the sweep proves the online claim: the epoch guard
354/// must skip values written after the sweep started (`skipped_recent > 0`), and every concurrently
355/// written value must survive byte-identical. A version of the sweep that retired a fresh value's
356/// chunks would be the catastrophic bug this rules out.
357fn concurrency_is_safe(
358 db_root: &PathBuf,
359 cfg: &oxedyne_fe2o3_o3db_sync::base::cfg::OzoneConfig,
360 schms_input: &RestSchemesInput<EncryptionScheme, HashScheme, HashScheme, ChecksumScheme>,
361 schms2: Option<&RestSchemesOverride<EncryptionScheme, HashScheme>>,
362 aes_gcm: &EncryptionScheme,
363 user: setup::Uid,
364)
365 -> Outcome<()>
366{
367 test!(sync_log::stream(), "+--- sweep: concurrency is safe (online) ---");
368
369 let db = Arc::new(res!(setup::start_db(
370 db_root.clone(),
371 Some(cfg.clone()),
372 schms_input.clone(),
373 None,
374 true,
375 true,
376 )));
377
378 const EPOCH_SKEW_SECS: u64 = 3;
379
380 // Seed geometry-change orphans and age them well past the skew: these are the genuine, settled
381 // orphans the sweep must retire.
382 let n_geo = 12usize;
383 for k in 0..n_geo {
384 res!(db.insert(dat!(fmt!("cgeo:{:03}", k)), value_of(k as u8, VALUE_BYTES), user, schms2));
385 res!(db.insert(dat!(fmt!("cgeo:{:03}", k)), value_of(k as u8, SMALL_BYTES), user, schms2));
386 }
387 thread::sleep(Duration::from_secs(EPOCH_SKEW_SECS + 3)); // age past the skew
388
389 // A writer thread hammering fresh chunked values under new keys, with its own schemes so it
390 // borrows nothing from this frame. These are values written concurrently with the sweep; none
391 // of them may be retired.
392 let stop = Arc::new(AtomicBool::new(false));
393 let written = Arc::new(AtomicUsize::new(0));
394 let db_w = Arc::clone(&db);
395 let stop_w = Arc::clone(&stop);
396 let written_w = Arc::clone(&written);
397 let aes_w = aes_gcm.clone();
398 let writer = thread::spawn(move || -> Outcome<()> {
399 let schms_owned: RestSchemesOverride<EncryptionScheme, HashScheme> =
400 RestSchemesOverride::default().set_encrypter(Override::Default(aes_w));
401 let schms_w = Some(&schms_owned);
402 let mut i = 0u32;
403 while !stop_w.load(Ordering::Relaxed) {
404 let key = dat!(fmt!("fresh:{:06}", i));
405 res!(db_w.insert(key, chunky_value((i % 200) as u8), user, schms_w));
406 written_w.fetch_add(1, Ordering::Relaxed);
407 i += 1;
408 thread::sleep(Duration::from_millis(25));
409 }
410 Ok(())
411 });
412
413 // Let the writer get going, then -- immediately before the sweep -- overwrite some chunked
414 // values at a new geometry. Their OLD chunk sets become genuine orphans, but they are younger
415 // than the skew, so the epoch guard must DEFER them (count them skipped_recent, not retire them):
416 // this is exactly the hazard of a value overwritten just as the sweep starts, and it makes the
417 // guard fire deterministically rather than relying on a scan-timing window.
418 let n_recent = 6usize;
419 for k in 0..n_recent {
420 res!(db.insert(dat!(fmt!("recent:{:03}", k)), value_of(k as u8, VALUE_BYTES), user, schms2));
421 }
422 thread::sleep(Duration::from_millis(200));
423 for k in 0..n_recent {
424 res!(db.insert(dat!(fmt!("recent:{:03}", k)), value_of(k as u8, SMALL_BYTES), user, schms2));
425 }
426
427 let report = res!(sweep::sweep_orphans(
428 db.api(),
429 user,
430 schms2,
431 scan_wait(),
432 Duration::from_secs(EPOCH_SKEW_SECS),
433 ));
434
435 // Keep writing a touch longer, then stop and join.
436 thread::sleep(Duration::from_millis(300));
437 stop.store(true, Ordering::Relaxed);
438 match writer.join() {
439 Ok(r) => res!(r),
440 Err(_) => return Err(err!("The concurrent writer thread panicked."; Test, Bug)),
441 }
442
443 test!(sync_log::stream(), "sweep/concurrent: {}", report.summary().replace('\n', " | "));
444
445 // The online-safety proof: the epoch guard deferred the recently overwritten values' old chunks
446 // rather than retiring them. Without the guard these recent orphans would have been retired.
447 if report.skipped_recent == 0 {
448 return Err(err!(
449 "The epoch guard never engaged (skipped_recent == 0): a value overwritten just before \
450 the sweep had its old chunks treated as retirable. The online safety guard is not being \
451 applied.";
452 Test, Invalid, Data));
453 }
454
455 // The aged seeded orphans should have been retired.
456 if report.orphans_found == 0 {
457 return Err(err!("The concurrency sweep found no orphans, but aged orphans were seeded.";
458 Test, Invalid, Data));
459 }
460
461 // The recently overwritten values read back as their current (small) value, byte-identical: the
462 // guard deferred their old chunks without touching the live value.
463 for k in 0..n_recent {
464 let want = value_of(k as u8, SMALL_BYTES);
465 match res!(db.get(&dat!(fmt!("recent:{:03}", k)), schms2)) {
466 Some((got, _)) => if got != want {
467 return Err(err!("A value overwritten just before the sweep (recent:{:03}) came back \
468 changed.", k; Test, Invalid, Data));
469 },
470 None => return Err(err!("A value overwritten just before the sweep (recent:{:03}) is gone.",
471 k; Test, Missing, Data)),
472 }
473 }
474
475 // Every concurrently written value survives byte-identical -- none of its chunks were retired.
476 let total = written.load(Ordering::Relaxed);
477 if total == 0 {
478 return Err(err!("The concurrent writer wrote nothing."; Test, Invalid, Data));
479 }
480 for i in 0..(total as u32) {
481 let want = chunky_value((i % 200) as u8);
482 match res!(db.get(&dat!(fmt!("fresh:{:06}", i)), schms2)) {
483 Some((got, _)) => if got != want {
484 return Err(err!(
485 "A value written concurrently with the sweep (fresh:{:06}) came back changed: \
486 the sweep retired a live value's chunks.", i;
487 Test, Invalid, Data));
488 },
489 None => return Err(err!(
490 "A value written concurrently with the sweep (fresh:{:06}) is gone: the sweep \
491 retired a live value's chunks.", i;
492 Test, Missing, Data)),
493 }
494 }
495 test!(sync_log::stream(),
496 "sweep/concurrent: {} fresh values all intact, {} skipped by the epoch guard.",
497 total, report.skipped_recent);
498
499 // Recover the store for a clean shutdown (the writer's Arc clone is dropped after the join).
500 match Arc::try_unwrap(db) {
501 Ok(db) => res!(db.shutdown()),
502 Err(_) => return Err(err!(
503 "Could not reclaim sole ownership of the store to shut it down.";
504 Test, Bug)),
505 }
506 thread::sleep(Duration::from_millis(200));
507 test!(sync_log::stream(), "+--- sweep: concurrency is safe (online) : passed ---");
508 Ok(())
509}
510
511/// The caller's own settle: waits for the asynchronous collector to engage and then finish, and
512/// returns the quiesced data-file footprint. `sweep_orphans` is sleep-free -- it returns the moment
513/// its tombstones are acknowledged, before the collector has reclaimed anything -- so a footprint
514/// read immediately after it would be the un-reclaimed figure. An initial wait lets collection
515/// start (otherwise the footprint would read high and stable and the loop would conclude "settled"
516/// before a single byte was reclaimed), then the loop polls until two reads two seconds apart differ
517/// by under 3%, i.e. collection has stopped shrinking the store.
518fn settle_footprint(db_root: &Path) -> Outcome<u64> {
519 thread::sleep(Duration::from_secs(4));
520 let mut prev = res!(zone_data_bytes(db_root));
521 for _ in 0..20 {
522 thread::sleep(Duration::from_secs(2));
523 let now = res!(zone_data_bytes(db_root));
524 // The footprint only falls as GC runs; stable means it fell by less than 3% this step.
525 if now >= prev.saturating_mul(97) / 100 {
526 return Ok(now);
527 }
528 prev = now;
529 }
530 Ok(prev)
531}
532
533fn zone_data_bytes(db_root: &Path) -> Outcome<u64> {
534 let mut total = 0u64;
535 res!(walk_data_files(db_root, &mut total));
536 Ok(total)
537}
538
539fn walk_data_files(dir: &Path, total: &mut u64) -> Outcome<()> {
540 for entry in res!(fs::read_dir(dir)) {
541 let entry = res!(entry);
542 let path = entry.path();
543 if path.is_dir() {
544 res!(walk_data_files(&path, total));
545 } else if path.extension().map(|e| e == constant::DATA_FILE_EXT).unwrap_or(false) {
546 let meta = res!(entry.metadata());
547 *total += meta.len();
548 }
549 }
550 Ok(())
551}
552
553/// Creates the directory if it does not exist.
554fn canonical_dir(p: &str) -> Outcome<PathBuf> {
555 match Path::new(p).canonicalize() {
556 Ok(path) => Ok(path),
557 Err(_) => {
558 res!(fs::create_dir_all(p));
559 match Path::new(p).canonicalize() {
560 Ok(path) => Ok(path),
561 Err(e) => Err(err!(e, "Cannot canonicalise {:?}.", p; IO, Path)),
562 }
563 },
564 }
565}
566
567#[test]
568fn main() -> Outcome<()> {
569 log_set_level!("trace");
570 let outcome = test_sweep("all");
571 log_finish_wait!();
572 outcome
573}