OXYGENHUB
# 9aa399044d9d oxygen / crates / oxy-core / src / strata.rs
25.5 KiB · Rust · last changed oxy remote add takes a store's settings; oxy remote check 1 month ago source blame history raw
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(()) => {}
344
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()),
345
            Err(e) => return Err(e.into()),
346
        }
347
        // A directory left empty goes too, up to the root, so the store
348
        // holds nothing but its objects.
349
        let mut dir = path.parent();
350
        while let Some(d) = dir {
351
            if d == self.root || fs::remove_dir(d).is_err() {
352
                break;
353
            }
354
            dir = d.parent();
355
        }
356
        Ok(())
357
    }
358
359
    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> {
360
        let path = self.path(key)?;
361
        match fs::read(&path) {
362
            Ok(b) => {
363
                let v = content_version(&b);
364
                Ok(Some((b, v)))
365
            }
366
            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),
367
            Err(e) => Err(e.into()),
368
        }
369
    }
370
371
    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> {
372
        let path = self.path(key)?;
373
        if let Some(dir) = path.parent() {
374
            fs::create_dir_all(dir)?;
375
        }
376
        let lock = self.lock(key, &path)?;
377
        let result = (|| {
378
            let current = self.read_ref(key)?.map(|(_, v)| v);
379
            check_expected(key, current.as_ref(), expected)?;
380
            crate::durable::write_atomic(&path, bytes)?;
381
            Ok(content_version(bytes))
382
        })();
383
        let _ = fs::remove_file(&lock);
384
        result
385
    }
386
}
387
388
// ---------------------------------------------------------------- memory
389
390
/// A store in memory, for tests: the conditional writes behave as the
391
/// cloud stores' do.
392
#[derive(Default)]
393
pub struct MemoryStore {
394
    objects: Mutex<BTreeMap<String, Vec<u8>>>,
395
}
396
397
impl MemoryStore {
398
    pub fn new() -> MemoryStore {
399
        MemoryStore::default()
400
    }
401
402
    fn map(&self) -> std::sync::MutexGuard<'_, BTreeMap<String, Vec<u8>>> {
403
        self.objects.lock().unwrap_or_else(|p| p.into_inner())
404
    }
405
}
406
407
impl Store for MemoryStore {
408
    fn describe(&self) -> String {
409
        "memory".into()
410
    }
411
412
    fn has(&self, key: &str) -> Result<bool> {
413
        check_key(key)?;
414
        Ok(self.map().contains_key(key))
415
    }
416
417
    fn get(&self, key: &str) -> Result<Vec<u8>> {
418
        check_key(key)?;
419
        self.map()
420
            .get(key)
421
            .cloned()
422
            .ok_or_else(|| Error::NotFound(key.to_string()))
423
    }
424
425
    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {
426
        let all = self.get(key)?;
427
        let len = all.len() as u64;
428
        if range.end > len || range.start > range.end {
429
            return Err(Error::Backend(format!(
430
                "{key}: bytes {}..{} asked of an object of {len} bytes",
431
                range.start, range.end
432
            )));
433
        }
434
        Ok(all[range.start as usize..range.end as usize].to_vec())
435
    }
436
437
    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> {
438
        check_key(key)?;
439
        let mut map = self.map();
440
        if map.contains_key(key) {
441
            return Err(Error::Exists(key.to_string()));
442
        }
443
        map.insert(key.to_string(), bytes.to_vec());
444
        Ok(())
445
    }
446
447
    fn list(&self, prefix: &str) -> Result<Vec<Entry>> {
448
        Ok(self
449
            .map()
450
            .range(prefix.to_string()..)
451
            .take_while(|(k, _)| k.starts_with(prefix))
452
            .map(|(k, v)| Entry {
453
                key: k.clone(),
454
                size: v.len() as u64,
455
            })
456
            .collect())
457
    }
458
459
    fn delete(&self, key: &str) -> Result<()> {
460
        check_key(key)?;
461
        self.map().remove(key);
462
        Ok(())
463
    }
464
465
    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> {
466
        check_key(key)?;
467
        Ok(self.map().get(key).map(|b| (b.clone(), content_version(b))))
468
    }
469
470
    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> {
471
        check_key(key)?;
472
        let mut map = self.map();
473
        let current = map.get(key).map(|b| content_version(b));
474
        check_expected(key, current.as_ref(), expected)?;
475
        map.insert(key.to_string(), bytes.to_vec());
476
        Ok(content_version(bytes))
477
    }
478
}
479
480
// ---------------------------------------------------------------- conformance
481
482
/// Exercise a store under `prefix` (which must be free, and is left
483
/// empty) and report what passed: object round trips, ranges, create-only
484
/// refusal, listing, deletion, and compare-and-swap on a reference. Every
485
/// store runs this in its tests, and `oxy remote check` runs it against a
486
/// remote before trusting it with a repository.
487
pub fn conformance(store: &dyn Store, prefix: &str) -> Result<Vec<&'static str>> {
488
    let mut passed = Vec::new();
489
    let key = |name: &str| format!("{prefix}/{name}");
490
    let fail = |what: &str| Error::Backend(format!("{}: {what}", store.describe()));
491
    let expect = |ok: bool, what: &str| if ok { Ok(()) } else { Err(fail(what)) };
492
493
    expect(
494
        store.list(prefix)?.is_empty(),
495
        "the probe prefix is not empty",
496
    )?;
497
    expect(
498
        !store.has(&key("a"))?,
499
        "has() of a missing object says true",
500
    )?;
501
    expect(
502
        matches!(store.get(&key("a")), Err(Error::NotFound(_))),
503
        "get() of a missing object is not NotFound",
504
    )?;
505
    passed.push("missing objects are reported missing");
506
507
    let a: Vec<u8> = (0..256u32 * 1024).map(|i| (i % 251) as u8).collect();
508
    store.put_new(&key("a"), &a)?;
509
    expect(store.has(&key("a"))?, "has() after put_new says false")?;
510
    expect(
511
        store.get(&key("a"))? == a,
512
        "get() returns other bytes than put_new wrote",
513
    )?;
514
    passed.push("an object round-trips");
515
516
    expect(
517
        store.get_range(&key("a"), 1000..1500)? == a[1000..1500],
518
        "get_range() returns the wrong bytes",
519
    )?;
520
    expect(
521
        store.get_range(&key("a"), 0..0)?.is_empty(),
522
        "an empty range is not empty",
523
    )?;
524
    expect(
525
        store.get_range(&key("a"), 0..a.len() as u64 + 1).is_err(),
526
        "a range past the end is not refused",
527
    )?;
528
    passed.push("ranges read correctly and are bounded");
529
530
    expect(
531
        matches!(store.put_new(&key("a"), b"other"), Err(Error::Exists(_))),
532
        "put_new() over an existing object is not refused",
533
    )?;
534
    expect(
535
        store.get(&key("a"))? == a,
536
        "a refused put_new() changed the object",
537
    )?;
538
    passed.push("objects are written once");
539
540
    store.put_new(&key("dir/b"), b"bb")?;
541
    let listed = store.list(prefix)?;
542
    expect(
543
        listed
544
            == vec![
545
                Entry {
546
                    key: key("a"),
547
                    size: a.len() as u64,
548
                },
549
                Entry {
550
                    key: key("dir/b"),
551
                    size: 2,
552
                },
553
            ],
554
        "list() does not return the two objects, sorted, with their sizes",
555
    )?;
556
    expect(
557
        store.list(&key("dir/"))?
558
            == vec![Entry {
559
                key: key("dir/b"),
560
                size: 2,
561
            }],
562
        "list() of a sub-prefix is wrong",
563
    )?;
564
    expect(
565
        store.list(&key("nothing/"))?.is_empty(),
566
        "list() of an unused prefix is not empty",
567
    )?;
568
    passed.push("listing by prefix");
569
570
    store.delete(&key("dir/b"))?;
571
    expect(
572
        !store.has(&key("dir/b"))?,
573
        "an object is still there after delete()",
574
    )?;
575
    store.delete(&key("dir/b"))?;
576
    passed.push("deletion, and deleting again, succeed");
577
578
    let r = key("refs");
579
    expect(
580
        store.read_ref(&r)?.is_none(),
581
        "an unwritten reference reads as something",
582
    )?;
583
    let v1 = store.write_ref(&r, b"one", None)?;
584
    expect(
585
        matches!(store.write_ref(&r, b"again", None), Err(Error::Conflict(_))),
586
        "a second create-only write of the reference is not refused",
587
    )?;
588
    let (bytes, seen) = store
589
        .read_ref(&r)?
590
        .ok_or_else(|| fail("the reference vanished"))?;
591
    expect(
592
        bytes == b"one" && seen == v1,
593
        "the reference reads back wrong",
594
    )?;
595
    let v2 = store.write_ref(&r, b"two", Some(&v1))?;
596
    expect(v2 != v1, "the version did not change on write")?;
597
    expect(
598
        matches!(
599
            store.write_ref(&r, b"three", Some(&v1)),
600
            Err(Error::Conflict(_))
601
        ),
602
        "a write with a stale version is not refused",
603
    )?;
604
    let (bytes, seen) = store
605
        .read_ref(&r)?
606
        .ok_or_else(|| fail("the reference vanished"))?;
607
    expect(
608
        bytes == b"two" && seen == v2,
609
        "a refused write changed the reference",
610
    )?;
611
    passed.push("references update by compare-and-swap");
612
613
    store.delete(&key("a"))?;
614
    store.delete(&r)?;
615
    expect(
616
        store.list(prefix)?.is_empty(),
617
        "the probe prefix is not empty after cleanup",
618
    )?;
619
    passed.push("cleanup");
620
    Ok(passed)
621
}
622
623
#[cfg(test)]
624
mod tests {
625
    use super::*;
626
    use std::sync::Arc;
627
628
    #[test]
629
    fn keys_are_plain_relative_paths() {
630
        for ok in ["a", "packs/ab.pack", "a-b_c.d/e"] {
631
            check_key(ok).unwrap();
632
        }
633
        for bad in [
634
            "", "/a", "a/", "a//b", "..", "a/../b", "./a", "a b", "a\\b", "ä",
635
        ] {
636
            assert!(check_key(bad).is_err(), "{bad:?}");
637
        }
638
        assert!(check_key(&"x".repeat(MAX_KEY_LEN + 1)).is_err());
639
    }
640
641
    #[test]
642
    fn the_memory_store_conforms() {
643
        let store = MemoryStore::new();
644
        let passed = conformance(&store, "probe").unwrap();
645
        assert_eq!(passed.len(), 8, "{passed:?}");
646
    }
647
648
    #[test]
649
    fn the_directory_store_conforms() {
650
        let dir = tempfile::tempdir().unwrap();
651
        let store = LocalStore::open(dir.path().join("bucket")).unwrap();
652
        let passed = conformance(&store, "repos/one").unwrap();
653
        assert_eq!(passed.len(), 8, "{passed:?}");
654
        // Nothing but the (empty) directories and no stray temporaries.
655
        let stray: Vec<_> = walkdir(dir.path())
656
            .into_iter()
657
            .filter(|p| p.is_file())
658
            .collect();
659
        assert!(stray.is_empty(), "{stray:?}");
660
        assert_eq!(
661
            fs::read_dir(dir.path().join("bucket")).unwrap().count(),
662
            0,
663
            "no empty directories left"
664
        );
665
    }
666
667
    fn walkdir(dir: &Path) -> Vec<PathBuf> {
668
        let mut out = vec![];
669
        if let Ok(rd) = fs::read_dir(dir) {
670
            for e in rd.flatten() {
671
                let p = e.path();
672
                if p.is_dir() {
673
                    out.extend(walkdir(&p));
674
                } else {
675
                    out.push(p);
676
                }
677
            }
678
        }
679
        out
680
    }
681
682
    #[test]
683
    fn a_directory_store_refuses_keys_that_leave_its_root() {
684
        let dir = tempfile::tempdir().unwrap();
685
        let store = LocalStore::open(dir.path().join("bucket")).unwrap();
686
        assert!(matches!(
687
            store.put_new("../escape", b"x"),
688
            Err(Error::Key(..))
689
        ));
690
        assert!(matches!(store.get("/etc/passwd"), Err(Error::Key(..))));
691
        assert!(!dir.path().join("escape").exists());
692
    }
693
694
    /// Many threads race to move a reference from the same version: exactly
695
    /// one wins each round, on both stores.
696
    #[test]
697
    fn compare_and_swap_admits_one_writer_per_round() {
698
        let dir = tempfile::tempdir().unwrap();
699
        let local: Arc<dyn Store> = Arc::new(LocalStore::open(dir.path()).unwrap());
700
        let memory: Arc<dyn Store> = Arc::new(MemoryStore::new());
701
        for store in [local, memory] {
702
            let v0 = store.write_ref("refs", b"0", None).unwrap();
703
            let wins: Vec<bool> = (0..8)
704
                .map(|i| {
705
                    let store = store.clone();
706
                    let v0 = v0.clone();
707
                    std::thread::spawn(move || {
708
                        store
709
                            .write_ref("refs", format!("{}", i + 1).as_bytes(), Some(&v0))
710
                            .is_ok()
711
                    })
712
                })
713
                .collect::<Vec<_>>()
714
                .into_iter()
715
                .map(|h| h.join().unwrap())
716
                .collect();
717
            assert_eq!(
718
                wins.iter().filter(|w| **w).count(),
719
                1,
720
                "{}: {wins:?}",
721
                store.describe()
722
            );
723
            let (bytes, _) = store.read_ref("refs").unwrap().unwrap();
724
            assert_ne!(bytes, b"0");
725
        }
726
    }
727
728
    #[test]
729
    fn a_stale_lock_is_broken_and_a_live_one_waited_for() {
730
        let dir = tempfile::tempdir().unwrap();
731
        let store = LocalStore::open(dir.path()).unwrap();
732
        let lock = dir.path().join(".refs.lock");
733
        fs::write(&lock, "dead").unwrap();
734
        let old = std::time::SystemTime::now() - LOCK_STALE - Duration::from_secs(5);
735
        fs::File::open(&lock).unwrap().set_modified(old).unwrap();
736
        store.write_ref("refs", b"after a crash", None).unwrap();
737
        assert!(!lock.exists());
738
    }
739
740
    #[test]
741
    fn temporaries_and_locks_are_not_listed() {
742
        let dir = tempfile::tempdir().unwrap();
743
        let store = LocalStore::open(dir.path()).unwrap();
744
        store.put_new("packs/a", b"a").unwrap();
745
        fs::write(dir.path().join("packs/.a.tmp-ffffff"), "junk").unwrap();
746
        fs::write(dir.path().join(".refs.lock"), "1").unwrap();
747
        let keys: Vec<String> = store.list("").unwrap().into_iter().map(|e| e.key).collect();
748
        assert_eq!(keys, ["packs/a"]);
749
    }
750
}