OXYGENHUB
# a5f76c79666c oxygen / crates / oxy-core / src / strata.rs
25.0 KiB · Rust · last changed Strata: the Store trait, a directory store, a memory store, and the conformance check 1 month ago source blame history raw
a5f76c79 Andy McCall ↶ 1
//! Oxygen Strata (docs/stores.md): the storage a repository can be kept
2
//! on, as a trait, so that a bucket, a directory or a hub's disk all look
3
//! the same to the code that pushes to and fetches from them.
4
//!
5
//! A store has two halves. **Objects** are written once and never
6
//! change: packs and their indexes, named by content, which any store
7
//! can hold safely. **References** are the one mutable record (bookmarks,
8
//! tags, the policy), updated only by compare-and-swap on a version the
9
//! store hands out: an ETag on S3, a generation on GCS, a content hash on
10
//! a directory. Two writers can therefore share a store with no lock
11
//! server: the loser of a race reads again, merges, and tries again.
12
//!
13
//! The working repository (`.oxy/`) is not a `Store`; it is read and
14
//! written directly, as before. A store is what a *remote* is kept on.
15
//! This module holds the trait, a directory store, an in-memory store
16
//! for tests, and the conformance check every store must pass. Cloud
17
//! stores live in a crate of their own, because of what they pull in.
18
19
use std::collections::BTreeMap;
20
use std::fs;
21
use std::io::{self, Read, Seek, SeekFrom, Write};
22
use std::ops::Range;
23
use std::path::{Path, PathBuf};
24
use std::sync::Mutex;
25
use std::time::{Duration, Instant};
26
27
/// The version of a reference record as the store reports it: opaque,
28
/// compared only for equality, and only ever valid for the store that
29
/// issued it. Writing the same bytes again may keep the version (S3's
30
/// ETag and a directory's content hash are both of the content): a
31
/// compare-and-swap guards against losing a *different* record, which is
32
/// all it needs to.
33
#[derive(Clone, Debug, PartialEq, Eq)]
34
pub struct Version(pub String);
35
36
/// One key in a listing.
37
#[derive(Clone, Debug, PartialEq, Eq)]
38
pub struct Entry {
39
    pub key: String,
40
    pub size: u64,
41
}
42
43
#[derive(Debug, thiserror::Error)]
44
pub enum Error {
45
    #[error("{0}: not in the store")]
46
    NotFound(String),
47
    #[error("{0}: already in the store")]
48
    Exists(String),
49
    #[error("{0}: changed since it was read")]
50
    Conflict(String),
51
    #[error("bad store key {0:?}: {1}")]
52
    Key(String, &'static str),
53
    #[error("{0}: busy (another writer holds it)")]
54
    Busy(String),
55
    #[error(transparent)]
56
    Io(#[from] io::Error),
57
    #[error("{0}")]
58
    Backend(String),
59
}
60
61
pub type Result<T> = std::result::Result<T, Error>;
62
63
/// Longest key a store accepts; S3 allows 1024 bytes.
64
pub const MAX_KEY_LEN: usize = 1024;
65
66
/// Refuse a key that is not a plain relative path: `/`-separated
67
/// segments of `[A-Za-z0-9._-]`, none empty, none `.` or `..`. The same
68
/// rule on every store, so a key written to one can be read from another,
69
/// and so a key can never escape a directory store's root.
70
pub fn check_key(key: &str) -> Result<()> {
71
    let bad = |why| Error::Key(key.to_string(), why);
72
    if key.is_empty() {
73
        return Err(bad("empty"));
74
    }
75
    if key.len() > MAX_KEY_LEN {
76
        return Err(bad("longer than 1024 bytes"));
77
    }
78
    for seg in key.split('/') {
79
        if seg.is_empty() {
80
            return Err(bad(
81
                "an empty segment (a leading, trailing or doubled slash)",
82
            ));
83
        }
84
        if seg == "." || seg == ".." {
85
            return Err(bad("a . or .. segment"));
86
        }
87
        if !seg
88
            .bytes()
89
            .all(|b| b.is_ascii_alphanumeric() || b == b'.' || b == b'_' || b == b'-')
90
        {
91
            return Err(bad("a character outside A-Z a-z 0-9 . _ -"));
92
        }
93
    }
94
    Ok(())
95
}
96
97
/// Where a repository's packs and references are kept. Every method is
98
/// synchronous and may be called from several threads at once.
99
pub trait Store: Send + Sync {
100
    /// A name for messages: the URL, or the directory.
101
    fn describe(&self) -> String;
102
103
    // ---- objects: written once, never changed
104
105
    fn has(&self, key: &str) -> Result<bool>;
106
    /// The whole object; `NotFound` when there is none.
107
    fn get(&self, key: &str) -> Result<Vec<u8>>;
108
    /// Bytes `range` of an object; an error when the range runs past its end.
109
    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>>;
110
    /// Write an object that does not exist yet; `Exists` when it does, and
111
    /// then nothing is changed. Readers never see a partial object.
112
    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()>;
113
    /// Every key under `prefix` (a key prefix, not necessarily a whole
114
    /// segment), sorted, with sizes.
115
    fn list(&self, prefix: &str) -> Result<Vec<Entry>>;
116
    /// Remove an object; fine when it is already gone.
117
    fn delete(&self, key: &str) -> Result<()>;
118
119
    // ---- references: one small record, updated by compare-and-swap
120
121
    /// The record and its current version, or `None` when it was never
122
    /// written.
123
    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>>;
124
    /// Replace the record only if its version is still `expected`
125
    /// (`None`: only if it does not exist yet). `Conflict` otherwise, and
126
    /// then nothing is changed. Returns the new version.
127
    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version>;
128
}
129
130
/// A reference record's version on stores that have no native one: the
131
/// hash of its bytes, which two different records never share.
132
fn content_version(bytes: &[u8]) -> Version {
133
    Version(blake3::hash(bytes).to_hex().to_string())
134
}
135
136
fn check_expected(key: &str, current: Option<&Version>, expected: Option<&Version>) -> Result<()> {
137
    if current == expected {
138
        Ok(())
139
    } else {
140
        Err(Error::Conflict(key.to_string()))
141
    }
142
}
143
144
// ---------------------------------------------------------------- directory
145
146
/// A store in a directory: each key a file under the root. Create-only
147
/// writes go through a hard link, which the file system refuses when the
148
/// name is taken, and a reference is updated under a lock file beside it.
149
/// Several processes on one machine can share it; it is also what the
150
/// tests and `oxy remote check` use.
151
pub struct LocalStore {
152
    root: PathBuf,
153
}
154
155
/// How long a reference's lock may be held before it counts as abandoned
156
/// by a process that died, and how long a writer waits for it.
157
const LOCK_STALE: Duration = Duration::from_secs(60);
158
const LOCK_WAIT: Duration = Duration::from_secs(10);
159
160
impl LocalStore {
161
    /// Open the store at `root`, creating the directory if needed.
162
    pub fn open(root: impl Into<PathBuf>) -> Result<LocalStore> {
163
        let root = root.into();
164
        fs::create_dir_all(&root)?;
165
        Ok(LocalStore { root })
166
    }
167
168
    pub fn root(&self) -> &Path {
169
        &self.root
170
    }
171
172
    fn path(&self, key: &str) -> Result<PathBuf> {
173
        check_key(key)?;
174
        Ok(self.root.join(key))
175
    }
176
177
    fn temp_beside(&self, path: &Path) -> PathBuf {
178
        let mut nonce = [0u8; 6];
179
        rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut nonce);
180
        let name = path
181
            .file_name()
182
            .map(|n| n.to_string_lossy().into_owned())
183
            .unwrap_or_default();
184
        path.with_file_name(format!(".{name}.tmp-{}", hex::encode(nonce)))
185
    }
186
187
    fn list_dir(&self, dir: &Path, rel: &str, prefix: &str, out: &mut Vec<Entry>) -> Result<()> {
188
        let entries = match fs::read_dir(dir) {
189
            Ok(e) => e,
190
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()),
191
            Err(e) => return Err(e.into()),
192
        };
193
        for entry in entries {
194
            let entry = entry?;
195
            let name = entry.file_name().to_string_lossy().into_owned();
196
            // Temporary and lock files are not objects.
197
            if name.starts_with('.') {
198
                continue;
199
            }
200
            let key = if rel.is_empty() {
201
                name.clone()
202
            } else {
203
                format!("{rel}/{name}")
204
            };
205
            let ty = entry.file_type()?;
206
            if ty.is_dir() {
207
                // Descend only where the prefix could still match.
208
                let dir_prefix = format!("{key}/");
209
                if prefix.starts_with(&dir_prefix) || dir_prefix.starts_with(prefix) {
210
                    self.list_dir(&entry.path(), &key, prefix, out)?;
211
                }
212
            } else if ty.is_file() && key.starts_with(prefix) {
213
                out.push(Entry {
214
                    key,
215
                    size: entry.metadata()?.len(),
216
                });
217
            }
218
        }
219
        Ok(())
220
    }
221
222
    /// Take `path.lock`, waiting for another writer, and removing a lock
223
    /// a dead process left behind.
224
    fn lock(&self, key: &str, path: &Path) -> Result<PathBuf> {
225
        let lock = path.with_file_name(format!(
226
            ".{}.lock",
227
            path.file_name()
228
                .map(|n| n.to_string_lossy().into_owned())
229
                .unwrap_or_default()
230
        ));
231
        let started = Instant::now();
232
        loop {
233
            match fs::OpenOptions::new()
234
                .write(true)
235
                .create_new(true)
236
                .open(&lock)
237
            {
238
                Ok(mut f) => {
239
                    let _ = writeln!(f, "{}", std::process::id());
240
                    return Ok(lock);
241
                }
242
                Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {
243
                    if let Ok(m) = fs::metadata(&lock) {
244
                        let age = m.modified().ok().and_then(|t| t.elapsed().ok());
245
                        if age.is_some_and(|a| a > LOCK_STALE) {
246
                            let _ = fs::remove_file(&lock);
247
                            continue;
248
                        }
249
                    }
250
                    if started.elapsed() > LOCK_WAIT {
251
                        return Err(Error::Busy(key.to_string()));
252
                    }
253
                    std::thread::sleep(Duration::from_millis(10));
254
                }
255
                Err(e) => return Err(e.into()),
256
            }
257
        }
258
    }
259
}
260
261
impl Store for LocalStore {
262
    fn describe(&self) -> String {
263
        self.root.display().to_string()
264
    }
265
266
    fn has(&self, key: &str) -> Result<bool> {
267
        Ok(self.path(key)?.is_file())
268
    }
269
270
    fn get(&self, key: &str) -> Result<Vec<u8>> {
271
        let path = self.path(key)?;
272
        match fs::read(&path) {
273
            Ok(b) => Ok(b),
274
            Err(e) if e.kind() == io::ErrorKind::NotFound => Err(Error::NotFound(key.to_string())),
275
            Err(e) => Err(e.into()),
276
        }
277
    }
278
279
    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {
280
        let path = self.path(key)?;
281
        let mut f = match fs::File::open(&path) {
282
            Ok(f) => f,
283
            Err(e) if e.kind() == io::ErrorKind::NotFound => {
284
                return Err(Error::NotFound(key.to_string()))
285
            }
286
            Err(e) => return Err(e.into()),
287
        };
288
        let len = f.metadata()?.len();
289
        if range.end > len || range.start > range.end {
290
            return Err(Error::Backend(format!(
291
                "{key}: bytes {}..{} asked of an object of {len} bytes",
292
                range.start, range.end
293
            )));
294
        }
295
        f.seek(SeekFrom::Start(range.start))?;
296
        let mut out = vec![0u8; (range.end - range.start) as usize];
297
        f.read_exact(&mut out)?;
298
        Ok(out)
299
    }
300
301
    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> {
302
        let path = self.path(key)?;
303
        if path.exists() {
304
            return Err(Error::Exists(key.to_string()));
305
        }
306
        if let Some(dir) = path.parent() {
307
            fs::create_dir_all(dir)?;
308
        }
309
        // Written beside its name, then linked into it: the link fails if
310
        // the name was taken meanwhile, and a reader never sees a part.
311
        let tmp = self.temp_beside(&path);
312
        let written = (|| {
313
            let mut f = fs::File::create(&tmp)?;
314
            f.write_all(bytes)?;
315
            f.sync_all()
316
        })();
317
        let linked = written.and_then(|()| fs::hard_link(&tmp, &path));
318
        let _ = fs::remove_file(&tmp);
319
        match linked {
320
            Ok(()) => {
321
                if let Some(dir) = path.parent() {
322
                    crate::durable::sync_dir(dir)?;
323
                }
324
                Ok(())
325
            }
326
            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {
327
                Err(Error::Exists(key.to_string()))
328
            }
329
            Err(e) => Err(e.into()),
330
        }
331
    }
332
333
    fn list(&self, prefix: &str) -> Result<Vec<Entry>> {
334
        let mut out = Vec::new();
335
        self.list_dir(&self.root, "", prefix, &mut out)?;
336
        out.sort_by(|a, b| a.key.cmp(&b.key));
337
        Ok(out)
338
    }
339
340
    fn delete(&self, key: &str) -> Result<()> {
341
        let path = self.path(key)?;
342
        match fs::remove_file(&path) {
343
            Ok(()) => Ok(()),
344
            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
345
            Err(e) => Err(e.into()),
346
        }
347
    }
348
349
    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> {
350
        let path = self.path(key)?;
351
        match fs::read(&path) {
352
            Ok(b) => {
353
                let v = content_version(&b);
354
                Ok(Some((b, v)))
355
            }
356
            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),
357
            Err(e) => Err(e.into()),
358
        }
359
    }
360
361
    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> {
362
        let path = self.path(key)?;
363
        if let Some(dir) = path.parent() {
364
            fs::create_dir_all(dir)?;
365
        }
366
        let lock = self.lock(key, &path)?;
367
        let result = (|| {
368
            let current = self.read_ref(key)?.map(|(_, v)| v);
369
            check_expected(key, current.as_ref(), expected)?;
370
            crate::durable::write_atomic(&path, bytes)?;
371
            Ok(content_version(bytes))
372
        })();
373
        let _ = fs::remove_file(&lock);
374
        result
375
    }
376
}
377
378
// ---------------------------------------------------------------- memory
379
380
/// A store in memory, for tests: the conditional writes behave as the
381
/// cloud stores' do.
382
#[derive(Default)]
383
pub struct MemoryStore {
384
    objects: Mutex<BTreeMap<String, Vec<u8>>>,
385
}
386
387
impl MemoryStore {
388
    pub fn new() -> MemoryStore {
389
        MemoryStore::default()
390
    }
391
392
    fn map(&self) -> std::sync::MutexGuard<'_, BTreeMap<String, Vec<u8>>> {
393
        self.objects.lock().unwrap_or_else(|p| p.into_inner())
394
    }
395
}
396
397
impl Store for MemoryStore {
398
    fn describe(&self) -> String {
399
        "memory".into()
400
    }
401
402
    fn has(&self, key: &str) -> Result<bool> {
403
        check_key(key)?;
404
        Ok(self.map().contains_key(key))
405
    }
406
407
    fn get(&self, key: &str) -> Result<Vec<u8>> {
408
        check_key(key)?;
409
        self.map()
410
            .get(key)
411
            .cloned()
412
            .ok_or_else(|| Error::NotFound(key.to_string()))
413
    }
414
415
    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {
416
        let all = self.get(key)?;
417
        let len = all.len() as u64;
418
        if range.end > len || range.start > range.end {
419
            return Err(Error::Backend(format!(
420
                "{key}: bytes {}..{} asked of an object of {len} bytes",
421
                range.start, range.end
422
            )));
423
        }
424
        Ok(all[range.start as usize..range.end as usize].to_vec())
425
    }
426
427
    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> {
428
        check_key(key)?;
429
        let mut map = self.map();
430
        if map.contains_key(key) {
431
            return Err(Error::Exists(key.to_string()));
432
        }
433
        map.insert(key.to_string(), bytes.to_vec());
434
        Ok(())
435
    }
436
437
    fn list(&self, prefix: &str) -> Result<Vec<Entry>> {
438
        Ok(self
439
            .map()
440
            .range(prefix.to_string()..)
441
            .take_while(|(k, _)| k.starts_with(prefix))
442
            .map(|(k, v)| Entry {
443
                key: k.clone(),
444
                size: v.len() as u64,
445
            })
446
            .collect())
447
    }
448
449
    fn delete(&self, key: &str) -> Result<()> {
450
        check_key(key)?;
451
        self.map().remove(key);
452
        Ok(())
453
    }
454
455
    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> {
456
        check_key(key)?;
457
        Ok(self.map().get(key).map(|b| (b.clone(), content_version(b))))
458
    }
459
460
    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> {
461
        check_key(key)?;
462
        let mut map = self.map();
463
        let current = map.get(key).map(|b| content_version(b));
464
        check_expected(key, current.as_ref(), expected)?;
465
        map.insert(key.to_string(), bytes.to_vec());
466
        Ok(content_version(bytes))
467
    }
468
}
469
470
// ---------------------------------------------------------------- conformance
471
472
/// Exercise a store under `prefix` (which must be free, and is left
473
/// empty) and report what passed: object round trips, ranges, create-only
474
/// refusal, listing, deletion, and compare-and-swap on a reference. Every
475
/// store runs this in its tests, and `oxy remote check` runs it against a
476
/// remote before trusting it with a repository.
477
pub fn conformance(store: &dyn Store, prefix: &str) -> Result<Vec<&'static str>> {
478
    let mut passed = Vec::new();
479
    let key = |name: &str| format!("{prefix}/{name}");
480
    let fail = |what: &str| Error::Backend(format!("{}: {what}", store.describe()));
481
    let expect = |ok: bool, what: &str| if ok { Ok(()) } else { Err(fail(what)) };
482
483
    expect(
484
        store.list(prefix)?.is_empty(),
485
        "the probe prefix is not empty",
486
    )?;
487
    expect(
488
        !store.has(&key("a"))?,
489
        "has() of a missing object says true",
490
    )?;
491
    expect(
492
        matches!(store.get(&key("a")), Err(Error::NotFound(_))),
493
        "get() of a missing object is not NotFound",
494
    )?;
495
    passed.push("missing objects are reported missing");
496
497
    let a: Vec<u8> = (0..256u32 * 1024).map(|i| (i % 251) as u8).collect();
498
    store.put_new(&key("a"), &a)?;
499
    expect(store.has(&key("a"))?, "has() after put_new says false")?;
500
    expect(
501
        store.get(&key("a"))? == a,
502
        "get() returns other bytes than put_new wrote",
503
    )?;
504
    passed.push("an object round-trips");
505
506
    expect(
507
        store.get_range(&key("a"), 1000..1500)? == a[1000..1500],
508
        "get_range() returns the wrong bytes",
509
    )?;
510
    expect(
511
        store.get_range(&key("a"), 0..0)?.is_empty(),
512
        "an empty range is not empty",
513
    )?;
514
    expect(
515
        store.get_range(&key("a"), 0..a.len() as u64 + 1).is_err(),
516
        "a range past the end is not refused",
517
    )?;
518
    passed.push("ranges read correctly and are bounded");
519
520
    expect(
521
        matches!(store.put_new(&key("a"), b"other"), Err(Error::Exists(_))),
522
        "put_new() over an existing object is not refused",
523
    )?;
524
    expect(
525
        store.get(&key("a"))? == a,
526
        "a refused put_new() changed the object",
527
    )?;
528
    passed.push("objects are written once");
529
530
    store.put_new(&key("dir/b"), b"bb")?;
531
    let listed = store.list(prefix)?;
532
    expect(
533
        listed
534
            == vec![
535
                Entry {
536
                    key: key("a"),
537
                    size: a.len() as u64,
538
                },
539
                Entry {
540
                    key: key("dir/b"),
541
                    size: 2,
542
                },
543
            ],
544
        "list() does not return the two objects, sorted, with their sizes",
545
    )?;
546
    expect(
547
        store.list(&key("dir/"))?
548
            == vec![Entry {
549
                key: key("dir/b"),
550
                size: 2,
551
            }],
552
        "list() of a sub-prefix is wrong",
553
    )?;
554
    expect(
555
        store.list(&key("nothing/"))?.is_empty(),
556
        "list() of an unused prefix is not empty",
557
    )?;
558
    passed.push("listing by prefix");
559
560
    store.delete(&key("dir/b"))?;
561
    expect(
562
        !store.has(&key("dir/b"))?,
563
        "an object is still there after delete()",
564
    )?;
565
    store.delete(&key("dir/b"))?;
566
    passed.push("deletion, and deleting again, succeed");
567
568
    let r = key("refs");
569
    expect(
570
        store.read_ref(&r)?.is_none(),
571
        "an unwritten reference reads as something",
572
    )?;
573
    let v1 = store.write_ref(&r, b"one", None)?;
574
    expect(
575
        matches!(store.write_ref(&r, b"again", None), Err(Error::Conflict(_))),
576
        "a second create-only write of the reference is not refused",
577
    )?;
578
    let (bytes, seen) = store
579
        .read_ref(&r)?
580
        .ok_or_else(|| fail("the reference vanished"))?;
581
    expect(
582
        bytes == b"one" && seen == v1,
583
        "the reference reads back wrong",
584
    )?;
585
    let v2 = store.write_ref(&r, b"two", Some(&v1))?;
586
    expect(v2 != v1, "the version did not change on write")?;
587
    expect(
588
        matches!(
589
            store.write_ref(&r, b"three", Some(&v1)),
590
            Err(Error::Conflict(_))
591
        ),
592
        "a write with a stale version is not refused",
593
    )?;
594
    let (bytes, seen) = store
595
        .read_ref(&r)?
596
        .ok_or_else(|| fail("the reference vanished"))?;
597
    expect(
598
        bytes == b"two" && seen == v2,
599
        "a refused write changed the reference",
600
    )?;
601
    passed.push("references update by compare-and-swap");
602
603
    store.delete(&key("a"))?;
604
    store.delete(&r)?;
605
    expect(
606
        store.list(prefix)?.is_empty(),
607
        "the probe prefix is not empty after cleanup",
608
    )?;
609
    passed.push("cleanup");
610
    Ok(passed)
611
}
612
613
#[cfg(test)]
614
mod tests {
615
    use super::*;
616
    use std::sync::Arc;
617
618
    #[test]
619
    fn keys_are_plain_relative_paths() {
620
        for ok in ["a", "packs/ab.pack", "a-b_c.d/e"] {
621
            check_key(ok).unwrap();
622
        }
623
        for bad in [
624
            "", "/a", "a/", "a//b", "..", "a/../b", "./a", "a b", "a\\b", "ä",
625
        ] {
626
            assert!(check_key(bad).is_err(), "{bad:?}");
627
        }
628
        assert!(check_key(&"x".repeat(MAX_KEY_LEN + 1)).is_err());
629
    }
630
631
    #[test]
632
    fn the_memory_store_conforms() {
633
        let store = MemoryStore::new();
634
        let passed = conformance(&store, "probe").unwrap();
635
        assert_eq!(passed.len(), 8, "{passed:?}");
636
    }
637
638
    #[test]
639
    fn the_directory_store_conforms() {
640
        let dir = tempfile::tempdir().unwrap();
641
        let store = LocalStore::open(dir.path().join("bucket")).unwrap();
642
        let passed = conformance(&store, "repos/one").unwrap();
643
        assert_eq!(passed.len(), 8, "{passed:?}");
644
        // Nothing but the (empty) directories and no stray temporaries.
645
        let stray: Vec<_> = walkdir(dir.path())
646
            .into_iter()
647
            .filter(|p| p.is_file())
648
            .collect();
649
        assert!(stray.is_empty(), "{stray:?}");
650
    }
651
652
    fn walkdir(dir: &Path) -> Vec<PathBuf> {
653
        let mut out = vec![];
654
        if let Ok(rd) = fs::read_dir(dir) {
655
            for e in rd.flatten() {
656
                let p = e.path();
657
                if p.is_dir() {
658
                    out.extend(walkdir(&p));
659
                } else {
660
                    out.push(p);
661
                }
662
            }
663
        }
664
        out
665
    }
666
667
    #[test]
668
    fn a_directory_store_refuses_keys_that_leave_its_root() {
669
        let dir = tempfile::tempdir().unwrap();
670
        let store = LocalStore::open(dir.path().join("bucket")).unwrap();
671
        assert!(matches!(
672
            store.put_new("../escape", b"x"),
673
            Err(Error::Key(..))
674
        ));
675
        assert!(matches!(store.get("/etc/passwd"), Err(Error::Key(..))));
676
        assert!(!dir.path().join("escape").exists());
677
    }
678
679
    /// Many threads race to move a reference from the same version: exactly
680
    /// one wins each round, on both stores.
681
    #[test]
682
    fn compare_and_swap_admits_one_writer_per_round() {
683
        let dir = tempfile::tempdir().unwrap();
684
        let local: Arc<dyn Store> = Arc::new(LocalStore::open(dir.path()).unwrap());
685
        let memory: Arc<dyn Store> = Arc::new(MemoryStore::new());
686
        for store in [local, memory] {
687
            let v0 = store.write_ref("refs", b"0", None).unwrap();
688
            let wins: Vec<bool> = (0..8)
689
                .map(|i| {
690
                    let store = store.clone();
691
                    let v0 = v0.clone();
692
                    std::thread::spawn(move || {
693
                        store
694
                            .write_ref("refs", format!("{}", i + 1).as_bytes(), Some(&v0))
695
                            .is_ok()
696
                    })
697
                })
698
                .collect::<Vec<_>>()
699
                .into_iter()
700
                .map(|h| h.join().unwrap())
701
                .collect();
702
            assert_eq!(
703
                wins.iter().filter(|w| **w).count(),
704
                1,
705
                "{}: {wins:?}",
706
                store.describe()
707
            );
708
            let (bytes, _) = store.read_ref("refs").unwrap().unwrap();
709
            assert_ne!(bytes, b"0");
710
        }
711
    }
712
713
    #[test]
714
    fn a_stale_lock_is_broken_and_a_live_one_waited_for() {
715
        let dir = tempfile::tempdir().unwrap();
716
        let store = LocalStore::open(dir.path()).unwrap();
717
        let lock = dir.path().join(".refs.lock");
718
        fs::write(&lock, "dead").unwrap();
719
        let old = std::time::SystemTime::now() - LOCK_STALE - Duration::from_secs(5);
720
        fs::File::open(&lock).unwrap().set_modified(old).unwrap();
721
        store.write_ref("refs", b"after a crash", None).unwrap();
722
        assert!(!lock.exists());
723
    }
724
725
    #[test]
726
    fn temporaries_and_locks_are_not_listed() {
727
        let dir = tempfile::tempdir().unwrap();
728
        let store = LocalStore::open(dir.path()).unwrap();
729
        store.put_new("packs/a", b"a").unwrap();
730
        fs::write(dir.path().join("packs/.a.tmp-ffffff"), "junk").unwrap();
731
        fs::write(dir.path().join(".refs.lock"), "1").unwrap();
732
        let keys: Vec<String> = store.list("").unwrap().into_iter().map(|e| e.key).collect();
733
        assert_eq!(keys, ["packs/a"]);
734
    }
735
}