OXYGENHUB
# 8156967b882c oxygen / crates / oxy-core / src / store.rs
174.5 KiB · Rust · last changed Fold the older one-list keys on gc and stop reading them 16 days ago source blame history raw
1
//! Object store (spec §13.1, §13.3, §13.4): loose objects, one per key
2
//! under `loose/<hh>/<rest>`, and packs, `pack-<id>.pack` with a sorted
3
//! `pack-<id>.idx`, holding zstd-compressed objects. Reads check loose
4
//! objects first, then every pack; every read re-hashes.
5
//!
6
//! Everything is kept on a [`crate::strata::Store`] (docs/stores.md): a
7
//! directory, which is what `.oxy/objects` is, or a bucket, which is how a
8
//! repository lives in its owner's region with the hub as one client of
9
//! it. The keys are the same wherever the store is, so a bucket holds
10
//! exactly what a directory would, and a directory store is as fast as
11
//! the files were before there was a trait: a loose object is one file, a
12
//! packed one is one ranged read of the pack. What must be local, the
13
//! pack being streamed out and the derived `metas` cache, lives in a
14
//! scratch directory.
15
//!
16
//! Pack file: `OXYPACK1` then entries of
17
//! `type u8 | flags u8 (bit 0: zstd) | stored_len u32 | raw_len u32 | ref 32 | body`.
18
//! Index file: `OXYIDX01` then `count u32` then sorted entries of
19
//! `ref 32 | offset u64 | stored_len u32 | raw_len u32 | type u8 | flags u8`.
20
//!
21
//! Beside the objects: `nocompress/<hex>` marks a chunk packs keep raw;
22
//! `sig/<hh>/<rest>/<key>` holds one signature over an object, written
23
//! once per signer; `seal/<hh>/<rest>/<attest>` and
24
//! `receipt/<hh>/<rest>/<attest>` index the seals covering a change and
25
//! the receipts on a state. Older stores hold these as one list per
26
//! object (`sigs/`, `seals/`, `receipts/`), which are still read.
27
//! `pack-retired` records the packs and loose objects a repack replaced,
28
//! which stay for readers until a collection an hour later (oxygenhub#249).
29
//!
30
//! A store that is not this disk's directory, where finding an object's
31
//! entries is a LIST request, also keeps every entry key in the
32
//! `pack-entries` journal and, once a collection folds them, in entry
33
//! lists (`pack-<id>.entries`), read once per open (spec §13.7, #116).
34
35
use crate::attest::Attest;
36
use crate::hash::{hash_object, header, ObjectRef, ObjectType};
37
use crate::identity::{Identity, Signature};
38
use crate::objects::{Blob, Change, ObjectError, StateObject};
39
use crate::receipt::Receipt;
40
use crate::strata::{self, Error as BackendError, LocalStore, MemoryStore};
41
use rand::RngCore;
42
use std::cell::RefCell;
43
use std::fs;
44
use std::io::{self, Write};
45
use std::path::{Path, PathBuf};
46
use std::sync::Arc;
47
48
const PACK_MAGIC: &[u8; 8] = b"OXYPACK1";
49
const IDX_MAGIC: &[u8; 8] = b"OXYIDX01";
50
const IDX_ENTRY: usize = 32 + 8 + 4 + 4 + 1 + 1;
51
const FLAG_ZSTD: u8 = 1;
52
/// Compressed with the pack's trained dictionary (`pack-ID.dict`), which a
53
/// full repack makes from a sample of the objects it packs: small files
54
/// share most of their vocabulary, which each one alone cannot use.
55
const FLAG_DICT: u8 = 2;
56
/// The size of a pack's dictionary, and the least a repack must pack for
57
/// one to pay.
58
const DICT_BYTES: usize = 110 << 10;
59
const DICT_MIN_OBJECTS: usize = 1000;
60
/// Objects up to this size are compressed with the dictionary.
61
const DICT_OBJECT_MAX: usize = 64 << 10;
62
const ZSTD_LEVEL: i32 = 3;
63
/// The level of a full repack (`oxy gc`): slower, once, for a smaller store.
64
pub const REPACK_ZSTD_LEVEL: i32 = 9;
65
/// Largest object a pack entry may claim; anything bigger is corruption,
66
/// and refusing it keeps a bad index from driving huge allocations.
67
const MAX_OBJECT_LEN: u32 = 1 << 30;
68
/// What reading an entry of a pack whose dictionary is gone says. A pack
69
/// being deleted can lose its dictionary first, so a read that meets this
70
/// looks at the packs again before it is believed.
71
const NO_DICT: &str = "packed with a dictionary the pack lacks";
72
73
/// The record of what repacks replaced (oxygenhub#249): packs and loose
74
/// objects whose contents a newer pack holds, kept for readers that
75
/// still use them until garbage collection deletes them, at least
76
/// [`RETIRED_FOR`] later. One line per key, `<retired at, wall ms> <key>`.
77
/// It is a reference record (updated by compare-and-swap, so concurrent
78
/// repacks and collections never lose each other's lines), named under
79
/// `pack-` so the pack listing every open makes shows whether it holds
80
/// anything.
81
const RETIRED_KEY: &str = "pack-retired";
82
/// How long a retired pack or loose object stays in the store: a reader
83
/// that loaded its list of packs before a repack reads on from the old
84
/// packs this long, and after that finds the objects in the new pack by
85
/// looking again (see [`Store::get`]).
86
pub const RETIRED_FOR: std::time::Duration = std::time::Duration::from_secs(60 * 60);
87
/// Compare-and-swap rounds for the retired record before giving up.
88
const RETIRED_TRIES: usize = 32;
89
/// After a look at the pack listing found nothing new, how long a missed
90
/// [`Store::has`] answers without looking again: a writer probing for
91
/// thousands of objects it is about to store would otherwise list the
92
/// packs for each.
93
const HAS_RELIST_AFTER: std::time::Duration = std::time::Duration::from_secs(1);
94
95
/// The journal of signatures and index entries (spec §13.7): every entry
96
/// key a writer on a store that is not this disk's directory wrote since
97
/// the last fold, one line each, `<folded at, wall ms; 0 until folded>
98
/// <key>`, as the retired record's lines are. A reference record updated
99
/// by compare-and-swap, so concurrent writers never lose each other's
100
/// lines; read once per open, instead of a listing per lookup.
101
pub const JOURNAL_KEY: &str = "pack-entries";
102
/// A complete entry list (spec §13.7): every entry a fold found.
103
const LIST_SUFFIX: &str = ".entries";
104
/// A partial entry list: journal lines a writer folded out.
105
const PART_SUFFIX: &str = ".entries-part";
106
/// The first line of an entry list.
107
const LIST_MAGIC: &str = "oxy-entries 1";
108
/// Written once a collection has copied the older one-list keys into
109
/// per-object entries: readers stop looking for them.
110
const LEGACY_FOLDED_KEY: &str = "pack-legacy-folded";
111
/// Unfolded journal lines a writer lets the journal hold before it folds
112
/// them into a partial list.
113
const JOURNAL_FOLD_AT: usize = 1000;
114
/// The per-object entry indexes, each with the one-list name older
115
/// stores kept it under.
116
const ENTRY_INDEXES: [(&str, &str); 6] = [
117
    ("sig", "sigs"),
118
    ("seal", "seals"),
119
    ("receipt", "receipts"),
120
    ("approval", "approvals"),
121
    ("cert", "certs"),
122
    ("accepted", "accepted-lists"),
123
];
124
125
/// A pack listing as compared: each key with its size and time.
126
type Listed = Vec<(String, u64, Option<u64>)>;
127
128
/// The retired record's lines: each key with the time it was retired.
129
fn parse_retired(bytes: &[u8]) -> std::collections::BTreeMap<String, u64> {
130
    String::from_utf8_lossy(bytes)
131
        .lines()
132
        .filter_map(|l| {
133
            let (t, k) = l.split_once(' ')?;
134
            Some((k.to_string(), t.parse().ok()?))
135
        })
136
        .collect()
137
}
138
139
fn encode_retired(m: &std::collections::BTreeMap<String, u64>) -> Vec<u8> {
140
    m.iter()
141
        .map(|(k, t)| format!("{t} {k}\n"))
142
        .collect::<String>()
143
        .into_bytes()
144
}
145
146
/// What deleting retired packs and loose objects did.
147
#[derive(Debug, Clone, Default, PartialEq, Eq)]
148
pub struct RetiredReport {
149
    /// Retired packs deleted, each with its index and dictionary.
150
    pub packs: usize,
151
    /// Retired loose objects deleted.
152
    pub loose: usize,
153
    /// Retired entries younger than [`RETIRED_FOR`], left for the next
154
    /// collection.
155
    pub kept: usize,
156
}
157
158
/// An object read from the store: its type and body.
159
pub type Object = (ObjectType, Vec<u8>);
160
161
/// Where a long operation reports what it is doing (oxygen#66): a phase,
162
/// or a count within one, as a line for a spinner or a log.
163
pub type Progress<'a> = &'a mut dyn FnMut(&str);
164
165
/// The storage under a store: objects by key, see [`crate::strata`].
166
pub type Backend = dyn strata::Store;
167
168
fn silent() -> impl FnMut(&str) {
169
    |_| {}
170
}
171
172
/// A backend error as a store error: what is not found or already there
173
/// is handled where it matters; the rest is an I/O failure.
174
fn backend_err(e: BackendError) -> StoreError {
175
    match e {
176
        BackendError::Io(e) => StoreError::Io(e),
177
        other => StoreError::Io(io::Error::other(other.to_string())),
178
    }
179
}
180
181
/// `put_new` where an object already there is what we wanted.
182
fn put_new_or_existing(backend: &Backend, key: &str, bytes: &[u8]) -> Result<()> {
183
    match backend.put_new(key, bytes) {
184
        Ok(()) | Err(BackendError::Exists(_)) => Ok(()),
185
        Err(e) => Err(backend_err(e)),
186
    }
187
}
188
189
/// `get` as an option: `None` when there is no such key.
190
fn get_optional(backend: &Backend, key: &str) -> Result<Option<Vec<u8>>> {
191
    match backend.get(key) {
192
        Ok(b) => Ok(Some(b)),
193
        Err(BackendError::NotFound(_)) => Ok(None),
194
        Err(e) => Err(backend_err(e)),
195
    }
196
}
197
198
fn delete(backend: &Backend, key: &str) -> Result<()> {
199
    backend.delete(key).map_err(backend_err)
200
}
201
202
/// `loose/<hh>/<rest>`.
203
fn loose_key(r: &ObjectRef) -> String {
204
    let h = r.to_hex();
205
    format!("loose/{}/{}", &h[..2], &h[2..])
206
}
207
208
/// The object a loose key names, if it is one.
209
fn loose_ref(key: &str) -> Option<ObjectRef> {
210
    let rest = key.strip_prefix("loose/")?;
211
    let (hh, tail) = rest.split_once('/')?;
212
    if hh.len() != 2 || tail.contains('/') {
213
        return None;
214
    }
215
    ObjectRef::from_hex(&format!("{hh}{tail}"))
216
}
217
218
/// `<index>/<hh>/<rest>/` for the entries kept under an object.
219
fn fanout_prefix(index: &str, r: &ObjectRef) -> String {
220
    let h = r.to_hex();
221
    format!("{index}/{}/{}/", &h[..2], &h[2..])
222
}
223
224
/// Whether a key names an entry list (spec §13.7), complete or partial.
225
fn is_entry_list(key: &str) -> bool {
226
    key.starts_with("pack-") && (key.ends_with(LIST_SUFFIX) || key.ends_with(PART_SUFFIX))
227
}
228
229
/// An entry list's bytes: zstd-compressed text, [`LIST_MAGIC`] and then
230
/// one key per line, sorted.
231
fn encode_entry_list<'a>(keys: impl Iterator<Item = &'a String>) -> Result<Vec<u8>> {
232
    let mut text = String::from(LIST_MAGIC);
233
    text.push('\n');
234
    for k in keys {
235
        text.push_str(k);
236
        text.push('\n');
237
    }
238
    Ok(zstd::bulk::compress(text.as_bytes(), ZSTD_LEVEL)?)
239
}
240
241
/// The keys an entry list holds.
242
fn decode_entry_list(key: &str, bytes: &[u8]) -> Result<Vec<String>> {
243
    let bad = |why: &str| StoreError::Io(io::Error::other(format!("{key}: {why}")));
244
    let text = zstd::stream::decode_all(bytes).map_err(|_| bad("not zstd"))?;
245
    let text = String::from_utf8(text).map_err(|_| bad("not UTF-8"))?;
246
    let mut lines = text.lines();
247
    if lines.next() != Some(LIST_MAGIC) {
248
        return Err(bad("not an entry list"));
249
    }
250
    Ok(lines
251
        .filter(|l| !l.is_empty())
252
        .map(str::to_string)
253
        .collect())
254
}
255
256
/// Drop journal lines folded at least [`RETIRED_FOR`] before `now`:
257
/// every reader that could need them has a list holding them by then.
258
fn expire_folded(m: &mut std::collections::BTreeMap<String, u64>, now: u64) {
259
    let hour = RETIRED_FOR.as_millis() as u64;
260
    m.retain(|_, t| *t == 0 || now.saturating_sub(*t) < hour);
261
}
262
263
/// `<index>/<hh>/<rest>`: the one-file-per-object layout of older stores.
264
fn legacy_key(index: &str, r: &ObjectRef) -> String {
265
    let h = r.to_hex();
266
    format!("{index}/{}/{}", &h[..2], &h[2..])
267
}
268
269
fn now_ms() -> u64 {
270
    std::time::SystemTime::now()
271
        .duration_since(std::time::UNIX_EPOCH)
272
        .map(|d| d.as_millis() as u64)
273
        .unwrap_or(0)
274
}
275
276
#[derive(Clone, Copy, Debug)]
277
struct PackEntry {
278
    offset: u64,
279
    stored_len: u32,
280
    raw_len: u32,
281
    ty: ObjectType,
282
    flags: u8,
283
}
284
285
/// One pack's index, fully in memory and sorted by ObjectRef.
286
struct PackIndex {
287
    /// The pack's key, `pack-<id>.pack`.
288
    key: String,
289
    refs: Vec<ObjectRef>,
290
    entries: Vec<PackEntry>,
291
    /// The pack's dictionary, prepared once, when it has one.
292
    dict: Option<Arc<zstd::dict::DecoderDictionary<'static>>>,
293
}
294
295
impl std::fmt::Debug for PackIndex {
296
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
297
        f.debug_struct("PackIndex")
298
            .field("key", &self.key)
299
            .field("objects", &self.refs.len())
300
            .field("dict", &self.dict.is_some())
301
            .finish()
302
    }
303
}
304
305
/// Decode a stored pack entry and check it against its reference.
306
fn decode_entry(
307
    r: &ObjectRef,
308
    e: PackEntry,
309
    stored: Vec<u8>,
310
    dict: Option<&Arc<zstd::dict::DecoderDictionary<'static>>>,
311
) -> Result<Object> {
312
    let body = if e.flags & FLAG_DICT != 0 {
313
        let dict = dict.ok_or_else(|| StoreError::Corrupt(*r, NO_DICT.into()))?;
314
        zstd::bulk::Decompressor::with_prepared_dictionary(dict)
315
            .and_then(|mut d| d.decompress(&stored, e.raw_len as usize))
316
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
317
    } else if e.flags & FLAG_ZSTD != 0 {
318
        zstd::bulk::decompress(&stored, e.raw_len as usize)
319
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
320
    } else {
321
        stored
322
    };
323
    if body.len() != e.raw_len as usize || hash_object(e.ty, &body) != *r {
324
        return Err(StoreError::Corrupt(
325
            *r,
326
            "packed object does not match its reference".into(),
327
        ));
328
    }
329
    Ok((e.ty, body))
330
}
331
332
impl PackIndex {
333
    /// Load a pack's index; `pack_len` is the pack's size, from the listing
334
    /// that found it (a stat of it would be a request of its own). `None`
335
    /// when the index is gone since the listing: the pack was deleted.
336
    fn load(
337
        backend: &Backend,
338
        idx_key: &str,
339
        pack_key: &str,
340
        pack_len: u64,
341
    ) -> Result<Option<PackIndex>> {
342
        let data = match backend.get(idx_key) {
343
            Ok(d) => d,
344
            Err(BackendError::NotFound(_)) => return Ok(None),
345
            Err(e) => return Err(backend_err(e)),
346
        };
347
        let bad = |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{idx_key}: {m}"));
348
        if data.len() < 12 || &data[..8] != IDX_MAGIC {
349
            return Err(bad("bad index header"));
350
        }
351
        let count = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize;
352
        if data.len() != 12 + count * IDX_ENTRY {
353
            return Err(bad("index length mismatch"));
354
        }
355
        let mut refs = Vec::with_capacity(count);
356
        let mut entries = Vec::with_capacity(count);
357
        for i in 0..count {
358
            let e = &data[12 + i * IDX_ENTRY..12 + (i + 1) * IDX_ENTRY];
359
            let r = ObjectRef::from_slice(&e[..32]).unwrap();
360
            if let Some(prev) = refs.last() {
361
                if r <= *prev {
362
                    return Err(bad("index not sorted"));
363
                }
364
            }
365
            refs.push(r);
366
            let entry = PackEntry {
367
                offset: u64::from_le_bytes(e[32..40].try_into().unwrap()),
368
                stored_len: u32::from_le_bytes(e[40..44].try_into().unwrap()),
369
                raw_len: u32::from_le_bytes(e[44..48].try_into().unwrap()),
370
                ty: ObjectType::ALL
371
                    .get(e[48] as usize)
372
                    .copied()
373
                    .ok_or_else(|| bad("unknown object type"))?,
374
                flags: e[49],
375
            };
376
            if entry.stored_len > MAX_OBJECT_LEN || entry.raw_len > MAX_OBJECT_LEN {
377
                return Err(StoreError::Corrupt(
378
                    r,
379
                    format!(
380
                        "{idx_key}: the index claims an object of {} bytes ({} stored), more than the {MAX_OBJECT_LEN} an object may be",
381
                        entry.raw_len, entry.stored_len
382
                    ),
383
                ));
384
            }
385
            if entry.offset.saturating_add(entry.stored_len as u64) > pack_len {
386
                return Err(bad("index entry points outside the pack"));
387
            }
388
            entries.push(entry);
389
        }
390
        let dict_key = format!("{}.dict", pack_key.trim_end_matches(".pack"));
391
        let dict = get_optional(backend, &dict_key)?
392
            .map(|d| Arc::new(zstd::dict::DecoderDictionary::copy(&d)));
393
        Ok(Some(PackIndex {
394
            key: pack_key.to_string(),
395
            refs,
396
            entries,
397
            dict,
398
        }))
399
    }
400
401
    fn find(&self, r: &ObjectRef) -> Option<PackEntry> {
402
        self.refs.binary_search(r).ok().map(|i| self.entries[i])
403
    }
404
405
    /// Read one entry from the pack: one ranged read, then decode.
406
    fn fetch(&self, backend: &Backend, r: &ObjectRef, e: PackEntry) -> Result<Object> {
407
        let stored = backend
408
            .get_range(&self.key, e.offset..e.offset + e.stored_len as u64)
409
            .map_err(|err| match err {
410
                BackendError::NotFound(_) => StoreError::NotFound(*r),
411
                other => backend_err(other),
412
            })?;
413
        decode_entry(r, e, stored, self.dict.as_ref())
414
    }
415
}
416
417
fn type_index(ty: ObjectType) -> u8 {
418
    ObjectType::ALL.iter().position(|t| *t == ty).unwrap() as u8
419
}
420
421
/// What `pack_loose` did.
422
#[derive(Debug, Clone, Default, PartialEq, Eq)]
423
pub struct PackReport {
424
    pub objects: usize,
425
    pub raw_bytes: u64,
426
    pub packed_bytes: u64,
427
}
428
429
/// Packs tolerated before `merge_packs` folds them into one (spec §13.4).
430
pub const MAX_PACKS: usize = 16;
431
432
/// What `prune` did.
433
#[derive(Debug, Clone, Default, PartialEq, Eq)]
434
pub struct PruneReport {
435
    pub objects: usize,
436
    pub bytes: u64,
437
    /// Unreachable objects left alone because they are younger than the grace period.
438
    pub deferred: usize,
439
}
440
441
#[derive(Debug, thiserror::Error)]
442
pub enum StoreError {
443
    #[error("io: {0}")]
444
    Io(#[from] io::Error),
445
    #[error("corrupt object {0}: {1}")]
446
    Corrupt(ObjectRef, String),
447
    #[error("object {0}")]
448
    Object(#[from] ObjectError),
449
    #[error("object {0} is a {1}, expected {2}")]
450
    WrongType(ObjectRef, &'static str, &'static str),
451
    #[error("object {0} not found")]
452
    NotFound(ObjectRef),
453
    #[error("a {kind} object of {size} bytes is larger than the {limit} bytes an object may be; nothing was stored. A change this large adds or touches too many files at once: commit it in parts (a directory at a time)")]
454
    TooLarge {
455
        kind: &'static str,
456
        size: usize,
457
        limit: u32,
458
    },
459
}
460
461
pub type Result<T> = std::result::Result<T, StoreError>;
462
463
pub struct Store {
464
    backend: Arc<Backend>,
465
    /// Temporary files (a pack being streamed) and derived caches (`metas`).
466
    /// For a store in a directory it is that directory.
467
    scratch: PathBuf,
468
    /// The backend is that directory, so the old file paths still hold.
469
    local: bool,
470
    packs: RefCell<Vec<Arc<PackIndex>>>,
471
    /// The pack listing `packs` was loaded from (key, size, time), to tell
472
    /// whether a look at the store would find anything new.
473
    packs_listed: RefCell<Listed>,
474
    /// When a look at the pack listing for a missed [`Store::has`] last
475
    /// found nothing new.
476
    has_relisted: std::cell::Cell<Option<std::time::Instant>>,
477
    /// An open write batch (spec §13.4): objects go here instead of to loose
478
    /// files, and are written out with one `fsync` when the batch finishes.
479
    pending: RefCell<Option<Pending>>,
480
    batch_depth: std::cell::Cell<u32>,
481
    /// Change metadata already read, from the `metas` cache file.
482
    metas: RefCell<Option<crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta>>>,
483
    /// The live entry lists (spec §13.7) the pack listing showed: listed
484
    /// and not retired.
485
    entry_lists: RefCell<Vec<String>>,
486
    /// Whether the store is folded: a live entry list is complete.
487
    folded: std::cell::Cell<bool>,
488
    /// Whether the pack listing showed `pack-legacy-folded`.
489
    legacy_folded: std::cell::Cell<bool>,
490
    /// The entry keys of a folded store, read from its lists and journal
491
    /// on the first lookup after the packs were loaded.
492
    entries: RefCell<Option<std::collections::BTreeSet<String>>>,
493
    /// Entry keys written in an open batch, added to the journal when the
494
    /// batch is flushed.
495
    journal_queue: RefCell<Vec<String>>,
496
}
497
498
impl std::fmt::Debug for Store {
499
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
500
        f.debug_struct("Store")
501
            .field("backend", &self.backend.describe())
502
            .field("scratch", &self.scratch)
503
            .field("packs", &self.packs.borrow().len())
504
            .finish()
505
    }
506
}
507
508
/// Objects written since `begin_batch`. Small batches stay in memory and
509
/// become loose objects; once `SPILL_OBJECTS` or `SPILL_BYTES` is exceeded
510
/// the batch streams into a pack file instead.
511
#[derive(Debug)]
512
struct Pending {
513
    buffered: Vec<(ObjectType, Vec<u8>)>,
514
    buffered_bytes: usize,
515
    lookup: std::collections::HashMap<ObjectRef, PendingEntry>,
516
    pack: Option<PendingPack>,
517
    report: PackReport,
518
}
519
520
#[derive(Debug, Clone, Copy)]
521
enum PendingEntry {
522
    Buffered(usize),
523
    Packed(PackEntry),
524
}
525
526
#[derive(Debug)]
527
struct PendingPack {
528
    id: String,
529
    tmp_path: PathBuf,
530
    writer: io::BufWriter<fs::File>,
531
    offset: u64,
532
    index: Vec<(ObjectRef, PackEntry)>,
533
}
534
535
const SPILL_OBJECTS: usize = 256;
536
const SPILL_BYTES: usize = 8 << 20;
537
538
/// Finishes the enclosing batch when dropped; call `finish` to see errors.
539
pub struct Batch<'a> {
540
    store: &'a Store,
541
    done: bool,
542
}
543
544
impl Batch<'_> {
545
    pub fn finish(mut self) -> Result<()> {
546
        self.done = true;
547
        self.store.finish_batch()
548
    }
549
}
550
551
impl Drop for Batch<'_> {
552
    fn drop(&mut self) {
553
        if !self.done {
554
            let _ = self.store.finish_batch();
555
        }
556
    }
557
}
558
559
/// Append one stored object to a pending pack and index it.
560
fn write_pack_entry(
561
    pack: &mut PendingPack,
562
    r: &ObjectRef,
563
    ty: ObjectType,
564
    raw_len: usize,
565
    flags: u8,
566
    stored: &[u8],
567
    report: &mut PackReport,
568
) -> Result<PackEntry> {
569
    let header_len = 1 + 1 + 4 + 4 + 32;
570
    pack.writer.write_all(&[type_index(ty), flags])?;
571
    pack.writer
572
        .write_all(&(stored.len() as u32).to_le_bytes())?;
573
    pack.writer.write_all(&(raw_len as u32).to_le_bytes())?;
574
    pack.writer.write_all(&r.0)?;
575
    pack.writer.write_all(stored)?;
576
    let entry = PackEntry {
577
        offset: pack.offset + header_len,
578
        stored_len: stored.len() as u32,
579
        raw_len: raw_len as u32,
580
        ty,
581
        flags,
582
    };
583
    pack.offset += header_len + stored.len() as u64;
584
    pack.index.push((*r, entry));
585
    report.objects += 1;
586
    report.raw_bytes += raw_len as u64;
587
    report.packed_bytes += stored.len() as u64;
588
    Ok(entry)
589
}
590
591
/// A random name for a temporary file or a pack.
592
fn fresh_id() -> String {
593
    let mut nonce = [0u8; 8];
594
    rand::rngs::OsRng.fill_bytes(&mut nonce);
595
    hex::encode(nonce)
596
}
597
598
/// Under `OXY_STORE_BACKEND=memory` every store opens on a store in
599
/// memory shared by directory, so the whole test suite can run without a
600
/// directory store, as it would over a bucket. Objects live for the
601
/// process.
602
fn memory_backend_for(dir: &Path) -> Option<Arc<Backend>> {
603
    use std::collections::HashMap;
604
    use std::sync::{Mutex, OnceLock};
605
    if std::env::var_os("OXY_STORE_BACKEND").as_deref() != Some(std::ffi::OsStr::new("memory")) {
606
        return None;
607
    }
608
    static STORES: OnceLock<Mutex<HashMap<PathBuf, Arc<MemoryStore>>>> = OnceLock::new();
609
    let key = dir.canonicalize().unwrap_or_else(|_| dir.to_path_buf());
610
    let mut map = STORES
611
        .get_or_init(Default::default)
612
        .lock()
613
        .unwrap_or_else(|p| p.into_inner());
614
    let store = map
615
        .entry(key)
616
        .or_insert_with(|| Arc::new(MemoryStore::new()))
617
        .clone();
618
    Some(store)
619
}
620
621
#[cfg(test)]
622
thread_local! {
623
    /// A backend every store opened on this thread uses instead of its
624
    /// directory: tests of a repository kept on a store that is not this
625
    /// disk's directory, with several handles on one store.
626
    pub(crate) static TEST_BACKEND: RefCell<Option<Arc<Backend>>> = const { RefCell::new(None) };
627
}
628
629
impl Store {
630
    /// Open the store in `dir`, which is also its scratch directory.
631
    pub fn open(dir: impl Into<PathBuf>) -> io::Result<Store> {
632
        let dir = dir.into();
633
        fs::create_dir_all(dir.join("loose"))?;
634
        #[cfg(test)]
635
        if let Some(backend) = TEST_BACKEND.with(|t| t.borrow().clone()) {
636
            return Store::open_on(backend, dir);
637
        }
638
        if let Some(backend) = memory_backend_for(&dir) {
639
            return Store::open_on(backend, dir);
640
        }
641
        let backend: Arc<Backend> =
642
            Arc::new(LocalStore::open(&dir).map_err(|e| io::Error::other(e.to_string()))?);
643
        let mut store = Store::open_on(backend, dir)?;
644
        store.local = true;
645
        Ok(store)
646
    }
647
648
    /// Open the store kept on `backend`, with `scratch` for what must be
649
    /// on this disk: a pack while it is being written, and caches.
650
    pub fn open_on(backend: Arc<Backend>, scratch: impl Into<PathBuf>) -> io::Result<Store> {
651
        let scratch = scratch.into();
652
        fs::create_dir_all(&scratch)?;
653
        let store = Store {
654
            backend,
655
            scratch,
656
            local: false,
657
            packs: RefCell::new(Vec::new()),
658
            packs_listed: RefCell::new(Vec::new()),
659
            has_relisted: std::cell::Cell::new(None),
660
            pending: RefCell::new(None),
661
            batch_depth: std::cell::Cell::new(0),
662
            metas: RefCell::new(None),
663
            entry_lists: RefCell::new(Vec::new()),
664
            folded: std::cell::Cell::new(false),
665
            legacy_folded: std::cell::Cell::new(false),
666
            entries: RefCell::new(None),
667
            journal_queue: RefCell::new(Vec::new()),
668
        };
669
        store
670
            .reload_packs()
671
            .map_err(|e| io::Error::other(e.to_string()))?;
672
        Ok(store)
673
    }
674
675
    /// The storage under this store.
676
    pub fn backend(&self) -> &Arc<Backend> {
677
        &self.backend
678
    }
679
680
    /// Whether the store is the directory `dir()` names, where every key
681
    /// is a file.
682
    pub fn is_local(&self) -> bool {
683
        self.local
684
    }
685
686
    /// Load the list of packs again, from a new listing: what other
687
    /// writers packed since becomes readable, and packs since retired or
688
    /// deleted are let go. Packs already loaded are kept, not read again.
689
    pub fn refresh(&self) -> Result<()> {
690
        *self.entries.borrow_mut() = None;
691
        self.reload_packs()
692
    }
693
694
    /// The pack listing, with how `packs_listed` records it.
695
    fn list_packs(&self) -> Result<(Vec<strata::Entry>, Listed)> {
696
        let listed = self.backend.list("pack-").map_err(backend_err)?;
697
        let sig = listed
698
            .iter()
699
            .map(|e| (e.key.clone(), e.size, e.modified_ms))
700
            .collect();
701
        Ok((listed, sig))
702
    }
703
704
    /// Load every pack the listing names that is not retired. An index
705
    /// that is gone by the time it is read (its pack deleted after the
706
    /// listing) is passed over, and the store listed once more, since
707
    /// what replaced it may be new since the first listing.
708
    fn reload_packs(&self) -> Result<()> {
709
        let (listed, sig) = self.list_packs()?;
710
        self.load_listed(listed, sig)
711
    }
712
713
    fn load_listed(&self, mut listed: Vec<strata::Entry>, mut sig: Listed) -> Result<()> {
714
        let had: std::collections::HashMap<String, Arc<PackIndex>> = self
715
            .packs
716
            .borrow()
717
            .iter()
718
            .map(|p| (p.key.clone(), p.clone()))
719
            .collect();
720
        let mut loaded: std::collections::HashMap<String, Arc<PackIndex>> = Default::default();
721
        for attempt in 0..2 {
722
            let retired = self.retired_in(&listed)?;
723
            let mut idx_keys: Vec<&str> = listed
724
                .iter()
725
                .map(|e| e.key.as_str())
726
                .filter(|k| k.ends_with(".idx"))
727
                .collect();
728
            idx_keys.sort();
729
            let mut packs = Vec::new();
730
            let mut vanished = false;
731
            for idx in idx_keys {
732
                let pack = format!("{}.pack", idx.trim_end_matches(".idx"));
733
                if retired.contains_key(&pack) {
734
                    continue;
735
                }
736
                let Some(p) = listed.iter().find(|e| e.key == pack) else {
737
                    continue;
738
                };
739
                if let Some(kept) = had.get(&pack).or_else(|| loaded.get(&pack)) {
740
                    packs.push(kept.clone());
741
                    continue;
742
                }
743
                match PackIndex::load(self.backend.as_ref(), idx, &pack, p.size)? {
744
                    Some(index) => {
745
                        let index = Arc::new(index);
746
                        loaded.insert(pack, index.clone());
747
                        packs.push(index);
748
                    }
749
                    None => vanished = true,
750
                }
751
            }
752
            if vanished && attempt == 0 {
753
                (listed, sig) = self.list_packs()?;
754
                continue;
755
            }
756
            let lists: Vec<String> = listed
757
                .iter()
758
                .map(|e| e.key.as_str())
759
                .filter(|k| is_entry_list(k) && !retired.contains_key(*k))
760
                .map(str::to_string)
761
                .collect();
762
            self.folded
763
                .set(lists.iter().any(|k| k.ends_with(LIST_SUFFIX)));
764
            self.legacy_folded
765
                .set(listed.iter().any(|e| e.key == LEGACY_FOLDED_KEY));
766
            *self.entry_lists.borrow_mut() = lists;
767
            if *self.packs_listed.borrow() != sig {
768
                *self.entries.borrow_mut() = None;
769
            }
770
            *self.packs.borrow_mut() = packs;
771
            *self.packs_listed.borrow_mut() = sig;
772
            return Ok(());
773
        }
774
        unreachable!("the second round always returns")
775
    }
776
777
    /// Load the packs again when the pack listing differs from the one
778
    /// they were loaded from; whether it did. What a miss calls before it
779
    /// is believed (oxygenhub#249): another writer may have packed the
780
    /// object, or a repack moved it, since this store loaded its packs.
781
    fn reload_if_changed(&self) -> Result<bool> {
782
        let (listed, sig) = self.list_packs()?;
783
        if *self.packs_listed.borrow() == sig {
784
            return Ok(false);
785
        }
786
        self.load_listed(listed, sig)?;
787
        Ok(true)
788
    }
789
790
    /// The retired record, read when `listed` (a pack listing) shows it
791
    /// holds anything.
792
    fn retired_in(
793
        &self,
794
        listed: &[strata::Entry],
795
    ) -> Result<std::collections::BTreeMap<String, u64>> {
796
        match listed.iter().find(|e| e.key == RETIRED_KEY) {
797
            Some(e) if e.size > 0 => Ok(self.read_retired()?.0),
798
            _ => Ok(Default::default()),
799
        }
800
    }
801
802
    /// The retired record and its version (`None`: never written).
803
    fn read_retired(
804
        &self,
805
    ) -> Result<(
806
        std::collections::BTreeMap<String, u64>,
807
        Option<strata::Version>,
808
    )> {
809
        self.read_record(RETIRED_KEY)
810
    }
811
812
    /// A record of `<wall ms> <key>` lines (the retired record, the
813
    /// journal) and its version (`None`: never written).
814
    fn read_record(
815
        &self,
816
        key: &str,
817
    ) -> Result<(
818
        std::collections::BTreeMap<String, u64>,
819
        Option<strata::Version>,
820
    )> {
821
        match self.backend.read_ref(key).map_err(backend_err)? {
822
            Some((bytes, v)) => Ok((parse_retired(&bytes), Some(v))),
823
            None => Ok((Default::default(), None)),
824
        }
825
    }
826
827
    /// Change the retired record by compare-and-swap, reading it again
828
    /// whenever another writer changed it first. `edit` must give the
829
    /// same result however often it runs.
830
    fn update_retired(
831
        &self,
832
        edit: impl FnMut(&mut std::collections::BTreeMap<String, u64>),
833
    ) -> Result<()> {
834
        self.update_record(RETIRED_KEY, "the record of retired packs", edit)
835
    }
836
837
    /// Change a record of `<wall ms> <key>` lines by compare-and-swap, as
838
    /// [`Store::update_retired`] does; `what` names it in the error when
839
    /// other writers kept winning.
840
    fn update_record(
841
        &self,
842
        key: &str,
843
        what: &str,
844
        mut edit: impl FnMut(&mut std::collections::BTreeMap<String, u64>),
845
    ) -> Result<()> {
846
        for _ in 0..RETIRED_TRIES {
847
            let (mut m, version) = self.read_record(key)?;
848
            let before = m.clone();
849
            edit(&mut m);
850
            // Unchanged; a record never written stays unwritten.
851
            if m == before && (version.is_some() || m.is_empty()) {
852
                return Ok(());
853
            }
854
            match self
855
                .backend
856
                .write_ref(key, &encode_retired(&m), version.as_ref())
857
            {
858
                Ok(_) => return Ok(()),
859
                Err(BackendError::Conflict(_)) => continue,
860
                Err(e) => return Err(backend_err(e)),
861
            }
862
        }
863
        Err(StoreError::Io(io::Error::other(format!(
864
            "other writers kept changing {what}; run the command again"
865
        ))))
866
    }
867
868
    /// Record `keys` (packs, by their `.pack` key, and loose objects) as
869
    /// retired now: kept for the readers still using them, and deleted by
870
    /// a collection at least [`RETIRED_FOR`] later. A key already retired
871
    /// keeps its first time.
872
    fn retire(&self, keys: &[String]) -> Result<()> {
873
        if keys.is_empty() {
874
            return Ok(());
875
        }
876
        let now = now_ms();
877
        self.update_retired(|m| {
878
            for k in keys {
879
                m.entry(k.clone()).or_insert(now);
880
            }
881
        })
882
    }
883
884
    /// Delete what was retired at least `older_than` ago (a collection
885
    /// does, with [`RETIRED_FOR`]): each pack with its index and
886
    /// dictionary, and each loose object a pack now holds. A retired loose
887
    /// object no pack holds any more (the pack that had it was pruned, and
888
    /// the object stored again since) is kept, as an ordinary loose object.
889
    pub fn expire_retired(
890
        &self,
891
        older_than: std::time::Duration,
892
        progress: Progress,
893
    ) -> Result<RetiredReport> {
894
        let (listed, _) = self.list_packs()?;
895
        let retired = self.retired_in(&listed)?;
896
        let mut report = RetiredReport::default();
897
        if retired.is_empty() {
898
            return Ok(report);
899
        }
900
        let now = now_ms();
901
        let older_than = older_than.as_millis() as u64;
902
        let due: Vec<(String, u64)> = retired
903
            .iter()
904
            .filter(|(_, t)| now.saturating_sub(**t) >= older_than)
905
            .map(|(k, t)| (k.clone(), *t))
906
            .collect();
907
        report.kept = retired.len() - due.len();
908
        if due.is_empty() {
909
            return Ok(report);
910
        }
911
        progress(&format!("deleting {} retired files", due.len()));
912
        // Which objects the live packs hold.
913
        self.reload_packs()?;
914
        for (key, _) in &due {
915
            if key.ends_with(".pack") {
916
                self.remove_pack(key)?;
917
                report.packs += 1;
918
            } else if is_entry_list(key) {
919
                // An entry list a fold replaced (spec §13.7).
920
                delete(self.backend.as_ref(), key)?;
921
            } else if let Some(r) = loose_ref(key) {
922
                if self.find_packed(&r).is_some() {
923
                    delete(self.backend.as_ref(), key)?;
924
                    report.loose += 1;
925
                }
926
            }
927
        }
928
        self.update_retired(|m| {
929
            for (k, t) in &due {
930
                if m.get(k) == Some(t) {
931
                    m.remove(k);
932
                }
933
            }
934
        })?;
935
        self.reload_packs()?;
936
        Ok(report)
937
    }
938
939
    /// The loose objects retired and not yet deleted, by key.
940
    fn retired_loose(&self) -> Result<std::collections::HashSet<String>> {
941
        let (listed, _) = self.list_packs()?;
942
        Ok(self
943
            .retired_in(&listed)?
944
            .into_keys()
945
            .filter(|k| k.starts_with("loose/"))
946
            .collect())
947
    }
948
949
    pub fn pack_count(&self) -> usize {
950
        self.packs.borrow().len()
951
    }
952
953
    fn find_packed(&self, r: &ObjectRef) -> Option<(usize, PackEntry)> {
954
        self.packs
955
            .borrow()
956
            .iter()
957
            .enumerate()
958
            .find_map(|(i, p)| p.find(r).map(|e| (i, e)))
959
    }
960
961
    /// Loose objects, with their keys and ages; not those retired, whose
962
    /// objects a pack holds.
963
    fn loose_objects(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> {
964
        let retired = self.retired_loose()?;
965
        let mut out = self.loose_objects_all()?;
966
        out.retain(|(_, e)| !retired.contains(&e.key));
967
        Ok(out)
968
    }
969
970
    /// Every loose object, retired or not.
971
    fn loose_objects_all(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> {
972
        let mut out: Vec<(ObjectRef, strata::Entry)> = self
973
            .backend
974
            .list("loose/")
975
            .map_err(backend_err)?
976
            .into_iter()
977
            .filter_map(|e| loose_ref(&e.key).map(|r| (r, e)))
978
            .collect();
979
        out.sort_by_key(|(r, _)| *r);
980
        Ok(out)
981
    }
982
983
    pub fn loose_count(&self) -> io::Result<usize> {
984
        self.loose_objects()
985
            .map(|l| l.len())
986
            .map_err(|e| io::Error::other(e.to_string()))
987
    }
988
989
    /// Move every loose object into a new pack (spec §13.4). Loose objects
990
    /// are deleted only after the pack and its index are durable.
991
    pub fn pack_loose(&self) -> Result<PackReport> {
992
        let loose = self.loose_objects()?;
993
        self.write_pack(
994
            &loose.iter().map(|(r, _)| *r).collect::<Vec<_>>(),
995
            &loose.iter().map(|(_, e)| e.key.clone()).collect::<Vec<_>>(),
996
            &[],
997
            ZSTD_LEVEL,
998
            &mut silent(),
999
        )
1000
    }
1001
1002
    /// Fold every pack and loose object into one pack compressed at
1003
    /// `level` (spec §13.4): what an explicit `oxy gc` does, as `git gc`
1004
    /// repacks. Returns the report and how many packs were folded in.
1005
    pub fn repack_all(&self, level: i32) -> Result<(PackReport, usize)> {
1006
        self.repack_all_with(level, &mut silent())
1007
    }
1008
1009
    /// [`Store::repack_all`], reporting progress.
1010
    pub fn repack_all_with(&self, level: i32, progress: Progress) -> Result<(PackReport, usize)> {
1011
        let loose = self.loose_objects()?;
1012
        let (mut refs, keys, trained): (Vec<ObjectRef>, Vec<String>, bool) = {
1013
            let packs = self.packs.borrow();
1014
            (
1015
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
1016
                packs.iter().map(|p| p.key.clone()).collect(),
1017
                packs.first().is_some_and(|p| p.dict.is_some()),
1018
            )
1019
        };
1020
        // Nothing loose and one pack already made by a full repack (only
1021
        // such a pack has a trained dictionary): rewriting it would give
1022
        // the same pack. A hub's nightly gc rewrote andymccall/oxygen's
1023
        // 48 000 objects (37 s under its write lock) every time.
1024
        if loose.is_empty() && keys.len() <= 1 && (level == ZSTD_LEVEL || trained) {
1025
            return Ok((PackReport::default(), 0));
1026
        }
1027
        refs.extend(loose.iter().map(|(r, _)| *r));
1028
        let loose_keys: Vec<String> = loose.into_iter().map(|(_, e)| e.key).collect();
1029
        let n = keys.len();
1030
        let report = self.write_pack(&refs, &loose_keys, &keys, level, progress)?;
1031
        Ok((report, n))
1032
    }
1033
1034
    /// Fold every pack into one when there are more than `MAX_PACKS`
1035
    /// (spec §13.4). Returns how many packs were merged.
1036
    pub fn merge_packs(&self) -> Result<usize> {
1037
        let (refs, keys): (Vec<ObjectRef>, Vec<String>) = {
1038
            let packs = self.packs.borrow();
1039
            if packs.len() <= MAX_PACKS {
1040
                return Ok(0);
1041
            }
1042
            (
1043
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
1044
                packs.iter().map(|p| p.key.clone()).collect(),
1045
            )
1046
        };
1047
        let n = keys.len();
1048
        self.write_pack(&refs, &[], &keys, ZSTD_LEVEL, &mut silent())?;
1049
        Ok(n)
1050
    }
1051
1052
    /// Delete every object not in `keep` (spec §21). Loose objects and packs
1053
    /// younger than `grace` are left alone, since a concurrent writer may not
1054
    /// have recorded its references yet. Packs holding unreachable objects
1055
    /// are rewritten without them. Signatures, seal and receipt entries of
1056
    /// deleted objects go too.
1057
    pub fn prune(
1058
        &self,
1059
        keep: &std::collections::HashSet<ObjectRef>,
1060
        grace: std::time::Duration,
1061
    ) -> Result<PruneReport> {
1062
        self.prune_with(keep, grace, &mut silent())
1063
    }
1064
1065
    /// [`Store::prune`], reporting progress.
1066
    pub fn prune_with(
1067
        &self,
1068
        keep: &std::collections::HashSet<ObjectRef>,
1069
        grace: std::time::Duration,
1070
        progress: Progress,
1071
    ) -> Result<PruneReport> {
1072
        progress("pruning loose objects");
1073
        let now = now_ms();
1074
        let grace_ms = grace.as_millis() as u64;
1075
        // An object whose age the store does not know is old enough.
1076
        let old_enough =
1077
            |modified: Option<u64>| modified.is_none_or(|t| now.saturating_sub(t) >= grace_ms);
1078
        let mut report = PruneReport::default();
1079
        let mut gone: Vec<String> = Vec::new();
1080
        // Loose objects. A retired one (its object packed) whose object is
1081
        // unreachable goes with it, uncounted: the pack's copy is counted.
1082
        let retired = self.retired_loose()?;
1083
        let mut unretired: Vec<String> = Vec::new();
1084
        for (r, entry) in self.loose_objects_all()? {
1085
            if keep.contains(&r) {
1086
                continue;
1087
            }
1088
            if retired.contains(&entry.key) {
1089
                if old_enough(entry.modified_ms) {
1090
                    delete(self.backend.as_ref(), &entry.key)?;
1091
                    unretired.push(entry.key);
1092
                }
1093
                continue;
1094
            }
1095
            if !old_enough(entry.modified_ms) {
1096
                report.deferred += 1;
1097
                continue;
1098
            }
1099
            report.bytes += entry.size;
1100
            report.objects += 1;
1101
            delete(self.backend.as_ref(), &entry.key)?;
1102
            self.remove_sidecars(&r, &mut gone)?;
1103
        }
1104
        if !unretired.is_empty() {
1105
            self.update_retired(|m| {
1106
                for k in &unretired {
1107
                    m.remove(k);
1108
                }
1109
            })?;
1110
        }
1111
        // Packs: rewrite those that hold unreachable objects.
1112
        let mut keep_refs: Vec<ObjectRef> = Vec::new();
1113
        let mut old_keys: Vec<String> = Vec::new();
1114
        let mut rewrite = false;
1115
        progress("pruning packs");
1116
        for p in self.packs.borrow().iter() {
1117
            let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count();
1118
            if unreachable == 0 {
1119
                continue;
1120
            }
1121
            let modified = self
1122
                .backend
1123
                .stat(&p.key)
1124
                .map_err(backend_err)?
1125
                .and_then(|e| e.modified_ms);
1126
            if !old_enough(modified) {
1127
                report.deferred += unreachable;
1128
                continue;
1129
            }
1130
            rewrite = true;
1131
            old_keys.push(p.key.clone());
1132
            for (r, e) in p.refs.iter().zip(p.entries.iter()) {
1133
                if keep.contains(r) {
1134
                    keep_refs.push(*r);
1135
                } else {
1136
                    report.objects += 1;
1137
                    report.bytes += e.stored_len as u64;
1138
                }
1139
            }
1140
        }
1141
        if rewrite {
1142
            let dropped: Vec<ObjectRef> = {
1143
                let packs = self.packs.borrow();
1144
                packs
1145
                    .iter()
1146
                    .filter(|p| old_keys.contains(&p.key))
1147
                    .flat_map(|p| p.refs.iter().copied())
1148
                    .filter(|r| !keep.contains(r))
1149
                    .collect()
1150
            };
1151
            self.write_pack(&keep_refs, &[], &old_keys, ZSTD_LEVEL, progress)?;
1152
            for r in dropped {
1153
                self.remove_sidecars(&r, &mut gone)?;
1154
            }
1155
        }
1156
        self.journal_forget(&gone)?;
1157
        Ok(report)
1158
    }
1159
1160
    /// Everything kept beside an object: its signatures and the seal and
1161
    /// receipt entries under it, in both layouts.
1162
    /// The entry keys deleted are added to `gone`, for the journal.
1163
    fn remove_sidecars(&self, r: &ObjectRef, gone: &mut Vec<String>) -> Result<()> {
1164
        for index in ["sigs", "seals", "receipts"] {
1165
            delete(self.backend.as_ref(), &legacy_key(index, r))?;
1166
        }
1167
        for index in ["sig", "seal", "receipt"] {
1168
            for e in self
1169
                .backend
1170
                .list(&fanout_prefix(index, r))
1171
                .map_err(backend_err)?
1172
            {
1173
                delete(self.backend.as_ref(), &e.key)?;
1174
                gone.push(e.key);
1175
            }
1176
        }
1177
        Ok(())
1178
    }
1179
1180
    /// Take entry keys a collection deleted out of the journal, so the
1181
    /// next complete fold leaves them out (spec §13.7).
1182
    fn journal_forget(&self, gone: &[String]) -> Result<()> {
1183
        if self.local || gone.is_empty() {
1184
            return Ok(());
1185
        }
1186
        if let Some(set) = self.entries.borrow_mut().as_mut() {
1187
            for k in gone {
1188
                set.remove(k);
1189
            }
1190
        }
1191
        self.update_record(JOURNAL_KEY, "the journal of entries", |m| {
1192
            for k in gone {
1193
                m.remove(k);
1194
            }
1195
        })
1196
    }
1197
1198
    /// Put a complete, synced temporary pack file and its index into the
1199
    /// store; `index` must be sorted by ObjectRef. The pack goes first: an
1200
    /// index never names a pack that is not there.
1201
    fn finish_pack_files(
1202
        &self,
1203
        id: &str,
1204
        tmp_pack: &Path,
1205
        index: &[(ObjectRef, PackEntry)],
1206
    ) -> Result<()> {
1207
        let mut idx = Vec::with_capacity(12 + index.len() * IDX_ENTRY);
1208
        idx.extend_from_slice(IDX_MAGIC);
1209
        idx.extend_from_slice(&(index.len() as u32).to_le_bytes());
1210
        for (r, e) in index {
1211
            idx.extend_from_slice(&r.0);
1212
            idx.extend_from_slice(&e.offset.to_le_bytes());
1213
            idx.extend_from_slice(&e.stored_len.to_le_bytes());
1214
            idx.extend_from_slice(&e.raw_len.to_le_bytes());
1215
            idx.extend_from_slice(&[type_index(e.ty), e.flags]);
1216
        }
1217
        let put = self
1218
            .backend
1219
            .put_new_file(&format!("pack-{id}.pack"), tmp_pack)
1220
            .map_err(backend_err);
1221
        // A store on another disk copied it; the one in this directory
1222
        // moved it and this finds nothing.
1223
        let _ = fs::remove_file(tmp_pack);
1224
        put?;
1225
        self.backend
1226
            .put_new(&format!("pack-{id}.idx"), &idx)
1227
            .map_err(backend_err)?;
1228
        self.reload_packs()
1229
    }
1230
1231
    /// Remove a pack with its index and dictionary: the index first, so
1232
    /// a store opened meanwhile does not load a pack that is going.
1233
    fn remove_pack(&self, pack_key: &str) -> Result<()> {
1234
        let stem = pack_key.trim_end_matches(".pack");
1235
        delete(self.backend.as_ref(), &format!("{stem}.idx"))?;
1236
        delete(self.backend.as_ref(), pack_key)?;
1237
        delete(self.backend.as_ref(), &format!("{stem}.dict"))
1238
    }
1239
1240
    /// Write `refs` into a new pack, then retire `loose_keys` and
1241
    /// `old_packs`: they stay in the store, unused by stores opened from
1242
    /// now on, until a collection [`RETIRED_FOR`] later deletes them, so a
1243
    /// reader that loaded its packs before goes on reading (oxygenhub#249).
1244
    fn write_pack(
1245
        &self,
1246
        refs: &[ObjectRef],
1247
        loose_keys: &[String],
1248
        old_packs: &[String],
1249
        level: i32,
1250
        progress: Progress,
1251
    ) -> Result<PackReport> {
1252
        if refs.is_empty() {
1253
            self.retire(old_packs)?;
1254
            self.reload_packs()?;
1255
            return Ok(PackReport::default());
1256
        }
1257
        let mut refs: Vec<ObjectRef> = refs.to_vec();
1258
        refs.sort();
1259
        refs.dedup();
1260
        let id = fresh_id();
1261
        let tmp_pack = self.scratch.join(format!(".tmp-{id}.pack"));
1262
        let mut report = PackReport::default();
1263
        let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len());
1264
        // A full repack of enough objects trains a dictionary for them.
1265
        let dict = if level == REPACK_ZSTD_LEVEL && refs.len() >= DICT_MIN_OBJECTS {
1266
            let _t = crate::timing::phase("pack: train a dictionary");
1267
            progress(&format!("training a dictionary on {} objects", refs.len()));
1268
            self.train_dictionary(&refs)
1269
        } else {
1270
            None
1271
        };
1272
        let encoder = dict
1273
            .as_ref()
1274
            .map(|d| zstd::dict::EncoderDictionary::copy(d, level));
1275
        let nocompress = self.nocompress_set()?;
1276
        {
1277
            let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?);
1278
            f.write_all(PACK_MAGIC)?;
1279
            let mut offset = PACK_MAGIC.len() as u64;
1280
            // Read a batch (the store is not shared between threads),
1281
            // compress it on every core, write it in order.
1282
            let mut batches = refs.chunks(1024);
1283
            let mut compressed_batch = Vec::new();
1284
            let total = refs.len();
1285
            let mut reported = usize::MAX;
1286
            loop {
1287
                if compressed_batch.is_empty() {
1288
                    let Some(batch) = batches.next() else { break };
1289
                    // Every tenth, and the first batch.
1290
                    let tenth = report.objects * 10 / total.max(1);
1291
                    if tenth != reported {
1292
                        reported = tenth;
1293
                        progress(&format!("packing {} of {total} objects", report.objects));
1294
                    }
1295
                    let mut read: Vec<(ObjectRef, ObjectType, Vec<u8>, bool)> =
1296
                        Vec::with_capacity(batch.len());
1297
                    for r in batch {
1298
                        let (ty, body) = self.get(r)?;
1299
                        read.push((*r, ty, body, nocompress.contains(r)));
1300
                    }
1301
                    use rayon::prelude::*;
1302
                    compressed_batch = read
1303
                        .into_par_iter()
1304
                        .map_init(
1305
                            || {
1306
                                encoder.as_ref().and_then(|e| {
1307
                                    zstd::bulk::Compressor::with_prepared_dictionary(e).ok()
1308
                                })
1309
                            },
1310
                            |with_dict, (r, ty, body, raw)| {
1311
                                let packed = if raw {
1312
                                    None
1313
                                } else {
1314
                                    // Small objects with the dictionary;
1315
                                    // a large one has context enough.
1316
                                    let dicted = with_dict
1317
                                        .as_mut()
1318
                                        .filter(|_| body.len() <= DICT_OBJECT_MAX)
1319
                                        .and_then(|c| c.compress(&body).ok())
1320
                                        .map(|c| (FLAG_ZSTD | FLAG_DICT, c));
1321
                                    dicted
1322
                                        .or_else(|| {
1323
                                            zstd::bulk::compress(&body, level)
1324
                                                .ok()
1325
                                                .map(|c| (FLAG_ZSTD, c))
1326
                                        })
1327
                                        .filter(|(_, c)| c.len() < body.len())
1328
                                };
1329
                                (r, ty, body, packed)
1330
                            },
1331
                        )
1332
                        .collect::<Vec<_>>();
1333
                    compressed_batch.reverse();
1334
                }
1335
                let Some((r, ty, body, packed)) = compressed_batch.pop() else {
1336
                    continue;
1337
                };
1338
                let r = &r;
1339
                let (flags, stored) = match packed {
1340
                    Some(c) => c,
1341
                    None => (0, body.clone()),
1342
                };
1343
                let header_len = 1 + 1 + 4 + 4 + 32;
1344
                f.write_all(&[type_index(ty), flags])?;
1345
                f.write_all(&(stored.len() as u32).to_le_bytes())?;
1346
                f.write_all(&(body.len() as u32).to_le_bytes())?;
1347
                f.write_all(&r.0)?;
1348
                f.write_all(&stored)?;
1349
                index.push((
1350
                    *r,
1351
                    PackEntry {
1352
                        offset: offset + header_len,
1353
                        stored_len: stored.len() as u32,
1354
                        raw_len: body.len() as u32,
1355
                        ty,
1356
                        flags,
1357
                    },
1358
                ));
1359
                offset += header_len + stored.len() as u64;
1360
                report.objects += 1;
1361
                report.raw_bytes += body.len() as u64;
1362
                report.packed_bytes += stored.len() as u64;
1363
            }
1364
            f.flush()?;
1365
            f.get_ref().sync_all()?;
1366
        }
1367
        // The dictionary before the pack that needs it becomes visible.
1368
        if let Some(d) = &dict {
1369
            self.backend
1370
                .put_new(&format!("pack-{id}.dict"), d)
1371
                .map_err(backend_err)?;
1372
        }
1373
        progress(&format!("writing the index of {} objects", index.len()));
1374
        self.finish_pack_files(&id, &tmp_pack, &index)?;
1375
        let mut retired = old_packs.to_vec();
1376
        retired.extend_from_slice(loose_keys);
1377
        self.retire(&retired)?;
1378
        self.reload_packs()?;
1379
        Ok(report)
1380
    }
1381
1382
    /// A dictionary for these objects, trained on a sample of them (at
1383
    /// most 10 000, 32 MB); `None` when training fails.
1384
    fn train_dictionary(&self, refs: &[ObjectRef]) -> Option<Vec<u8>> {
1385
        let step = (refs.len() / 10_000).max(1);
1386
        let mut samples: Vec<Vec<u8>> = Vec::new();
1387
        let mut total = 0usize;
1388
        for r in refs.iter().step_by(step) {
1389
            if total > 32 << 20 {
1390
                break;
1391
            }
1392
            let Ok((ty, body)) = self.get(r) else {
1393
                continue;
1394
            };
1395
            if matches!(
1396
                ty,
1397
                ObjectType::Chunk | ObjectType::Blob | ObjectType::Change
1398
            ) && !body.is_empty()
1399
                && body.len() < 1 << 20
1400
            {
1401
                total += body.len();
1402
                samples.push(body);
1403
            }
1404
        }
1405
        zstd::dict::from_samples(&samples, DICT_BYTES).ok()
1406
    }
1407
1408
    /// The scratch directory: for a store in a directory, that directory.
1409
    pub fn dir(&self) -> &Path {
1410
        &self.scratch
1411
    }
1412
1413
    /// Where a loose object is kept, for a store in a directory (tests
1414
    /// tamper with and remove objects there).
1415
    #[cfg(test)]
1416
    pub(crate) fn path_for(&self, r: &ObjectRef) -> PathBuf {
1417
        self.scratch.join(loose_key(r))
1418
    }
1419
1420
    /// Remove a loose object, whatever the store is on (tests make an
1421
    /// object go missing).
1422
    #[cfg(test)]
1423
    pub(crate) fn remove_loose_for_test(&self, r: &ObjectRef) -> Result<()> {
1424
        delete(self.backend.as_ref(), &loose_key(r))
1425
    }
1426
1427
    /// Whether the store holds `r`. A miss looks at the pack listing
1428
    /// again before it is believed, as [`Store::get`]'s does; a writer
1429
    /// asking about many objects it is about to store looks at most once
1430
    /// in [`HAS_RELIST_AFTER`] while nothing new turns up.
1431
    pub fn has(&self, r: &ObjectRef) -> bool {
1432
        if self.has_now(r) {
1433
            return true;
1434
        }
1435
        let recent = self
1436
            .has_relisted
1437
            .get()
1438
            .is_some_and(|t| t.elapsed() < HAS_RELIST_AFTER);
1439
        if recent {
1440
            return false;
1441
        }
1442
        match self.reload_if_changed() {
1443
            Ok(true) => {
1444
                self.has_relisted.set(None);
1445
                self.has_now(r)
1446
            }
1447
            _ => {
1448
                self.has_relisted.set(Some(std::time::Instant::now()));
1449
                false
1450
            }
1451
        }
1452
    }
1453
1454
    /// [`Store::has`] against the packs as loaded.
1455
    fn has_now(&self, r: &ObjectRef) -> bool {
1456
        self.find_packed(r).is_some()
1457
            || self
1458
                .pending
1459
                .borrow()
1460
                .as_ref()
1461
                .is_some_and(|p| p.lookup.contains_key(r))
1462
            || self.backend.has(&loose_key(r)).unwrap_or(false)
1463
    }
1464
1465
    /// Start a write batch: until the matching `finish`, `put` collects
1466
    /// objects and writes them out together, with one `fsync` for a large
1467
    /// batch (a pack) instead of one per object. Batches nest; only the
1468
    /// outermost does the work. Objects are readable while pending.
1469
    pub fn begin_batch(&self) -> Result<Batch<'_>> {
1470
        let depth = self.batch_depth.get();
1471
        if depth == 0 {
1472
            *self.pending.borrow_mut() = Some(Pending {
1473
                buffered: vec![],
1474
                buffered_bytes: 0,
1475
                lookup: Default::default(),
1476
                pack: None,
1477
                report: PackReport::default(),
1478
            });
1479
        }
1480
        self.batch_depth.set(depth + 1);
1481
        Ok(Batch {
1482
            store: self,
1483
            done: false,
1484
        })
1485
    }
1486
1487
    fn finish_batch(&self) -> Result<()> {
1488
        let depth = self.batch_depth.get();
1489
        if depth == 0 {
1490
            return Ok(());
1491
        }
1492
        self.batch_depth.set(depth - 1);
1493
        if depth > 1 {
1494
            return Ok(());
1495
        }
1496
        let Some(pending) = self.pending.borrow_mut().take() else {
1497
            return self.flush_journal();
1498
        };
1499
        self.write_pending(pending)?;
1500
        self.flush_journal()
1501
    }
1502
1503
    /// Make every object written so far durable, even inside an open
1504
    /// batch, which stays open. Call before a reference that names them
1505
    /// is moved (spec §13.6): the operation head, a bookmark pointer.
1506
    pub fn flush(&self) -> Result<()> {
1507
        let Some(pending) = self.pending.borrow_mut().take() else {
1508
            return self.flush_journal();
1509
        };
1510
        self.write_pending(pending)?;
1511
        self.flush_journal()?;
1512
        *self.pending.borrow_mut() = Some(Pending {
1513
            buffered: vec![],
1514
            buffered_bytes: 0,
1515
            lookup: Default::default(),
1516
            pack: None,
1517
            report: PackReport::default(),
1518
        });
1519
        Ok(())
1520
    }
1521
1522
    fn write_pending(&self, pending: Pending) -> Result<()> {
1523
        match pending.pack {
1524
            None => {
1525
                let objects: Vec<(ObjectType, &[u8])> = pending
1526
                    .buffered
1527
                    .iter()
1528
                    .map(|(ty, body)| (*ty, body.as_slice()))
1529
                    .collect();
1530
                self.put_loose_many(&objects)
1531
            }
1532
            Some(mut pack) => {
1533
                pack.writer.flush()?;
1534
                pack.writer.get_ref().sync_all()?;
1535
                drop(pack.writer);
1536
                pack.index.sort_by_key(|(r, _)| *r);
1537
                self.finish_pack_files(&pack.id, &pack.tmp_path, &pack.index)?;
1538
                Ok(())
1539
            }
1540
        }
1541
    }
1542
1543
    /// Whether a batch is open and holds `r`.
1544
    fn pending_get(&self, r: &ObjectRef) -> Option<Result<Object>> {
1545
        let mut guard = self.pending.borrow_mut();
1546
        let pending = guard.as_mut()?;
1547
        let entry = *pending.lookup.get(r)?;
1548
        Some(match entry {
1549
            PendingEntry::Buffered(i) => {
1550
                let (ty, body) = &pending.buffered[i];
1551
                Ok((*ty, body.clone()))
1552
            }
1553
            PendingEntry::Packed(e) => {
1554
                let Some(pack) = pending.pack.as_mut() else {
1555
                    return Some(Err(StoreError::NotFound(*r)));
1556
                };
1557
                // Read from the pack file being written, on this disk.
1558
                (|| {
1559
                    pack.writer.flush()?;
1560
                    use std::io::{Read, Seek, SeekFrom};
1561
                    let mut f = fs::File::open(&pack.tmp_path)?;
1562
                    f.seek(SeekFrom::Start(e.offset))?;
1563
                    let mut stored = vec![0u8; e.stored_len as usize];
1564
                    f.read_exact(&mut stored)?;
1565
                    decode_entry(r, e, stored, None)
1566
                })()
1567
            }
1568
        })
1569
    }
1570
1571
    /// Append one object to the pending pack file.
1572
    fn pending_pack_write(
1573
        &self,
1574
        pack: &mut PendingPack,
1575
        r: &ObjectRef,
1576
        ty: ObjectType,
1577
        body: &[u8],
1578
        report: &mut PackReport,
1579
    ) -> Result<PackEntry> {
1580
        let (flags, stored) = if self.is_nocompress(r) {
1581
            (0, body.to_vec())
1582
        } else {
1583
            let compressed = zstd::bulk::compress(body, ZSTD_LEVEL)
1584
                .map_err(|e| StoreError::Corrupt(*r, format!("zstd: {e}")))?;
1585
            if compressed.len() < body.len() {
1586
                (FLAG_ZSTD, compressed)
1587
            } else {
1588
                (0, body.to_vec())
1589
            }
1590
        };
1591
        write_pack_entry(pack, r, ty, body.len(), flags, &stored, report)
1592
    }
1593
1594
    fn put_pending(&self, r: ObjectRef, ty: ObjectType, body: &[u8]) -> Result<()> {
1595
        let mut guard = self.pending.borrow_mut();
1596
        let pending = guard.as_mut().expect("batch open");
1597
        if pending.lookup.contains_key(&r) {
1598
            return Ok(());
1599
        }
1600
        if pending.pack.is_none()
1601
            && pending.buffered.len() < SPILL_OBJECTS
1602
            && pending.buffered_bytes + body.len() <= SPILL_BYTES
1603
        {
1604
            pending.buffered.push((ty, body.to_vec()));
1605
            pending.buffered_bytes += body.len();
1606
            pending
1607
                .lookup
1608
                .insert(r, PendingEntry::Buffered(pending.buffered.len() - 1));
1609
            return Ok(());
1610
        }
1611
        self.ensure_pack(pending)?;
1612
        let mut report = std::mem::take(&mut pending.report);
1613
        let pack = pending.pack.as_mut().unwrap();
1614
        let e = self.pending_pack_write(pack, &r, ty, body, &mut report)?;
1615
        pending.report = report;
1616
        pending.lookup.insert(r, PendingEntry::Packed(e));
1617
        Ok(())
1618
    }
1619
1620
    /// Open the pending batch's pack, moving the objects buffered so far
1621
    /// into it (a spill).
1622
    fn ensure_pack(&self, pending: &mut Pending) -> Result<()> {
1623
        if pending.pack.is_none() {
1624
            // Spill: open the pack and move the buffered objects into it.
1625
            let id = fresh_id();
1626
            let tmp_path = self.scratch.join(format!(".tmp-{id}.pack"));
1627
            let mut writer = io::BufWriter::new(fs::File::create(&tmp_path)?);
1628
            writer.write_all(PACK_MAGIC)?;
1629
            let mut pack = PendingPack {
1630
                id,
1631
                tmp_path,
1632
                writer,
1633
                offset: PACK_MAGIC.len() as u64,
1634
                index: vec![],
1635
            };
1636
            let buffered = std::mem::take(&mut pending.buffered);
1637
            pending.buffered_bytes = 0;
1638
            let mut report = std::mem::take(&mut pending.report);
1639
            for (bty, bbody) in &buffered {
1640
                let br = hash_object(*bty, bbody);
1641
                let e = self.pending_pack_write(&mut pack, &br, *bty, bbody, &mut report)?;
1642
                pending.lookup.insert(br, PendingEntry::Packed(e));
1643
            }
1644
            pending.report = report;
1645
            pending.pack = Some(pack);
1646
        }
1647
        Ok(())
1648
    }
1649
1650
    /// Store many objects at once. Inside a large write batch they are
1651
    /// compressed on every core and appended in order; otherwise each is
1652
    /// stored as [`Store::put`] does.
1653
    pub fn put_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<Vec<ObjectRef>> {
1654
        use rayon::prelude::*;
1655
        let total: usize = objects.iter().map(|(_, b)| b.len()).sum();
1656
        if self.pending.borrow().is_none() || total < SPILL_BYTES {
1657
            return objects.iter().map(|(ty, b)| self.put(*ty, b)).collect();
1658
        }
1659
        for (ty, b) in objects {
1660
            if b.len() > MAX_OBJECT_LEN as usize {
1661
                return Err(StoreError::TooLarge {
1662
                    kind: ty.name(),
1663
                    size: b.len(),
1664
                    limit: MAX_OBJECT_LEN,
1665
                });
1666
            }
1667
        }
1668
        let refs: Vec<ObjectRef> = objects
1669
            .par_iter()
1670
            .map(|(ty, b)| hash_object(*ty, b))
1671
            .collect();
1672
        // What is new: not stored already, not earlier in this call.
1673
        let mut seen: std::collections::HashSet<ObjectRef> = std::collections::HashSet::new();
1674
        let mut todo: Vec<usize> = Vec::new();
1675
        for (i, r) in refs.iter().enumerate() {
1676
            let known = self
1677
                .pending
1678
                .borrow()
1679
                .as_ref()
1680
                .is_some_and(|p| p.lookup.contains_key(r));
1681
            if seen.insert(*r)
1682
                && !known
1683
                && self.find_packed(r).is_none()
1684
                && !self.backend.has(&loose_key(r)).unwrap_or(false)
1685
            {
1686
                todo.push(i);
1687
            }
1688
        }
1689
        let raw = self.nocompress_set()?;
1690
        let packed: Vec<(u8, Vec<u8>)> = todo
1691
            .par_iter()
1692
            .map(|i| {
1693
                let body = objects[*i].1;
1694
                if raw.contains(&refs[*i]) {
1695
                    return (0, body.to_vec());
1696
                }
1697
                match zstd::bulk::compress(body, ZSTD_LEVEL) {
1698
                    Ok(c) if c.len() < body.len() => (FLAG_ZSTD, c),
1699
                    _ => (0, body.to_vec()),
1700
                }
1701
            })
1702
            .collect();
1703
        let mut guard = self.pending.borrow_mut();
1704
        let pending = guard.as_mut().expect("batch open");
1705
        self.ensure_pack(pending)?;
1706
        let mut report = std::mem::take(&mut pending.report);
1707
        let pack = pending.pack.as_mut().unwrap();
1708
        for (i, (flags, stored)) in todo.iter().zip(packed) {
1709
            let (ty, body) = objects[*i];
1710
            let e = write_pack_entry(pack, &refs[*i], ty, body.len(), flags, &stored, &mut report)?;
1711
            pending.lookup.insert(refs[*i], PendingEntry::Packed(e));
1712
        }
1713
        pending.report = report;
1714
        Ok(refs)
1715
    }
1716
1717
    /// Store `body` as an object of type `ty`; idempotent.
1718
    pub fn put(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
1719
        // Refuse what the store could not read back (spec §13.2), before
1720
        // anything is written: a pack must never hold an entry its index
1721
        // would reject.
1722
        if body.len() > MAX_OBJECT_LEN as usize {
1723
            return Err(StoreError::TooLarge {
1724
                kind: ty.name(),
1725
                size: body.len(),
1726
                limit: MAX_OBJECT_LEN,
1727
            });
1728
        }
1729
        let r = hash_object(ty, body);
1730
        if self.find_packed(&r).is_some() || self.backend.has(&loose_key(&r)).unwrap_or(false) {
1731
            return Ok(r);
1732
        }
1733
        if self.pending.borrow().is_some() {
1734
            self.put_pending(r, ty, body)?;
1735
            return Ok(r);
1736
        }
1737
        self.put_loose(ty, body)?;
1738
        Ok(r)
1739
    }
1740
1741
    /// Write one loose object, durable when this returns.
1742
    fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
1743
        let r = hash_object(ty, body);
1744
        let mut bytes = header(ty, body.len());
1745
        bytes.extend_from_slice(body);
1746
        crate::durable::step("write loose object")?;
1747
        put_new_or_existing(self.backend.as_ref(), &loose_key(&r), &bytes)?;
1748
        Ok(r)
1749
    }
1750
1751
    /// Write several loose objects with as few waits on the disk as the
1752
    /// guarantees allow (a directory store writes them all, then syncs
1753
    /// each directory once). Durable when this returns, as `put_loose`
1754
    /// one at a time would be (spec §13.6).
1755
    fn put_loose_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<()> {
1756
        if objects.len() <= 1 {
1757
            for (ty, body) in objects {
1758
                self.put_loose(*ty, body)?;
1759
            }
1760
            return Ok(());
1761
        }
1762
        let mut seen = std::collections::HashSet::new();
1763
        let items: Vec<(String, Vec<u8>)> = objects
1764
            .iter()
1765
            .filter_map(|(ty, body)| {
1766
                let r = hash_object(*ty, body);
1767
                seen.insert(r).then(|| {
1768
                    let mut bytes = header(*ty, body.len());
1769
                    bytes.extend_from_slice(body);
1770
                    (loose_key(&r), bytes)
1771
                })
1772
            })
1773
            .collect();
1774
        self.backend.put_many_new(&items).map_err(backend_err)
1775
    }
1776
1777
    /// Read and verify an object, packed, loose, or pending in a batch.
1778
    /// The packs' indexes are in memory, so they are asked first: asking
1779
    /// for a loose copy of a packed object costs a request that finds
1780
    /// nothing on a store in a bucket, every time (oxygenhub#228).
1781
    ///
1782
    /// A miss is not believed at once (oxygenhub#249): when the pack it
1783
    /// was in is gone (deleted after a repack retired it) the packs are
1784
    /// loaded again, and when the object is nowhere the pack listing is
1785
    /// looked at, and the packs loaded again if it changed (another
1786
    /// writer packed it, or a repack moved it); then it is read once more.
1787
    pub fn get(&self, r: &ObjectRef) -> Result<Object> {
1788
        match self.get_now(r) {
1789
            Err(e) if is_miss(&e) => {
1790
                // Read from a pack this store loaded, which is gone.
1791
                let packed_miss =
1792
                    !matches!(e, StoreError::NotFound(_)) || self.find_packed(r).is_some();
1793
                let changed = if packed_miss {
1794
                    self.reload_packs()?;
1795
                    true
1796
                } else {
1797
                    self.reload_if_changed()?
1798
                };
1799
                if changed {
1800
                    self.get_now(r)
1801
                } else {
1802
                    Err(e)
1803
                }
1804
            }
1805
            other => other,
1806
        }
1807
    }
1808
1809
    /// [`Store::get`] against the packs as loaded.
1810
    fn get_now(&self, r: &ObjectRef) -> Result<Object> {
1811
        if let Some((i, e)) = self.find_packed(r) {
1812
            let pack = self.packs.borrow()[i].clone();
1813
            return pack.fetch(self.backend.as_ref(), r, e);
1814
        }
1815
        match get_optional(self.backend.as_ref(), &loose_key(r))? {
1816
            Some(bytes) => parse_loose(&bytes, r),
1817
            None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))),
1818
        }
1819
    }
1820
1821
    /// Many objects at once, read and verified in parallel: packed ones
1822
    /// from their packs, the others loose (as [`Store::get`], packs first),
1823
    /// then what is pending in a batch. Results are in the order of `refs`.
1824
    ///
1825
    /// Misses are looked for again as [`Store::get`]'s are, together.
1826
    pub fn get_many(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
1827
        let mut out = self.get_many_now(refs);
1828
        let missed: Vec<usize> = out
1829
            .iter()
1830
            .enumerate()
1831
            .filter(|(_, g)| g.as_ref().is_err_and(is_miss))
1832
            .map(|(i, _)| i)
1833
            .collect();
1834
        if missed.is_empty() {
1835
            return out;
1836
        }
1837
        let packed_miss = missed.iter().any(|i| {
1838
            !matches!(out[*i], Err(StoreError::NotFound(_)))
1839
                || self.find_packed(&refs[*i]).is_some()
1840
        });
1841
        let changed = if packed_miss {
1842
            self.reload_packs().map(|_| true)
1843
        } else {
1844
            self.reload_if_changed()
1845
        };
1846
        match changed {
1847
            Ok(true) => {
1848
                let again: Vec<ObjectRef> = missed.iter().map(|i| refs[*i]).collect();
1849
                for (i, g) in missed.into_iter().zip(self.get_many_now(&again)) {
1850
                    out[i] = g;
1851
                }
1852
            }
1853
            Ok(false) => {}
1854
            Err(e) => {
1855
                let why = e.to_string();
1856
                for i in missed {
1857
                    out[i] = Err(StoreError::Io(io::Error::other(why.clone())));
1858
                }
1859
            }
1860
        }
1861
        out
1862
    }
1863
1864
    /// [`Store::get_many`] against the packs as loaded.
1865
    fn get_many_now(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
1866
        use rayon::prelude::*;
1867
        let backend = self.backend.as_ref();
1868
        let packs = self.packs.borrow();
1869
        let place: Vec<Option<(usize, PackEntry)>> = refs
1870
            .iter()
1871
            .map(|r| {
1872
                packs
1873
                    .iter()
1874
                    .enumerate()
1875
                    .find_map(|(p, pack)| pack.find(r).map(|e| (p, e)))
1876
            })
1877
            .collect();
1878
        let indexes: &[Arc<PackIndex>] = &packs;
1879
        let found: Vec<Option<Result<Object>>> = refs
1880
            .par_iter()
1881
            .zip(place.par_iter())
1882
            .map(|(r, at)| match at {
1883
                Some((p, e)) => Some(indexes[*p].fetch(backend, r, *e)),
1884
                None => match get_optional(backend, &loose_key(r)) {
1885
                    Ok(Some(bytes)) => Some(parse_loose(&bytes, r)),
1886
                    Ok(None) => None,
1887
                    Err(e) => Some(Err(e)),
1888
                },
1889
            })
1890
            .collect();
1891
        drop(packs);
1892
        found
1893
            .into_iter()
1894
            .zip(refs)
1895
            .map(|(f, r)| {
1896
                f.unwrap_or_else(|| self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))))
1897
            })
1898
            .collect()
1899
    }
1900
1901
    pub fn get_typed(&self, r: &ObjectRef, want: ObjectType) -> Result<Vec<u8>> {
1902
        let (ty, body) = self.get(r)?;
1903
        if ty != want {
1904
            return Err(StoreError::WrongType(*r, ty.name(), want.name()));
1905
        }
1906
        Ok(body)
1907
    }
1908
1909
    /// Store a change and the text blobs (and their chunks) its large
1910
    /// inserts are stored as (spec §6.1).
1911
    pub fn put_change(&self, c: &Change) -> Result<ObjectRef> {
1912
        let encoding = crate::timing::phase("store: encode the change and its texts");
1913
        let bytes = c.encode_streaming(|batch| self.put_texts(&batch))?;
1914
        drop(encoding);
1915
        let _t = crate::timing::phase("store: put the encoding");
1916
        self.put(ObjectType::Change, &bytes)
1917
    }
1918
1919
    /// Store text blobs and their chunks: every chunk at once (compressed
1920
    /// on every core in a large batch), then the blobs.
1921
    pub fn put_texts(&self, texts: &[crate::objects::StoredText]) -> Result<()> {
1922
        let mut chunks: Vec<(ObjectType, &[u8])> = Vec::new();
1923
        for t in texts {
1924
            if !self.has(&t.blob.hash()) {
1925
                chunks.extend(
1926
                    crate::chunk::chunks(&t.content)
1927
                        .into_iter()
1928
                        .map(|c| (ObjectType::Chunk, c)),
1929
                );
1930
            }
1931
        }
1932
        self.put_many(&chunks)?;
1933
        for t in texts {
1934
            self.put(ObjectType::Blob, &t.blob.encode())?;
1935
        }
1936
        Ok(())
1937
    }
1938
1939
    pub fn get_change(&self, r: &ObjectRef) -> Result<Change> {
1940
        Ok(self.get_change_parts(r)?.0)
1941
    }
1942
1943
    /// A change and the objects its stored inserts are made of: each text
1944
    /// blob and its chunks, which travel and are kept with it.
1945
    pub fn get_change_parts(&self, r: &ObjectRef) -> Result<(Change, Vec<ObjectRef>)> {
1946
        let bytes = self.get_typed(r, ObjectType::Change)?;
1947
        let mut parts = Vec::new();
1948
        let mut failed = None;
1949
        let decoded = Change::decode_with(&bytes, &mut |text, count, form| match self
1950
            .read_text(text)
1951
        {
1952
            Ok((blob, content)) => {
1953
                parts.push(*text);
1954
                parts.extend(blob.chunks);
1955
                crate::objects::text_lines(&content, count, form)
1956
            }
1957
            Err(e) => {
1958
                let why = e.to_string();
1959
                failed = Some(e);
1960
                Err(ObjectError::Schema(format!(
1961
                    "text blob {}: {why}",
1962
                    text.short(12)
1963
                )))
1964
            }
1965
        });
1966
        match (decoded, failed) {
1967
            (Ok(c), _) => Ok((c, parts)),
1968
            (Err(_), Some(StoreError::NotFound(t))) => Err(StoreError::Corrupt(
1969
                *r,
1970
                format!("its text blob {} is missing", t.short(12)),
1971
            )),
1972
            (Err(e), _) => Err(StoreError::Corrupt(*r, e.to_string())),
1973
        }
1974
    }
1975
1976
    /// A change with every insert left without its lines (empty
1977
    /// placeholders), and where each one's lines are, in order: for
1978
    /// history-only states, which read text only for the graphs they build
1979
    /// (see [`Change::decode_skeleton`]).
1980
    pub fn get_change_lazy(&self, r: &ObjectRef) -> Result<(Change, Vec<crate::objects::TextRef>)> {
1981
        let bytes = self.get_typed(r, ObjectType::Change)?;
1982
        Change::decode_skeleton(&bytes, *r).map_err(|e| StoreError::Corrupt(*r, e.to_string()))
1983
    }
1984
1985
    /// A change's metadata (id, dependencies, author, clock, message, kind)
1986
    /// without its contents: read from its skeleton, so neither its text
1987
    /// blobs nor its inline lines are read (spec §11).
1988
    ///
1989
    /// Served from the `metas` cache when this store read it before (a
1990
    /// change object can be as large as the tree it adds), as long as the
1991
    /// store still has the change.
1992
    pub fn get_change_meta(&self, r: &ObjectRef) -> Result<crate::state::ChangeMeta> {
1993
        if self.metas.borrow().is_none() {
1994
            *self.metas.borrow_mut() = Some(self.read_metas_cache());
1995
        }
1996
        let cached = self.metas.borrow().as_ref().and_then(|m| m.get(r).cloned());
1997
        if let Some(meta) = cached {
1998
            if self.has(r) {
1999
                #[cfg(test)]
2000
                assert_eq!(
2001
                    meta,
2002
                    Change::decode_meta(&self.get_typed(r, ObjectType::Change)?).unwrap(),
2003
                    "the metadata cache holds a change's metadata"
2004
                );
2005
                return Ok(meta);
2006
            }
2007
        }
2008
        let bytes = self.get_typed(r, ObjectType::Change)?;
2009
        let meta =
2010
            Change::decode_meta(&bytes).map_err(|e| StoreError::Corrupt(*r, e.to_string()))?;
2011
        self.remember_meta(r, &meta);
2012
        Ok(meta)
2013
    }
2014
2015
    /// The metadata cache: records of a change hash, its metadata in the
2016
    /// flat layout, and the first 8 bytes of BLAKE3 over both, each
2017
    /// appended in one write. Derived only: a record that does not check
2018
    /// ends the reading (a torn write), and the cache can be deleted. It
2019
    /// lives in the scratch directory, on this disk.
2020
    fn read_metas_cache(&self) -> crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta> {
2021
        let mut out = crate::fasthash::FastMap::default();
2022
        let Ok(bytes) = fs::read(self.scratch.join("metas")) else {
2023
            return out;
2024
        };
2025
        let mut r = crate::flat::Reader::new(&bytes);
2026
        let mut good = 0;
2027
        while !r.at_end() {
2028
            let mut record = || -> std::result::Result<_, ObjectError> {
2029
                let h = r.object()?;
2030
                let body = r.bytes()?;
2031
                let check = r.fixed::<8>()?;
2032
                Ok((h, body, check))
2033
            };
2034
            let Ok((h, body, check)) = record() else {
2035
                break;
2036
            };
2037
            let mut sum = blake3::Hasher::new();
2038
            sum.update(&h.0).update(body);
2039
            if sum.finalize().as_bytes()[..8] != check {
2040
                break;
2041
            }
2042
            match crate::state::ChangeMeta::read_flat(&mut crate::flat::Reader::new(body)) {
2043
                Ok(m) => out.insert(h, m),
2044
                Err(_) => break,
2045
            };
2046
            good += 32 + 4 + body.len() + 8;
2047
        }
2048
        // Cut a bad tail, so records appended later are read.
2049
        if good < bytes.len() {
2050
            let _ = fs::OpenOptions::new()
2051
                .write(true)
2052
                .open(self.scratch.join("metas"))
2053
                .and_then(|f| f.set_len(good as u64));
2054
        }
2055
        out
2056
    }
2057
2058
    /// Add a change's metadata to the cache; quietly nothing when it
2059
    /// cannot be written (it is only a cache).
2060
    fn remember_meta(&self, r: &ObjectRef, meta: &crate::state::ChangeMeta) {
2061
        let mut body = crate::flat::Writer::default();
2062
        meta.write_flat(&mut body);
2063
        let mut w = crate::flat::Writer::default();
2064
        w.object(r);
2065
        w.bytes(&body.0);
2066
        let mut sum = blake3::Hasher::new();
2067
        sum.update(&r.0).update(&body.0);
2068
        w.fixed(&sum.finalize().as_bytes()[..8]);
2069
        let _ = fs::OpenOptions::new()
2070
            .create(true)
2071
            .append(true)
2072
            .open(self.scratch.join("metas"))
2073
            .and_then(|mut f| f.write_all(&w.0));
2074
        if let Some(m) = self.metas.borrow_mut().as_mut() {
2075
            m.insert(*r, meta.clone());
2076
        }
2077
    }
2078
2079
    fn read_text(&self, r: &ObjectRef) -> Result<(Blob, Vec<u8>)> {
2080
        let blob = Blob::decode(&self.get_typed(r, ObjectType::Blob)?)?;
2081
        let mut content = Vec::with_capacity(blob.length as usize);
2082
        for c in &blob.chunks {
2083
            content.extend(self.get_typed(c, ObjectType::Chunk)?);
2084
        }
2085
        Ok((blob, content))
2086
    }
2087
2088
    pub fn put_chunk(&self, bytes: &[u8]) -> Result<ObjectRef> {
2089
        self.put(ObjectType::Chunk, bytes)
2090
    }
2091
2092
    /// Store a chunk that packs will keep uncompressed (`nocompress`, spec §13.3).
2093
    pub fn put_chunk_nocompress(&self, bytes: &[u8]) -> Result<ObjectRef> {
2094
        let r = hash_object(ObjectType::Chunk, bytes);
2095
        put_new_or_existing(
2096
            self.backend.as_ref(),
2097
            &format!("nocompress/{}", r.to_hex()),
2098
            b"",
2099
        )?;
2100
        self.put(ObjectType::Chunk, bytes)
2101
    }
2102
2103
    fn is_nocompress(&self, r: &ObjectRef) -> bool {
2104
        self.backend
2105
            .has(&format!("nocompress/{}", r.to_hex()))
2106
            .unwrap_or(false)
2107
    }
2108
2109
    /// Every chunk marked to be kept raw.
2110
    fn nocompress_set(&self) -> Result<std::collections::HashSet<ObjectRef>> {
2111
        Ok(self
2112
            .backend
2113
            .list("nocompress/")
2114
            .map_err(backend_err)?
2115
            .into_iter()
2116
            .filter_map(|e| ObjectRef::from_hex(e.key.trim_start_matches("nocompress/")))
2117
            .collect())
2118
    }
2119
2120
    pub fn put_state(&self, st: &StateObject) -> Result<ObjectRef> {
2121
        self.put(ObjectType::State, &st.encode())
2122
    }
2123
2124
    pub fn get_state(&self, r: &ObjectRef) -> Result<StateObject> {
2125
        Ok(StateObject::decode(&self.get_typed(r, ObjectType::State)?)?)
2126
    }
2127
2128
    pub fn put_attest(&self, a: &Attest) -> Result<ObjectRef> {
2129
        self.put(ObjectType::Attest, &a.encode())
2130
    }
2131
2132
    pub fn get_attest(&self, r: &ObjectRef) -> Result<Attest> {
2133
        Ok(Attest::decode(&self.get_typed(r, ObjectType::Attest)?)?)
2134
    }
2135
2136
    pub fn put_receipt(&self, r: &Receipt) -> Result<ObjectRef> {
2137
        self.put(ObjectType::Receipt, &r.encode())
2138
    }
2139
2140
    pub fn get_receipt(&self, r: &ObjectRef) -> Result<Receipt> {
2141
        Ok(Receipt::decode(&self.get_typed(r, ObjectType::Receipt)?)?)
2142
    }
2143
2144
    /// Record `value` under `key` in `index`: one write-once entry, so two
2145
    /// writers of a shared store never lose each other's.
2146
    fn index_add(&self, index: &str, key: &ObjectRef, value: &ObjectRef) -> Result<()> {
2147
        debug_assert!(ENTRY_INDEXES.iter().any(|(i, _)| *i == index));
2148
        let entry = format!("{}{}", fanout_prefix(index, key), value.to_hex());
2149
        put_new_or_existing(self.backend.as_ref(), &entry, b"")?;
2150
        self.journal_add(entry)
2151
    }
2152
2153
    /// The values recorded under `key` in `index`, with those an older
2154
    /// store kept as one list in `legacy`, sorted.
2155
    fn index_get(&self, index: &str, legacy: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
2156
        let mut values: Vec<ObjectRef> = self
2157
            .entry_keys(&fanout_prefix(index, key))?
2158
            .into_iter()
2159
            .filter_map(|k| k.rsplit('/').next().and_then(ObjectRef::from_hex))
2160
            .collect();
2161
        if !self.reads_legacy() {
2162
            // Older lists folded into entries (spec §13.7).
2163
        } else if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key(legacy, key))? {
2164
            values.extend(
2165
                String::from_utf8_lossy(&bytes)
2166
                    .lines()
2167
                    .filter_map(ObjectRef::from_hex),
2168
            );
2169
        }
2170
        values.sort();
2171
        values.dedup();
2172
        Ok(values)
2173
    }
2174
2175
    /// Record that seal attestation `attest` covers `change`.
2176
    pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> {
2177
        self.index_add("seal", change, attest)
2178
    }
2179
2180
    /// Seal attestations known to cover `change`.
2181
    pub fn seals_for(&self, change: &ObjectRef) -> Result<Vec<ObjectRef>> {
2182
        self.index_get("seal", "seals", change)
2183
    }
2184
2185
    /// Record receipt attestation `attest` for `state` (spec §14.6).
2186
    /// Keep a short-lived key's certificate (spec §14.7), with its issuer's
2187
    /// signature, indexed by the key it certifies.
2188
    pub fn put_certificate(&self, c: &crate::cert::SignedCertificate) -> Result<ObjectRef> {
2189
        let r = self.put(ObjectType::Certificate, &c.cert.encode())?;
2190
        self.put_signature(&r, &c.sig)?;
2191
        self.index_add("cert", &c.cert.key_id(), &r)?;
2192
        Ok(r)
2193
    }
2194
2195
    /// A kept certificate with its issuer's signature.
2196
    pub fn get_certificate(&self, r: &ObjectRef) -> Result<crate::cert::SignedCertificate> {
2197
        let cert = crate::cert::Certificate::decode(&self.get_typed(r, ObjectType::Certificate)?)?;
2198
        let sig = self
2199
            .signatures(r)?
2200
            .into_iter()
2201
            .find(|s| s.key == cert.issuer)
2202
            .ok_or(StoreError::NotFound(*r))?;
2203
        Ok(crate::cert::SignedCertificate { cert, sig })
2204
    }
2205
2206
    /// The certificate kept for short-lived key `key`, if any.
2207
    pub fn certificate_of(&self, key: &ObjectRef) -> Result<Option<ObjectRef>> {
2208
        Ok(self
2209
            .index_get("cert", "certs", key)?
2210
            .into_iter()
2211
            .find(|r| self.has(r)))
2212
    }
2213
2214
    /// Record `value` under `key` in index `index` (for repository code).
2215
    pub(crate) fn index_add_pub(
2216
        &self,
2217
        index: &str,
2218
        key: &ObjectRef,
2219
        value: &ObjectRef,
2220
    ) -> Result<()> {
2221
        self.index_add(index, key, value)
2222
    }
2223
2224
    /// The values recorded under `key` in index `index` (which has no
2225
    /// older one-list form: the legacy name only has to differ from it).
2226
    pub(crate) fn index_get_pub(&self, index: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
2227
        self.index_get(index, &format!("{index}-lists"), key)
2228
    }
2229
2230
    pub fn index_receipt(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
2231
        self.index_add("receipt", state, attest)
2232
    }
2233
2234
    /// Receipt attestations recorded for `state`.
2235
    pub fn receipts_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
2236
        self.index_get("receipt", "receipts", state)
2237
    }
2238
2239
    pub fn index_approval(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
2240
        self.index_add("approval", state, attest)
2241
    }
2242
2243
    /// Approval attestations recorded for `state` (OxygenFlow approval
2244
    /// gates, oxygenhub#235).
2245
    pub fn approvals_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
2246
        self.index_get("approval", "approvals", state)
2247
    }
2248
2249
    /// Every state with an approval, for garbage collection.
2250
    pub fn approval_states(&self) -> Result<Vec<ObjectRef>> {
2251
        self.indexed_states("approval", "approvals")
2252
    }
2253
2254
    /// Every state with a receipt, for garbage collection.
2255
    pub fn receipt_states(&self) -> Result<Vec<ObjectRef>> {
2256
        self.indexed_states("receipt", "receipts")
2257
    }
2258
2259
    /// The states an index (`one/<hh>/<rest>/<attest>` entries, and the
2260
    /// packed `many/<hh>/<rest>` lists) holds attestations for.
2261
    fn indexed_states(&self, one: &str, many: &str) -> Result<Vec<ObjectRef>> {
2262
        let mut out: Vec<ObjectRef> = Vec::new();
2263
        for key in self.entry_keys(&format!("{one}/"))? {
2264
            // receipt/<hh>/<rest>/<attest>
2265
            let mut parts = key.splitn(4, '/');
2266
            let (Some(_), Some(hh), Some(rest), Some(_)) =
2267
                (parts.next(), parts.next(), parts.next(), parts.next())
2268
            else {
2269
                continue;
2270
            };
2271
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
2272
        }
2273
        let legacy = if self.reads_legacy() {
2274
            self.backend
2275
                .list(&format!("{many}/"))
2276
                .map_err(backend_err)?
2277
        } else {
2278
            Vec::new()
2279
        };
2280
        for e in legacy {
2281
            let mut parts = e.key.splitn(3, '/');
2282
            let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next()) else {
2283
                continue;
2284
            };
2285
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
2286
        }
2287
        out.sort();
2288
        out.dedup();
2289
        Ok(out)
2290
    }
2291
2292
    pub fn put_identity(&self, id: &Identity) -> Result<ObjectRef> {
2293
        self.put(ObjectType::Identity, &id.encode())
2294
    }
2295
2296
    pub fn get_identity(&self, r: &ObjectRef) -> Result<Identity> {
2297
        Ok(Identity::decode(&self.get_typed(r, ObjectType::Identity)?)?)
2298
    }
2299
2300
    /// Signatures stored alongside object `r` (spec §5.4), one per signing
2301
    /// key, sorted by key. Empty if none.
2302
    pub fn signatures(&self, r: &ObjectRef) -> Result<Vec<Signature>> {
2303
        let mut sigs: Vec<Signature> = Vec::new();
2304
        for key in self.entry_keys(&fanout_prefix("sig", r))? {
2305
            // Gone since it was listed or journalled: a collection deleted
2306
            // it with its object, or a new signature by the same signer is
2307
            // replacing it.
2308
            let Some(bytes) = get_optional(self.backend.as_ref(), &key)? else {
2309
                continue;
2310
            };
2311
            sigs.extend(Signature::decode_list(&bytes)?);
2312
        }
2313
        // An older store's one list per object; a signer kept there and
2314
        // here counts once, as written here.
2315
        let legacy = if self.reads_legacy() {
2316
            get_optional(self.backend.as_ref(), &legacy_key("sigs", r))?
2317
        } else {
2318
            None
2319
        };
2320
        if let Some(bytes) = legacy {
2321
            for sig in Signature::decode_list(&bytes)? {
2322
                if !sigs.iter().any(|s| s.key == sig.key) {
2323
                    sigs.push(sig);
2324
                }
2325
            }
2326
        }
2327
        sigs.sort_by_key(|s| s.key);
2328
        sigs.dedup_by_key(|s| s.key);
2329
        Ok(sigs)
2330
    }
2331
2332
    /// Attach a signature to object `r`, replacing any earlier one by the
2333
    /// same key: one entry per signer, so the latest by a key is the one
2334
    /// kept, and two signers never touch each other's.
2335
    pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> {
2336
        let key = format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex());
2337
        let bytes = Signature::encode_list(std::slice::from_ref(sig));
2338
        if get_optional(self.backend.as_ref(), &key)?.as_deref() != Some(bytes.as_slice()) {
2339
            delete(self.backend.as_ref(), &key)?;
2340
            put_new_or_existing(self.backend.as_ref(), &key, &bytes)?;
2341
        }
2342
        // Journalled even when the key was there: an older writer's
2343
        // entry is not in the journal or any list.
2344
        self.journal_add(key)
2345
    }
2346
2347
    // ---- the journal and entry lists (spec §13.7, #116)
2348
2349
    /// Whether lookups still read the one-list keys of older stores: until
2350
    /// a complete fold has copied them into per-object entries and the
2351
    /// pack listing shows its marker (spec §13.7).
2352
    fn reads_legacy(&self) -> bool {
2353
        !self.legacy_folded.get()
2354
    }
2355
2356
    /// Copy every older one-list key's contents into per-object entries,
2357
    /// keeping an entry already there, then write the marker that tells
2358
    /// readers to stop reading those keys. The keys themselves are left
2359
    /// for older readers. A list that cannot be read is passed over, and
2360
    /// said.
2361
    fn fold_legacy(&self, progress: Progress) -> Result<()> {
2362
        let mut copied = 0usize;
2363
        for (index, legacy) in ENTRY_INDEXES {
2364
            for e in self
2365
                .backend
2366
                .list(&format!("{legacy}/"))
2367
                .map_err(backend_err)?
2368
            {
2369
                // <legacy>/<hh>/<rest>
2370
                let mut parts = e.key.splitn(3, '/');
2371
                let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next())
2372
                else {
2373
                    continue;
2374
                };
2375
                let Some(r) = ObjectRef::from_hex(&format!("{hh}{rest}")) else {
2376
                    continue;
2377
                };
2378
                let Some(bytes) = get_optional(self.backend.as_ref(), &e.key)? else {
2379
                    continue;
2380
                };
2381
                if index == "sig" {
2382
                    let Ok(sigs) = Signature::decode_list(&bytes) else {
2383
                        progress(&format!("{}: not a list of signatures, passed over", e.key));
2384
                        continue;
2385
                    };
2386
                    for sig in sigs {
2387
                        let key = format!("{}{}", fanout_prefix("sig", &r), sig.key.to_hex());
2388
                        let one = Signature::encode_list(std::slice::from_ref(&sig));
2389
                        put_new_or_existing(self.backend.as_ref(), &key, &one)?;
2390
                        copied += 1;
2391
                    }
2392
                } else {
2393
                    for v in String::from_utf8_lossy(&bytes)
2394
                        .lines()
2395
                        .filter_map(ObjectRef::from_hex)
2396
                    {
2397
                        let key = format!("{}{}", fanout_prefix(index, &r), v.to_hex());
2398
                        put_new_or_existing(self.backend.as_ref(), &key, b"")?;
2399
                        copied += 1;
2400
                    }
2401
                }
2402
            }
2403
        }
2404
        if copied > 0 {
2405
            progress(&format!("{copied} entries copied out of older lists"));
2406
        }
2407
        put_new_or_existing(
2408
            self.backend.as_ref(),
2409
            LEGACY_FOLDED_KEY,
2410
            format!("{}\n", now_ms()).as_bytes(),
2411
        )?;
2412
        self.legacy_folded.set(true);
2413
        Ok(())
2414
    }
2415
2416
    /// The entry keys under `prefix` (an index's `<index>/` or one
2417
    /// object's `<index>/<hh>/<rest>/`), sorted: on a folded store from
2418
    /// its lists and journal, with no request; otherwise by listing.
2419
    fn entry_keys(&self, prefix: &str) -> Result<Vec<String>> {
2420
        if let Some(keys) = self.with_entries(|set| {
2421
            set.range(prefix.to_string()..)
2422
                .take_while(|k| k.starts_with(prefix))
2423
                .cloned()
2424
                .collect::<Vec<String>>()
2425
        })? {
2426
            return Ok(keys);
2427
        }
2428
        Ok(self
2429
            .backend
2430
            .list(prefix)
2431
            .map_err(backend_err)?
2432
            .into_iter()
2433
            .map(|e| e.key)
2434
            .collect())
2435
    }
2436
2437
    /// `f` over a folded store's entry keys, read on first use; `None`
2438
    /// when the store is not folded (or is this disk's directory), and
2439
    /// lookups list.
2440
    fn with_entries<T>(
2441
        &self,
2442
        f: impl FnOnce(&std::collections::BTreeSet<String>) -> T,
2443
    ) -> Result<Option<T>> {
2444
        if self.local || !self.folded.get() {
2445
            return Ok(None);
2446
        }
2447
        if self.entries.borrow().is_none() {
2448
            let Some(set) = self.read_entries()? else {
2449
                return Ok(None);
2450
            };
2451
            *self.entries.borrow_mut() = Some(set);
2452
        }
2453
        Ok(self.entries.borrow().as_ref().map(f))
2454
    }
2455
2456
    /// Every live entry list's keys and every journal line's, folded or
2457
    /// not, with what this handle has queued. A list deleted since the
2458
    /// packs were listed (retired over an hour ago) has the packs listed
2459
    /// again; `None` when one is still missing then, or the store is no
2460
    /// longer folded: lookups list.
2461
    fn read_entries(&self) -> Result<Option<std::collections::BTreeSet<String>>> {
2462
        'attempt: for attempt in 0..2 {
2463
            if !self.folded.get() {
2464
                return Ok(None);
2465
            }
2466
            let mut set = std::collections::BTreeSet::new();
2467
            let lists = self.entry_lists.borrow().clone();
2468
            for key in &lists {
2469
                match get_optional(self.backend.as_ref(), key)? {
2470
                    Some(bytes) => set.extend(decode_entry_list(key, &bytes)?),
2471
                    None if attempt == 0 => {
2472
                        self.reload_packs()?;
2473
                        continue 'attempt;
2474
                    }
2475
                    None => return Ok(None),
2476
                }
2477
            }
2478
            set.extend(self.read_record(JOURNAL_KEY)?.0.into_keys());
2479
            set.extend(self.journal_queue.borrow().iter().cloned());
2480
            return Ok(Some(set));
2481
        }
2482
        Ok(None)
2483
    }
2484
2485
    /// Add an entry key this handle wrote to the journal: at once, or
2486
    /// when the open batch is flushed. One the handle already knows from
2487
    /// the lists or the journal is not added again. Nothing on this
2488
    /// disk's directory.
2489
    fn journal_add(&self, key: String) -> Result<()> {
2490
        if self.local {
2491
            return Ok(());
2492
        }
2493
        if let Some(set) = self.entries.borrow_mut().as_mut() {
2494
            if !set.insert(key.clone()) {
2495
                return Ok(());
2496
            }
2497
        }
2498
        if self.batch_depth.get() > 0 {
2499
            self.journal_queue.borrow_mut().push(key);
2500
            return Ok(());
2501
        }
2502
        self.journal_append(std::slice::from_ref(&key))
2503
    }
2504
2505
    /// Add what the open batch queued to the journal.
2506
    fn flush_journal(&self) -> Result<()> {
2507
        let queued = std::mem::take(&mut *self.journal_queue.borrow_mut());
2508
        self.journal_append(&queued)
2509
    }
2510
2511
    /// Add `keys` to the journal, unfolded, in one compare-and-swap
2512
    /// update, dropping lines folded over an hour ago; past
2513
    /// [`JOURNAL_FOLD_AT`] unfolded lines, fold them into a partial list.
2514
    fn journal_append(&self, keys: &[String]) -> Result<()> {
2515
        if keys.is_empty() {
2516
            return Ok(());
2517
        }
2518
        let now = now_ms();
2519
        let mut unfolded = 0;
2520
        self.update_record(JOURNAL_KEY, "the journal of entries", |m| {
2521
            expire_folded(m, now);
2522
            for k in keys {
2523
                m.entry(k.clone()).or_insert(0);
2524
            }
2525
            unfolded = m.values().filter(|t| **t == 0).count();
2526
        })?;
2527
        if unfolded > JOURNAL_FOLD_AT {
2528
            self.fold_journal()?;
2529
        }
2530
        Ok(())
2531
    }
2532
2533
    /// Fold the journal's unfolded lines into a partial entry list, and
2534
    /// mark them folded. Lines added meanwhile stay unfolded.
2535
    fn fold_journal(&self) -> Result<()> {
2536
        let (journal, _) = self.read_record(JOURNAL_KEY)?;
2537
        let keys: Vec<String> = journal
2538
            .into_iter()
2539
            .filter(|(_, t)| *t == 0)
2540
            .map(|(k, _)| k)
2541
            .collect();
2542
        if keys.is_empty() {
2543
            return Ok(());
2544
        }
2545
        self.put_entry_list(PART_SUFFIX, keys.iter())?;
2546
        self.mark_folded(&keys)
2547
    }
2548
2549
    /// Mark these journal lines folded now, if they are not yet; drop
2550
    /// those folded over an hour ago.
2551
    fn mark_folded(&self, keys: &[String]) -> Result<()> {
2552
        let now = now_ms();
2553
        self.update_record(JOURNAL_KEY, "the journal of entries", |m| {
2554
            expire_folded(m, now);
2555
            for k in keys {
2556
                if let Some(t) = m.get_mut(k) {
2557
                    if *t == 0 {
2558
                        *t = now;
2559
                    }
2560
                }
2561
            }
2562
        })
2563
    }
2564
2565
    /// Write a new entry list of `keys` (sorted), complete or partial by
2566
    /// `suffix`; its key.
2567
    fn put_entry_list<'a>(
2568
        &self,
2569
        suffix: &str,
2570
        keys: impl Iterator<Item = &'a String>,
2571
    ) -> Result<String> {
2572
        let key = format!("pack-{}{suffix}", fresh_id());
2573
        let bytes = encode_entry_list(keys)?;
2574
        self.backend.put_new(&key, &bytes).map_err(backend_err)?;
2575
        Ok(key)
2576
    }
2577
2578
    /// Fold every signature and index entry into one complete entry list
2579
    /// (spec §13.7), what a collection does last: first the older
2580
    /// one-list keys into per-object entries, once; then the journal's lines and
2581
    /// a listing of every entry prefix, written as a new list, every list
2582
    /// the packs' listing showed retired, and the journal lines read
2583
    /// marked folded. Nothing when the store is this disk's directory, or
2584
    /// when the one live list already holds exactly these entries and the
2585
    /// journal has nothing unfolded. Whether a list was written.
2586
    pub fn fold_entries(&self, progress: Progress) -> Result<bool> {
2587
        if self.local {
2588
            return Ok(false);
2589
        }
2590
        self.reload_packs()?;
2591
        if !self.legacy_folded.get() {
2592
            progress("folding older lists of signatures and index entries");
2593
            self.fold_legacy(progress)?;
2594
        }
2595
        let (journal, _) = self.read_record(JOURNAL_KEY)?;
2596
        progress("folding signatures and index entries");
2597
        let mut all: std::collections::BTreeSet<String> = journal.keys().cloned().collect();
2598
        for (index, _) in ENTRY_INDEXES {
2599
            for e in self
2600
                .backend
2601
                .list(&format!("{index}/"))
2602
                .map_err(backend_err)?
2603
            {
2604
                all.insert(e.key);
2605
            }
2606
        }
2607
        let lists = self.entry_lists.borrow().clone();
2608
        let unfolded: Vec<String> = journal
2609
            .iter()
2610
            .filter(|(_, t)| **t == 0)
2611
            .map(|(k, _)| k.clone())
2612
            .collect();
2613
        if unfolded.is_empty() && lists.len() == 1 && lists[0].ends_with(LIST_SUFFIX) {
2614
            if let Some(bytes) = get_optional(self.backend.as_ref(), &lists[0])? {
2615
                let held: std::collections::BTreeSet<String> =
2616
                    decode_entry_list(&lists[0], &bytes)?.into_iter().collect();
2617
                if held == all {
2618
                    return Ok(false);
2619
                }
2620
            }
2621
        }
2622
        self.put_entry_list(LIST_SUFFIX, all.iter())?;
2623
        self.retire(&lists)?;
2624
        if !unfolded.is_empty() {
2625
            self.mark_folded(&unfolded)?;
2626
        }
2627
        *self.entries.borrow_mut() = None;
2628
        self.reload_packs()?;
2629
        Ok(true)
2630
    }
2631
2632
    /// Finish packs a writer left unfinished: a batch writes its objects to
2633
    /// `.tmp-ID.pack` and only indexes it at the end, so a process that
2634
    /// stopped part way (an import killed after hours) leaves every object
2635
    /// in a pack nothing can read. Each such pack untouched for `min_age`
2636
    /// is read through: every whole entry checked against its reference,
2637
    /// a torn last one cut off, then indexed as a pack of the store.
2638
    /// Returns the packs and objects adopted.
2639
    pub fn adopt_unfinished_packs(
2640
        &self,
2641
        min_age: std::time::Duration,
2642
        progress: &dyn Fn(&str),
2643
    ) -> Result<(usize, usize)> {
2644
        use std::io::{BufReader, Read};
2645
        let mut packs = 0;
2646
        let mut objects = 0;
2647
        let Ok(dir) = fs::read_dir(&self.scratch) else {
2648
            return Ok((0, 0));
2649
        };
2650
        for entry in dir.flatten() {
2651
            let name = entry.file_name().to_string_lossy().to_string();
2652
            let Some(id) = name
2653
                .strip_prefix(".tmp-")
2654
                .and_then(|n| n.strip_suffix(".pack"))
2655
            else {
2656
                continue;
2657
            };
2658
            let path = entry.path();
2659
            let meta = fs::metadata(&path)?;
2660
            let age = meta
2661
                .modified()
2662
                .ok()
2663
                .and_then(|m| m.elapsed().ok())
2664
                .unwrap_or_default();
2665
            if age < min_age {
2666
                progress(&format!(
2667
                    "{name}: written to {}s ago, left alone",
2668
                    age.as_secs()
2669
                ));
2670
                continue;
2671
            }
2672
            let mut f = BufReader::with_capacity(1 << 20, fs::File::open(&path)?);
2673
            let mut magic = [0u8; 8];
2674
            if f.read_exact(&mut magic).is_err() || &magic != PACK_MAGIC {
2675
                progress(&format!("{name}: not a pack, left alone"));
2676
                continue;
2677
            }
2678
            let mut offset = PACK_MAGIC.len() as u64;
2679
            let mut index: Vec<(ObjectRef, PackEntry)> = Vec::new();
2680
            loop {
2681
                let mut head = [0u8; 42];
2682
                if f.read_exact(&mut head).is_err() {
2683
                    break;
2684
                }
2685
                let Some(ty) = ObjectType::ALL.get(head[0] as usize).copied() else {
2686
                    break;
2687
                };
2688
                let flags = head[1];
2689
                let stored_len = u32::from_le_bytes(head[2..6].try_into().unwrap());
2690
                let raw_len = u32::from_le_bytes(head[6..10].try_into().unwrap());
2691
                let r = ObjectRef(head[10..42].try_into().unwrap());
2692
                let mut stored = vec![0u8; stored_len as usize];
2693
                if flags & FLAG_DICT != 0 || f.read_exact(&mut stored).is_err() {
2694
                    break;
2695
                }
2696
                let e = PackEntry {
2697
                    offset: offset + 42,
2698
                    stored_len,
2699
                    raw_len,
2700
                    ty,
2701
                    flags,
2702
                };
2703
                if decode_entry(&r, e, stored, None).is_err() {
2704
                    break;
2705
                }
2706
                index.push((r, e));
2707
                offset += 42 + stored_len as u64;
2708
                if index.len().is_multiple_of(200_000) {
2709
                    progress(&format!(
2710
                        "{name}: {} objects checked, {} MiB",
2711
                        index.len(),
2712
                        offset >> 20
2713
                    ));
2714
                }
2715
            }
2716
            if index.is_empty() {
2717
                progress(&format!("{name}: no whole object, left alone"));
2718
                continue;
2719
            }
2720
            if offset < meta.len() {
2721
                progress(&format!(
2722
                    "{name}: {} torn bytes at the end cut off",
2723
                    meta.len() - offset
2724
                ));
2725
                fs::OpenOptions::new()
2726
                    .write(true)
2727
                    .open(&path)?
2728
                    .set_len(offset)?;
2729
            }
2730
            index.sort_by_key(|(r, _)| *r);
2731
            index.dedup_by_key(|(r, _)| *r);
2732
            objects += index.len();
2733
            packs += 1;
2734
            progress(&format!("{name}: {} objects indexed", index.len()));
2735
            self.finish_pack_files(id, &path, &index)?;
2736
        }
2737
        Ok((packs, objects))
2738
    }
2739
2740
    /// Every object reference in the store, loose and packed (slow; for gc and tests).
2741
    pub fn list(&self) -> io::Result<Vec<ObjectRef>> {
2742
        let mut out: Vec<ObjectRef> = self
2743
            .loose_objects()
2744
            .map_err(|e| io::Error::other(e.to_string()))?
2745
            .into_iter()
2746
            .map(|(r, _)| r)
2747
            .collect();
2748
        for p in self.packs.borrow().iter() {
2749
            out.extend(p.refs.iter().copied());
2750
        }
2751
        out.sort();
2752
        out.dedup();
2753
        Ok(out)
2754
    }
2755
}
2756
2757
/// Whether a read failed in a way a newer list of packs may cure: the
2758
/// object was not found, or its pack's dictionary is gone.
2759
fn is_miss(e: &StoreError) -> bool {
2760
    match e {
2761
        StoreError::NotFound(_) => true,
2762
        StoreError::Corrupt(_, why) => why == NO_DICT,
2763
        _ => false,
2764
    }
2765
}
2766
2767
/// A loose object's bytes, verified against `r`.
2768
fn parse_loose(data: &[u8], r: &ObjectRef) -> Result<Object> {
2769
    let nul = data
2770
        .iter()
2771
        .position(|&b| b == 0)
2772
        .ok_or_else(|| StoreError::Corrupt(*r, "no header".into()))?;
2773
    let head = std::str::from_utf8(&data[..nul])
2774
        .map_err(|_| StoreError::Corrupt(*r, "header not UTF-8".into()))?;
2775
    let mut parts = head.split(' ');
2776
    let (Some("oxy1"), Some(ty), Some(len), None) =
2777
        (parts.next(), parts.next(), parts.next(), parts.next())
2778
    else {
2779
        return Err(StoreError::Corrupt(*r, format!("bad header {head:?}")));
2780
    };
2781
    let ty = ObjectType::from_name(ty)
2782
        .ok_or_else(|| StoreError::Corrupt(*r, format!("unknown type {ty}")))?;
2783
    let len: usize = len
2784
        .parse()
2785
        .map_err(|_| StoreError::Corrupt(*r, "bad length".into()))?;
2786
    let body = &data[nul + 1..];
2787
    if body.len() != len {
2788
        return Err(StoreError::Corrupt(
2789
            *r,
2790
            format!("length {} != header {len}", body.len()),
2791
        ));
2792
    }
2793
    if hash_object(ty, body) != *r {
2794
        return Err(StoreError::Corrupt(*r, "hash mismatch".into()));
2795
    }
2796
    Ok((ty, body.to_vec()))
2797
}
2798
2799
/// Whether the objects are files that a test may tamper with: not under
2800
/// `OXY_STORE_BACKEND=memory`.
2801
#[cfg(test)]
2802
fn files_on_disk() -> bool {
2803
    std::env::var_os("OXY_STORE_BACKEND").is_none()
2804
}
2805
2806
#[cfg(test)]
2807
mod tests {
2808
    use super::*;
2809
2810
    /// The store under test twice: in a directory, and on a store in
2811
    /// memory with a scratch directory, as over a bucket.
2812
    fn each_store(dir: &Path, f: impl Fn(&str, Store)) {
2813
        f("directory", Store::open(dir.join("objects")).unwrap());
2814
        f(
2815
            "memory",
2816
            Store::open_on(Arc::new(MemoryStore::new()), dir.join("scratch")).unwrap(),
2817
        );
2818
    }
2819
2820
    #[test]
2821
    fn put_get_verify() {
2822
        if !super::files_on_disk() {
2823
            return;
2824
        }
2825
        let dir = tempfile::tempdir().unwrap();
2826
        let s = Store::open(dir.path().join("objects")).unwrap();
2827
        let r = s.put(ObjectType::Chunk, b"hello").unwrap();
2828
        assert!(s.has(&r));
2829
        assert_eq!(s.get(&r).unwrap(), (ObjectType::Chunk, b"hello".to_vec()));
2830
        assert_eq!(s.put(ObjectType::Chunk, b"hello").unwrap(), r);
2831
        assert_eq!(s.list().unwrap(), vec![r]);
2832
        assert!(matches!(
2833
            s.get_typed(&r, ObjectType::Blob),
2834
            Err(StoreError::WrongType(..))
2835
        ));
2836
        assert!(matches!(
2837
            s.get(&ObjectRef([9; 32])),
2838
            Err(StoreError::NotFound(_))
2839
        ));
2840
2841
        // Tamper: flip a byte, read must fail.
2842
        let path = s.path_for(&r);
2843
        let mut data = fs::read(&path).unwrap();
2844
        let last = data.len() - 1;
2845
        data[last] ^= 1;
2846
        fs::write(&path, data).unwrap();
2847
        assert!(matches!(s.get(&r), Err(StoreError::Corrupt(..))));
2848
    }
2849
2850
    #[test]
2851
    fn packing_keeps_every_object_readable() {
2852
        if !super::files_on_disk() {
2853
            return;
2854
        }
2855
        let dir = tempfile::tempdir().unwrap();
2856
        let s = Store::open(dir.path()).unwrap();
2857
        let mut refs = Vec::new();
2858
        for i in 0..300u32 {
2859
            let body = format!("object {i} ").repeat(1 + (i % 7) as usize);
2860
            refs.push((s.put(ObjectType::Chunk, body.as_bytes()).unwrap(), body));
2861
        }
2862
        let c = crate::objects::tests::sample_change();
2863
        let cr = s.put_change(&c).unwrap();
2864
        assert_eq!(s.loose_count().unwrap(), 301);
2865
        let report = s.pack_loose().unwrap();
2866
        assert_eq!(report.objects, 301);
2867
        assert!(report.packed_bytes < report.raw_bytes, "{report:?}");
2868
        assert_eq!(s.loose_count().unwrap(), 0);
2869
        assert_eq!(s.pack_count(), 1);
2870
        for (r, body) in &refs {
2871
            assert!(s.has(r));
2872
            assert_eq!(
2873
                s.get(r).unwrap(),
2874
                (ObjectType::Chunk, body.as_bytes().to_vec())
2875
            );
2876
        }
2877
        assert_eq!(s.get_change(&cr).unwrap(), c);
2878
        assert_eq!(s.list().unwrap().len(), 301);
2879
        // Re-putting a packed object is a no-op; new objects go loose and a
2880
        // second pack coexists with the first.
2881
        assert_eq!(
2882
            s.put(ObjectType::Chunk, refs[0].1.as_bytes()).unwrap(),
2883
            refs[0].0
2884
        );
2885
        assert_eq!(s.loose_count().unwrap(), 0);
2886
        let extra = s.put(ObjectType::Chunk, b"after packing").unwrap();
2887
        s.pack_loose().unwrap();
2888
        assert_eq!(s.pack_count(), 2);
2889
        assert_eq!(s.get(&extra).unwrap().1, b"after packing");
2890
        // A reopened store sees the packs.
2891
        let s2 = Store::open(dir.path()).unwrap();
2892
        assert_eq!(s2.pack_count(), 2);
2893
        assert_eq!(s2.get(&refs[5].0).unwrap().1, refs[5].1.as_bytes());
2894
        assert!(matches!(
2895
            s2.get(&ObjectRef([9; 32])),
2896
            Err(StoreError::NotFound(_))
2897
        ));
2898
        // Corrupting an object's body inside the pack is detected on read.
2899
        let target = refs[5].0;
2900
        let (pack_path, offset) = {
2901
            let packs = s2.packs.borrow();
2902
            let (i, e) = s2.find_packed(&target).unwrap();
2903
            (dir.path().join(&packs[i].key), e.offset as usize)
2904
        };
2905
        let mut data = fs::read(&pack_path).unwrap();
2906
        data[offset] ^= 0xff;
2907
        fs::write(&pack_path, data).unwrap();
2908
        let s3 = Store::open(dir.path()).unwrap();
2909
        assert!(matches!(s3.get(&target), Err(StoreError::Corrupt(..))));
2910
        assert_eq!(
2911
            s3.get(&refs[6].0).unwrap().1,
2912
            refs[6].1.as_bytes(),
2913
            "other objects are unaffected"
2914
        );
2915
    }
2916
2917
    #[test]
2918
    fn a_full_repack_trains_a_dictionary_and_reads_back() {
2919
        if !super::files_on_disk() {
2920
            return;
2921
        }
2922
        let dir = tempfile::tempdir().unwrap();
2923
        let s = Store::open(dir.path()).unwrap();
2924
        // Many small, alike files: what a dictionary is for.
2925
        let bodies: Vec<Vec<u8>> = (0..1500)
2926
            .map(|i| {
2927
                format!(
2928
                    "// SPDX-License-Identifier: GPL-2.0\n#include <linux/module.h>\n\
2929
                     static int probe_{i}(struct device *dev)\n{{\n\treturn {i};\n}}\n"
2930
                )
2931
                .into_bytes()
2932
            })
2933
            .collect();
2934
        let refs: Vec<ObjectRef> = bodies
2935
            .iter()
2936
            .map(|b| s.put(ObjectType::Chunk, b).unwrap())
2937
            .collect();
2938
        s.repack_all(REPACK_ZSTD_LEVEL).unwrap();
2939
        let dicts = fs::read_dir(dir.path())
2940
            .unwrap()
2941
            .flatten()
2942
            .filter(|e| e.path().extension().is_some_and(|x| x == "dict"))
2943
            .count();
2944
        assert_eq!(dicts, 1, "the pack has a dictionary");
2945
        let with_dict = refs
2946
            .iter()
2947
            .filter(|r| s.find_packed(r).unwrap().1.flags & FLAG_DICT != 0)
2948
            .count();
2949
        assert!(with_dict > 1000, "most use it: {with_dict}");
2950
        let again = Store::open(dir.path()).unwrap();
2951
        for (r, b) in refs.iter().zip(&bodies) {
2952
            assert_eq!(&again.get(r).unwrap().1, b);
2953
        }
2954
        let packs = |dir: &Path| {
2955
            let mut v: Vec<String> = fs::read_dir(dir)
2956
                .unwrap()
2957
                .flatten()
2958
                .map(|e| e.file_name().to_string_lossy().into_owned())
2959
                .filter(|n| n.starts_with("pack-"))
2960
                .collect();
2961
            v.sort();
2962
            v
2963
        };
2964
        // With nothing new, a repack leaves that pack as it is: it would
2965
        // only write the same again.
2966
        let before = packs(dir.path());
2967
        let (report, merged) = again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
2968
        assert_eq!((report.objects, merged), (0, 0));
2969
        assert_eq!(packs(dir.path()), before);
2970
        // One new object: a repack replaces the pack and its dictionary,
2971
        // retiring the old ones, which a collection an hour on deletes.
2972
        let extra = again.put(ObjectType::Chunk, b"one more\n").unwrap();
2973
        again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
2974
        let files = packs(dir.path());
2975
        assert_eq!(
2976
            files.len(),
2977
            7,
2978
            "two of pack, index and dictionary, and the retired record: {files:?}"
2979
        );
2980
        assert_eq!(again.pack_count(), 1, "the old pack is not read");
2981
        assert_eq!(&again.get(&refs[7]).unwrap().1, &bodies[7]);
2982
        assert_eq!(again.get(&extra).unwrap().1, b"one more\n");
2983
        again
2984
            .expire_retired(std::time::Duration::ZERO, &mut silent())
2985
            .unwrap();
2986
        let files = packs(dir.path());
2987
        assert_eq!(files.len(), 4, "one pack, index, dictionary: {files:?}");
2988
        assert!(before
2989
            .iter()
2990
            .filter(|f| *f != RETIRED_KEY)
2991
            .all(|f| !files.contains(f)));
2992
        assert_eq!(&again.get(&refs[8]).unwrap().1, &bodies[8]);
2993
    }
2994
2995
    #[test]
2996
    fn nocompress_chunks_pack_raw() {
2997
        let dir = tempfile::tempdir().unwrap();
2998
        let s = Store::open(dir.path()).unwrap();
2999
        let body = vec![b'a'; 10_000];
3000
        let raw = s.put_chunk_nocompress(&body).unwrap();
3001
        let compressible = s.put(ObjectType::Chunk, &vec![b'b'; 10_000]).unwrap();
3002
        s.pack_loose().unwrap();
3003
        let packs = s.packs.borrow();
3004
        let (i, e) = s.find_packed(&raw).unwrap();
3005
        assert_eq!((e.flags, e.stored_len as usize), (0, 10_000));
3006
        let (_, c) = s.find_packed(&compressible).unwrap();
3007
        assert_eq!(c.flags, FLAG_ZSTD);
3008
        assert!(c.stored_len < 10_000);
3009
        assert_eq!(packs[i].fetch(s.backend.as_ref(), &raw, e).unwrap().1, body);
3010
    }
3011
3012
    #[test]
3013
    fn many_packs_merge_into_one() {
3014
        let dir = tempfile::tempdir().unwrap();
3015
        each_store(dir.path(), |which, s| {
3016
            let mut refs = Vec::new();
3017
            for i in 0..(MAX_PACKS + 2) {
3018
                refs.push(
3019
                    s.put(ObjectType::Chunk, format!("pack {i}").as_bytes())
3020
                        .unwrap(),
3021
                );
3022
                s.pack_loose().unwrap();
3023
            }
3024
            assert_eq!(s.pack_count(), MAX_PACKS + 2, "{which}");
3025
            assert_eq!(s.merge_packs().unwrap(), MAX_PACKS + 2);
3026
            assert_eq!(s.pack_count(), 1);
3027
            assert_eq!(s.merge_packs().unwrap(), 0);
3028
            for (i, r) in refs.iter().enumerate() {
3029
                assert_eq!(s.get(r).unwrap().1, format!("pack {i}").as_bytes());
3030
            }
3031
            let again = Store::open_on(s.backend().clone(), s.dir()).unwrap();
3032
            assert_eq!(again.pack_count(), 1, "{which}: reopened");
3033
        });
3034
    }
3035
3036
    #[test]
3037
    fn prune_removes_unreachable_loose_and_packed_objects() {
3038
        use std::collections::HashSet;
3039
        let dir = tempfile::tempdir().unwrap();
3040
        each_store(dir.path(), |which, s| {
3041
            let keep_a = s.put(ObjectType::Chunk, b"keep a").unwrap();
3042
            let drop_a = s.put(ObjectType::Chunk, b"drop a").unwrap();
3043
            s.pack_loose().unwrap();
3044
            let keep_b = s.put(ObjectType::Chunk, b"keep b").unwrap();
3045
            let drop_b = s.put(ObjectType::Chunk, b"drop b").unwrap();
3046
            let sk = crate::identity::SecretKey {
3047
                ed_seed: [1; 32],
3048
                ml_seed: [2; 32],
3049
            };
3050
            let id = sk.identity("a", "b", 0);
3051
            let key = s.put_identity(&id).unwrap();
3052
            s.put_signature(
3053
                &drop_b,
3054
                &sk.sign(crate::identity::Domain::Change, &key, &drop_b),
3055
            )
3056
            .unwrap();
3057
            s.index_seal(&drop_a, &ObjectRef([3; 32])).unwrap();
3058
            let keep: HashSet<ObjectRef> = [keep_a, keep_b, key].into_iter().collect();
3059
            // Everything is too young under a grace period.
3060
            let r = s
3061
                .prune(&keep, std::time::Duration::from_secs(3600))
3062
                .unwrap();
3063
            assert_eq!((r.objects, r.deferred), (0, 2), "{which}");
3064
            let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
3065
            assert_eq!(r.objects, 2, "{which}: {r:?}");
3066
            assert!(s.has(&keep_a) && s.has(&keep_b) && s.has(&key));
3067
            assert!(!s.has(&drop_a) && !s.has(&drop_b));
3068
            assert!(s.signatures(&drop_b).unwrap().is_empty());
3069
            assert!(s.seals_for(&drop_a).unwrap().is_empty());
3070
            assert_eq!(s.pack_count(), 1);
3071
            assert_eq!(s.get(&keep_a).unwrap().1, b"keep a");
3072
            assert_eq!(s.list().unwrap().len(), 3);
3073
        });
3074
    }
3075
3076
    #[test]
3077
    fn signatures_are_stored_beside_objects() {
3078
        use crate::identity::{Domain, SecretKey};
3079
        let dir = tempfile::tempdir().unwrap();
3080
        each_store(dir.path(), |which, s| {
3081
            let r = s.put(ObjectType::Chunk, b"x").unwrap();
3082
            assert!(s.signatures(&r).unwrap().is_empty());
3083
            let sk = SecretKey {
3084
                ed_seed: [1; 32],
3085
                ml_seed: [2; 32],
3086
            };
3087
            let id = sk.identity("a", "b", 0);
3088
            let key = s.put_identity(&id).unwrap();
3089
            assert_eq!(s.get_identity(&key).unwrap(), id);
3090
            let sig = sk.sign(Domain::Change, &key, &r);
3091
            s.put_signature(&r, &sig).unwrap();
3092
            s.put_signature(&r, &sig).unwrap();
3093
            assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()], "{which}");
3094
            // A second signer is kept beside the first, sorted by key.
3095
            let sk2 = SecretKey {
3096
                ed_seed: [3; 32],
3097
                ml_seed: [4; 32],
3098
            };
3099
            let key2 = s.put_identity(&sk2.identity("c", "d", 0)).unwrap();
3100
            let sig2 = sk2.sign(Domain::Change, &key2, &r);
3101
            s.put_signature(&r, &sig2).unwrap();
3102
            let mut both = vec![sig.clone(), sig2.clone()];
3103
            both.sort_by_key(|s| s.key);
3104
            assert_eq!(s.signatures(&r).unwrap(), both);
3105
            // The object itself is untouched by its signatures.
3106
            assert_eq!(s.get(&r).unwrap().1, b"x");
3107
            // Seal and receipt entries, and the states with receipts.
3108
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
3109
            s.index_seal(&r, &ObjectRef([4; 32])).unwrap();
3110
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
3111
            assert_eq!(
3112
                s.seals_for(&r).unwrap(),
3113
                vec![ObjectRef([4; 32]), ObjectRef([5; 32])]
3114
            );
3115
            let st = ObjectRef([6; 32]);
3116
            s.index_receipt(&st, &ObjectRef([7; 32])).unwrap();
3117
            assert_eq!(s.receipts_for(&st).unwrap(), vec![ObjectRef([7; 32])]);
3118
            assert_eq!(s.receipt_states().unwrap(), vec![st]);
3119
        });
3120
    }
3121
3122
    /// A store written before signatures, seals and receipts were one
3123
    /// entry each (`sigs/`, `seals/`, `receipts/` lists) still reads, and
3124
    /// what is written now sits beside what was there.
3125
    #[test]
3126
    fn an_older_store_s_sidecar_lists_are_still_read() {
3127
        if !super::files_on_disk() {
3128
            return;
3129
        }
3130
        use crate::identity::{Domain, SecretKey};
3131
        let dir = tempfile::tempdir().unwrap();
3132
        let s = Store::open(dir.path()).unwrap();
3133
        let r = s.put(ObjectType::Chunk, b"old").unwrap();
3134
        let h = r.to_hex();
3135
        let sk = SecretKey {
3136
            ed_seed: [1; 32],
3137
            ml_seed: [2; 32],
3138
        };
3139
        let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap();
3140
        let sig = sk.sign(Domain::Change, &key, &r);
3141
        let legacy = dir.path().join("sigs").join(&h[..2]);
3142
        fs::create_dir_all(&legacy).unwrap();
3143
        fs::write(
3144
            legacy.join(&h[2..]),
3145
            Signature::encode_list(std::slice::from_ref(&sig)),
3146
        )
3147
        .unwrap();
3148
        let seals = dir.path().join("seals").join(&h[..2]);
3149
        fs::create_dir_all(&seals).unwrap();
3150
        fs::write(
3151
            seals.join(&h[2..]),
3152
            format!("{}\n", ObjectRef([9; 32]).to_hex()),
3153
        )
3154
        .unwrap();
3155
        let receipts = dir.path().join("receipts").join(&h[..2]);
3156
        fs::create_dir_all(&receipts).unwrap();
3157
        fs::write(
3158
            receipts.join(&h[2..]),
3159
            format!("{}\n", ObjectRef([8; 32]).to_hex()),
3160
        )
3161
        .unwrap();
3162
3163
        assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()]);
3164
        assert_eq!(s.seals_for(&r).unwrap(), vec![ObjectRef([9; 32])]);
3165
        assert_eq!(s.receipts_for(&r).unwrap(), vec![ObjectRef([8; 32])]);
3166
        assert_eq!(s.receipt_states().unwrap(), vec![r]);
3167
        // New entries join the old ones; the same signer is not doubled.
3168
        s.put_signature(&r, &sig).unwrap();
3169
        s.index_seal(&r, &ObjectRef([1; 32])).unwrap();
3170
        assert_eq!(s.signatures(&r).unwrap().len(), 1);
3171
        assert_eq!(
3172
            s.seals_for(&r).unwrap(),
3173
            vec![ObjectRef([1; 32]), ObjectRef([9; 32])]
3174
        );
3175
        // Pruning the object takes both layouts with it.
3176
        let r2 = s.put(ObjectType::Chunk, b"kept").unwrap();
3177
        let keep = [r2, key].into_iter().collect();
3178
        s.prune(&keep, std::time::Duration::ZERO).unwrap();
3179
        assert!(s.signatures(&r).unwrap().is_empty());
3180
        assert!(s.seals_for(&r).unwrap().is_empty());
3181
        assert!(s.receipt_states().unwrap().is_empty());
3182
    }
3183
3184
    #[test]
3185
    fn change_round_trip_through_store() {
3186
        let dir = tempfile::tempdir().unwrap();
3187
        each_store(dir.path(), |which, s| {
3188
            let c = crate::objects::tests::sample_change();
3189
            let r = s.put_change(&c).unwrap();
3190
            assert_eq!(r, c.hash(), "{which}");
3191
            assert_eq!(s.get_change(&r).unwrap(), c);
3192
            assert_eq!(s.get_change_meta(&r).unwrap().message, c.message);
3193
        });
3194
    }
3195
3196
    fn xorshift(state: &mut u64) -> u64 {
3197
        *state ^= *state << 13;
3198
        *state ^= *state >> 7;
3199
        *state ^= *state << 17;
3200
        *state
3201
    }
3202
3203
    /// Random corruption of pack and index files never panics: reads fail
3204
    /// with an error or the store refuses to open, and untouched objects
3205
    /// stay readable when the index survives.
3206
    #[test]
3207
    fn corrupted_packs_and_indexes_fail_cleanly() {
3208
        if !super::files_on_disk() {
3209
            return;
3210
        }
3211
        let dir = tempfile::tempdir().unwrap();
3212
        let objects = dir.path().join("objects");
3213
        let s = Store::open(&objects).unwrap();
3214
        let mut refs = Vec::new();
3215
        let mut seed = 0x9e3779b97f4a7c15u64;
3216
        for i in 0..40u32 {
3217
            let len = (xorshift(&mut seed) % 5000) as usize;
3218
            let mut body = vec![0u8; len];
3219
            for b in body.iter_mut() {
3220
                *b = if i % 2 == 0 {
3221
                    (xorshift(&mut seed) & 0xff) as u8
3222
                } else {
3223
                    b'a' + (i % 26) as u8
3224
                };
3225
            }
3226
            refs.push(s.put(ObjectType::Chunk, &body).unwrap());
3227
        }
3228
        s.pack_loose().unwrap();
3229
        let pack = fs::read_dir(&objects)
3230
            .unwrap()
3231
            .filter_map(|e| e.ok().map(|e| e.path()))
3232
            .find(|p| p.extension().is_some_and(|x| x == "pack"))
3233
            .unwrap();
3234
        let idx = pack.with_extension("idx");
3235
        let pristine_pack = fs::read(&pack).unwrap();
3236
        let pristine_idx = fs::read(&idx).unwrap();
3237
3238
        for round in 0..300u32 {
3239
            let (path, pristine) = if round % 2 == 0 {
3240
                (&pack, &pristine_pack)
3241
            } else {
3242
                (&idx, &pristine_idx)
3243
            };
3244
            let mut data = pristine.clone();
3245
            match xorshift(&mut seed) % 4 {
3246
                0 => {
3247
                    let at = (xorshift(&mut seed) as usize) % data.len();
3248
                    data[at] ^= 1 << (xorshift(&mut seed) % 8);
3249
                }
3250
                1 => {
3251
                    let at = (xorshift(&mut seed) as usize) % data.len();
3252
                    data[at] = (xorshift(&mut seed) & 0xff) as u8;
3253
                }
3254
                2 => {
3255
                    let keep = (xorshift(&mut seed) as usize) % data.len();
3256
                    data.truncate(keep);
3257
                }
3258
                _ => {
3259
                    let at = (xorshift(&mut seed) as usize) % data.len();
3260
                    let n = ((xorshift(&mut seed) % 8) + 1) as usize;
3261
                    for k in 0..n.min(data.len() - at) {
3262
                        data[at + k] = 0xff;
3263
                    }
3264
                }
3265
            }
3266
            fs::write(path, &data).unwrap();
3267
            // Opening may refuse a broken index; a store that opens must
3268
            // either return each object intact or report it corrupt/missing.
3269
            if let Ok(s2) = Store::open(&objects) {
3270
                for r in &refs {
3271
                    match s2.get(r) {
3272
                        Ok((ty, body)) => assert_eq!(
3273
                            hash_object(ty, &body),
3274
                            *r,
3275
                            "a returned object always matches its reference"
3276
                        ),
3277
                        Err(StoreError::Corrupt(..))
3278
                        | Err(StoreError::NotFound(_))
3279
                        | Err(StoreError::Io(_)) => {}
3280
                        Err(e) => panic!("unexpected error {e}"),
3281
                    }
3282
                }
3283
            }
3284
            fs::write(path, pristine).unwrap();
3285
        }
3286
        // Restored files read back in full.
3287
        let s3 = Store::open(&objects).unwrap();
3288
        for r in &refs {
3289
            let (ty, body) = s3.get(r).unwrap();
3290
            assert_eq!(hash_object(ty, &body), *r);
3291
        }
3292
    }
3293
3294
    /// A small batch becomes loose objects; a large one becomes a single
3295
    /// pack. Objects are readable while pending, and batches nest.
3296
    #[test]
3297
    fn batches_write_loose_or_one_pack() {
3298
        let dir = tempfile::tempdir().unwrap();
3299
        each_store(dir.path(), |which, s| {
3300
            let outer = s.begin_batch().unwrap();
3301
            let a = s.put(ObjectType::Chunk, b"small a").unwrap();
3302
            {
3303
                let inner = s.begin_batch().unwrap();
3304
                let b = s.put(ObjectType::Chunk, b"small b").unwrap();
3305
                assert!(s.has(&b));
3306
                assert_eq!(s.get(&b).unwrap().1, b"small b");
3307
                inner.finish().unwrap();
3308
                assert_eq!(
3309
                    s.loose_count().unwrap(),
3310
                    0,
3311
                    "{which}: the inner batch defers to the outer one"
3312
                );
3313
            }
3314
            assert_eq!(s.get(&a).unwrap().1, b"small a");
3315
            assert_eq!(
3316
                s.put(ObjectType::Chunk, b"small a").unwrap(),
3317
                a,
3318
                "idempotent while pending"
3319
            );
3320
            outer.finish().unwrap();
3321
            assert_eq!(s.loose_count().unwrap(), 2);
3322
            assert_eq!(s.pack_count(), 0);
3323
            assert_eq!(s.get(&a).unwrap().1, b"small a");
3324
3325
            // Large: spills into one pack, readable during and after.
3326
            let big = s.begin_batch().unwrap();
3327
            let mut refs = Vec::new();
3328
            let mut seed = 7u64;
3329
            for i in 0..300u32 {
3330
                let mut body = vec![b'z'; 40_000];
3331
                if i % 3 == 0 {
3332
                    for b in body.iter_mut() {
3333
                        *b = (xorshift(&mut seed) & 0xff) as u8;
3334
                    }
3335
                }
3336
                body[..4].copy_from_slice(&i.to_le_bytes());
3337
                let r = if i % 5 == 0 {
3338
                    s.put_chunk_nocompress(&body).unwrap()
3339
                } else {
3340
                    s.put(ObjectType::Chunk, &body).unwrap()
3341
                };
3342
                assert_eq!(
3343
                    s.get(&r).unwrap().1,
3344
                    body,
3345
                    "readable while pending (object {i})"
3346
                );
3347
                refs.push((r, body));
3348
            }
3349
            assert!(s.has(&refs[0].0));
3350
            big.finish().unwrap();
3351
            assert_eq!(s.pack_count(), 1);
3352
            assert_eq!(
3353
                s.loose_count().unwrap(),
3354
                2,
3355
                "nothing from the big batch is loose"
3356
            );
3357
            assert!(fs::read_dir(s.dir())
3358
                .unwrap()
3359
                .flatten()
3360
                .all(|e| !e.file_name().to_string_lossy().contains("tmp-")));
3361
            let s2 = Store::open_on(s.backend().clone(), s.dir()).unwrap();
3362
            for (r, body) in &refs {
3363
                assert_eq!(s2.get(r).unwrap().1, *body);
3364
            }
3365
            // A dropped batch still writes its objects.
3366
            {
3367
                let _b = s2.begin_batch().unwrap();
3368
                s2.put(ObjectType::Chunk, b"dropped").unwrap();
3369
            }
3370
            assert_eq!(
3371
                s2.get(&hash_object(ObjectType::Chunk, b"dropped"))
3372
                    .unwrap()
3373
                    .1,
3374
                b"dropped"
3375
            );
3376
        });
3377
    }
3378
3379
    #[test]
3380
    fn get_many_reads_loose_and_packed_objects_in_order() {
3381
        let dir = tempfile::tempdir().unwrap();
3382
        each_store(dir.path(), |which, s| {
3383
            let packed: Vec<ObjectRef> = (0..20)
3384
                .map(|i| {
3385
                    s.put(ObjectType::Chunk, format!("packed {i}").as_bytes())
3386
                        .unwrap()
3387
                })
3388
                .collect();
3389
            s.pack_loose().unwrap();
3390
            let loose: Vec<ObjectRef> = (0..20)
3391
                .map(|i| {
3392
                    s.put(ObjectType::Chunk, format!("loose {i}").as_bytes())
3393
                        .unwrap()
3394
                })
3395
                .collect();
3396
            let missing = hash_object(ObjectType::Chunk, b"never stored");
3397
            let mut refs: Vec<ObjectRef> = Vec::new();
3398
            for i in 0..20 {
3399
                refs.push(loose[i]);
3400
                refs.push(packed[i]);
3401
            }
3402
            refs.push(missing);
3403
            let got = s.get_many(&refs);
3404
            assert_eq!(got.len(), refs.len());
3405
            for (i, r) in refs.iter().enumerate().take(40) {
3406
                let want = if i % 2 == 0 {
3407
                    format!("loose {}", i / 2)
3408
                } else {
3409
                    format!("packed {}", i / 2)
3410
                };
3411
                let (ty, body) = got[i].as_ref().unwrap();
3412
                assert_eq!(
3413
                    (*ty, body.as_slice()),
3414
                    (ObjectType::Chunk, want.as_bytes()),
3415
                    "{which}: {r:?}"
3416
                );
3417
            }
3418
            assert!(matches!(got[40], Err(StoreError::NotFound(_))));
3419
        });
3420
    }
3421
3422
    /// A packed object is read from its pack without first asking the
3423
    /// backend for a loose copy (oxygenhub#228: on a bucket each such ask
3424
    /// was a request that found nothing, every time).
3425
    #[test]
3426
    fn packed_objects_cost_no_loose_lookups() {
3427
        use std::sync::atomic::{AtomicUsize, Ordering};
3428
        struct Counting(MemoryStore, AtomicUsize);
3429
        impl crate::strata::Store for Counting {
3430
            fn describe(&self) -> String {
3431
                "counting".into()
3432
            }
3433
            fn has(&self, k: &str) -> crate::strata::Result<bool> {
3434
                self.0.has(k)
3435
            }
3436
            fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
3437
                if k.starts_with("loose/") {
3438
                    self.1.fetch_add(1, Ordering::Relaxed);
3439
                }
3440
                self.0.get(k)
3441
            }
3442
            fn get_range(
3443
                &self,
3444
                k: &str,
3445
                r: std::ops::Range<u64>,
3446
            ) -> crate::strata::Result<Vec<u8>> {
3447
                self.0.get_range(k, r)
3448
            }
3449
            fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
3450
                self.0.put_new(k, b)
3451
            }
3452
            fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
3453
                self.0.list(p)
3454
            }
3455
            fn delete(&self, k: &str) -> crate::strata::Result<()> {
3456
                self.0.delete(k)
3457
            }
3458
            fn read_ref(
3459
                &self,
3460
                k: &str,
3461
            ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
3462
                self.0.read_ref(k)
3463
            }
3464
            fn write_ref(
3465
                &self,
3466
                k: &str,
3467
                b: &[u8],
3468
                e: Option<&crate::strata::Version>,
3469
            ) -> crate::strata::Result<crate::strata::Version> {
3470
                self.0.write_ref(k, b, e)
3471
            }
3472
        }
3473
        let dir = tempfile::tempdir().unwrap();
3474
        let backend = Arc::new(Counting(MemoryStore::new(), AtomicUsize::new(0)));
3475
        let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap();
3476
        let refs: Vec<ObjectRef> = (0..10)
3477
            .map(|i| {
3478
                s.put(ObjectType::Chunk, format!("p {i}").as_bytes())
3479
                    .unwrap()
3480
            })
3481
            .collect();
3482
        s.pack_loose().unwrap();
3483
        let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap();
3484
        backend.1.store(0, Ordering::Relaxed);
3485
        for r in &refs {
3486
            assert!(s.get(r).is_ok());
3487
        }
3488
        assert!(s.get_many(&refs).iter().all(|g| g.is_ok()));
3489
        assert_eq!(backend.1.load(Ordering::Relaxed), 0, "no loose lookups");
3490
        // An object no pack holds is still looked for loose.
3491
        let missing = hash_object(ObjectType::Chunk, b"never stored");
3492
        assert!(matches!(s.get(&missing), Err(StoreError::NotFound(_))));
3493
        assert!(backend.1.load(Ordering::Relaxed) > 0);
3494
    }
3495
}
3496
3497
/// Readers never wait (oxygenhub#249): a store keeps reading while other
3498
/// handles repack, collect and push into the store under it.
3499
#[cfg(test)]
3500
pub(crate) mod reader_tests {
3501
    use super::*;
3502
3503
    fn chunk(i: usize) -> Vec<u8> {
3504
        format!("object {i} ").repeat(1 + i % 5).into_bytes()
3505
    }
3506
3507
    /// Two handles on one store: a writer, and a reader opened at once.
3508
    fn each_pair(f: impl Fn(&str, Store, Store)) {
3509
        let dir = tempfile::tempdir().unwrap();
3510
        let s = Store::open(dir.path().join("objects")).unwrap();
3511
        let r = Store::open(dir.path().join("objects")).unwrap();
3512
        f("directory", s, r);
3513
        let m: Arc<Backend> = Arc::new(MemoryStore::new());
3514
        let s = Store::open_on(m.clone(), dir.path().join("scratch")).unwrap();
3515
        let r = Store::open_on(m, dir.path().join("scratch-r")).unwrap();
3516
        f("memory", s, r);
3517
    }
3518
3519
    /// Set every retired entry's time back by `by`, as if retired then.
3520
    pub(crate) fn age_retired(s: &Store, by: std::time::Duration) {
3521
        s.update_retired(|m| {
3522
            for t in m.values_mut() {
3523
                *t = t.saturating_sub(by.as_millis() as u64);
3524
            }
3525
        })
3526
        .unwrap();
3527
    }
3528
3529
    fn retired_keys(s: &Store) -> Vec<String> {
3530
        s.read_retired().unwrap().0.into_keys().collect()
3531
    }
3532
3533
    #[test]
3534
    fn a_reader_opened_before_a_repack_reads_every_object_after_it() {
3535
        each_pair(|which, writer, _| {
3536
            let mut refs = Vec::new();
3537
            for i in 0..40 {
3538
                refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
3539
            }
3540
            writer.pack_loose().unwrap();
3541
            for i in 40..80 {
3542
                refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
3543
            }
3544
            // Opened now: one pack and forty loose objects.
3545
            let reader = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
3546
            assert_eq!(reader.pack_count(), 1, "{which}");
3547
            let (_, merged) = writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
3548
            assert_eq!(merged, 1);
3549
            // The old pack and the loose objects are retired, not deleted.
3550
            let retired = retired_keys(&writer);
3551
            assert_eq!(retired.len(), 81, "{which}: {retired:?}");
3552
            assert_eq!(writer.loose_count().unwrap(), 0, "retired is not loose");
3553
            // The reader reads on from what it loaded.
3554
            for (i, r) in refs.iter().enumerate() {
3555
                assert_eq!(reader.get(r).unwrap().1, chunk(i), "{which}: {i}");
3556
            }
3557
            assert_eq!(reader.pack_count(), 1);
3558
            // A store opened now reads the new pack only.
3559
            let fresh = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
3560
            assert_eq!(fresh.pack_count(), 1);
3561
            assert_ne!(fresh.packs.borrow()[0].key, reader.packs.borrow()[0].key);
3562
            // Once the retired files are deleted, the reader finds every
3563
            // object in the new pack, by get, get_many and has alike.
3564
            let report = writer
3565
                .expire_retired(std::time::Duration::ZERO, &mut silent())
3566
                .unwrap();
3567
            assert_eq!((report.packs, report.loose), (1, 80), "{which}");
3568
            assert!(retired_keys(&writer).is_empty());
3569
            for (i, r) in refs.iter().enumerate() {
3570
                assert_eq!(reader.get(r).unwrap().1, chunk(i), "{which}: {i}");
3571
            }
3572
            let other = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
3573
            let before = other.packs.borrow()[0].key.clone();
3574
            writer.put(ObjectType::Chunk, b"new").unwrap();
3575
            writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
3576
            writer
3577
                .expire_retired(std::time::Duration::ZERO, &mut silent())
3578
                .unwrap();
3579
            assert!(other.get_many(&refs).iter().all(|g| g.is_ok()), "{which}");
3580
            assert_ne!(other.packs.borrow()[0].key, before);
3581
            let third = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
3582
            writer.put(ObjectType::Chunk, b"newer").unwrap();
3583
            writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
3584
            writer
3585
                .expire_retired(std::time::Duration::ZERO, &mut silent())
3586
                .unwrap();
3587
            assert!(refs.iter().all(|r| third.has(r)), "{which}");
3588
            assert!(!third.has(&hash_object(ObjectType::Chunk, b"never")));
3589
        });
3590
    }
3591
3592
    #[test]
3593
    fn a_reader_opened_before_a_spilled_batch_reads_its_objects() {
3594
        each_pair(|which, writer, reader| {
3595
            assert_eq!(reader.pack_count(), 0);
3596
            let mut refs = Vec::new();
3597
            {
3598
                let batch = writer.begin_batch().unwrap();
3599
                for i in 0..(SPILL_OBJECTS + 50) {
3600
                    refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
3601
                }
3602
                batch.finish().unwrap();
3603
            }
3604
            assert_eq!(writer.pack_count(), 1, "{which}: the batch spilled");
3605
            assert_eq!(reader.get(&refs[0]).unwrap().1, chunk(0), "{which}");
3606
            assert_eq!(reader.pack_count(), 1, "the miss loaded the new pack");
3607
            let got = reader.get_many(&refs);
3608
            assert!(got.iter().all(|g| g.is_ok()), "{which}");
3609
            // has, too, from another reader.
3610
            let late = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
3611
            {
3612
                let batch = writer.begin_batch().unwrap();
3613
                for i in 1000..(1000 + SPILL_OBJECTS + 1) {
3614
                    writer.put(ObjectType::Chunk, &chunk(i)).unwrap();
3615
                }
3616
                batch.finish().unwrap();
3617
            }
3618
            assert!(late.has(&hash_object(ObjectType::Chunk, &chunk(1000))));
3619
        });
3620
    }
3621
3622
    #[test]
3623
    fn retired_entries_expire_after_the_hour_and_are_never_loaded() {
3624
        each_pair(|which, s, _| {
3625
            for i in 0..10 {
3626
                s.put(ObjectType::Chunk, &chunk(i)).unwrap();
3627
            }
3628
            s.pack_loose().unwrap();
3629
            s.put(ObjectType::Chunk, &chunk(10)).unwrap();
3630
            s.repack_all(ZSTD_LEVEL).unwrap();
3631
            // Retired over an hour ago: the first pack and 10 loose objects.
3632
            age_retired(&s, RETIRED_FOR + std::time::Duration::from_secs(60));
3633
            let old = retired_keys(&s);
3634
            assert_eq!(old.len(), 12, "{which}: {old:?}");
3635
            // Retired just now: the second pack.
3636
            s.put(ObjectType::Chunk, &chunk(11)).unwrap();
3637
            s.repack_all(ZSTD_LEVEL).unwrap();
3638
            assert_eq!(retired_keys(&s).len(), 14);
3639
            let listed = s
3640
                .backend()
3641
                .list("pack-")
3642
                .unwrap()
3643
                .into_iter()
3644
                .filter(|e| e.key.ends_with(".idx"))
3645
                .count();
3646
            assert_eq!(listed, 3, "{which}");
3647
            let fresh = Store::open_on(s.backend().clone(), s.dir()).unwrap();
3648
            assert_eq!(
3649
                fresh.pack_count(),
3650
                1,
3651
                "{which}: retired packs are not loaded"
3652
            );
3653
            let report = s.expire_retired(RETIRED_FOR, &mut silent()).unwrap();
3654
            assert_eq!(
3655
                report,
3656
                RetiredReport {
3657
                    packs: 1,
3658
                    loose: 11,
3659
                    kept: 2
3660
                },
3661
                "{which}"
3662
            );
3663
            let young = retired_keys(&s);
3664
            assert_eq!(young.len(), 2, "{which}: {young:?}");
3665
            assert!(young.iter().all(|k| !old.contains(k)));
3666
            for k in old.iter().filter(|k| k.ends_with(".pack")) {
3667
                assert!(!s.backend().has(k).unwrap(), "{which}: {k}");
3668
            }
3669
            for k in &young {
3670
                assert!(s.backend().has(k).unwrap(), "{which}: {k} kept");
3671
            }
3672
            for i in 0..12 {
3673
                let r = hash_object(ObjectType::Chunk, &chunk(i));
3674
                assert_eq!(fresh.get(&r).unwrap().1, chunk(i), "{which}");
3675
            }
3676
        });
3677
    }
3678
3679
    /// A retired loose object whose pack copy was pruned since is not
3680
    /// lost when the retired files go: it is loose again.
3681
    #[test]
3682
    fn a_retired_loose_object_no_pack_holds_is_kept() {
3683
        each_pair(|which, s, _| {
3684
            let a = s.put(ObjectType::Chunk, b"a").unwrap();
3685
            let b = s.put(ObjectType::Chunk, b"b").unwrap();
3686
            s.pack_loose().unwrap();
3687
            assert_eq!(s.loose_count().unwrap(), 0);
3688
            // The pack rewritten without b, as pruning does.
3689
            let first = s.packs.borrow()[0].key.clone();
3690
            s.write_pack(&[a], &[], &[first], ZSTD_LEVEL, &mut silent())
3691
                .unwrap();
3692
            let report = s
3693
                .expire_retired(std::time::Duration::ZERO, &mut silent())
3694
                .unwrap();
3695
            assert_eq!((report.packs, report.loose), (1, 1), "{which}: a's copy");
3696
            assert!(retired_keys(&s).is_empty());
3697
            assert_eq!(s.get(&b).unwrap().1, b"b", "{which}");
3698
            assert_eq!(s.get(&a).unwrap().1, b"a");
3699
            assert_eq!(s.loose_count().unwrap(), 1, "b is loose again");
3700
            // And pruning it, unreachable, deletes it.
3701
            let keep = [a].into_iter().collect();
3702
            let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
3703
            assert_eq!(r.objects, 1);
3704
            assert!(!s.has(&b));
3705
        });
3706
    }
3707
3708
    /// A backend whose first pack listing names an index that is deleted
3709
    /// before it is read, as a collection running meanwhile would.
3710
    pub(crate) struct Racing {
3711
        inner: MemoryStore,
3712
        lists: std::sync::atomic::AtomicUsize,
3713
        /// Deleted right after the first listing returns.
3714
        pub(crate) doomed: std::sync::Mutex<Option<String>>,
3715
        /// Named in every listing, never stored.
3716
        ghost: bool,
3717
    }
3718
3719
    impl crate::strata::Store for Racing {
3720
        fn describe(&self) -> String {
3721
            "racing".into()
3722
        }
3723
        fn has(&self, k: &str) -> crate::strata::Result<bool> {
3724
            self.inner.has(k)
3725
        }
3726
        fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
3727
            self.inner.get(k)
3728
        }
3729
        fn get_range(&self, k: &str, r: std::ops::Range<u64>) -> crate::strata::Result<Vec<u8>> {
3730
            self.inner.get_range(k, r)
3731
        }
3732
        fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
3733
            self.inner.put_new(k, b)
3734
        }
3735
        fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
3736
            let mut out = self.inner.list(p)?;
3737
            if p == "pack-" {
3738
                self.lists
3739
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
3740
                if let Some(k) = self.doomed.lock().unwrap().take() {
3741
                    self.inner.delete(&k)?;
3742
                }
3743
                if self.ghost {
3744
                    for k in ["pack-0000.idx", "pack-0000.pack"] {
3745
                        out.push(crate::strata::Entry {
3746
                            key: k.into(),
3747
                            size: 100,
3748
                            modified_ms: None,
3749
                        });
3750
                    }
3751
                    out.sort_by(|a, b| a.key.cmp(&b.key));
3752
                }
3753
            }
3754
            Ok(out)
3755
        }
3756
        fn delete(&self, k: &str) -> crate::strata::Result<()> {
3757
            self.inner.delete(k)
3758
        }
3759
        fn read_ref(
3760
            &self,
3761
            k: &str,
3762
        ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
3763
            self.inner.read_ref(k)
3764
        }
3765
        fn write_ref(
3766
            &self,
3767
            k: &str,
3768
            b: &[u8],
3769
            e: Option<&crate::strata::Version>,
3770
        ) -> crate::strata::Result<crate::strata::Version> {
3771
            self.inner.write_ref(k, b, e)
3772
        }
3773
    }
3774
3775
    pub(crate) fn racing(ghost: bool) -> Arc<Racing> {
3776
        Arc::new(Racing {
3777
            inner: MemoryStore::new(),
3778
            lists: Default::default(),
3779
            doomed: Default::default(),
3780
            ghost,
3781
        })
3782
    }
3783
3784
    #[test]
3785
    fn an_index_deleted_between_listing_and_loading_is_passed_over() {
3786
        let dir = tempfile::tempdir().unwrap();
3787
        let backend = racing(false);
3788
        let s = Store::open_on(backend.clone(), dir.path()).unwrap();
3789
        let a = s.put(ObjectType::Chunk, b"a").unwrap();
3790
        s.pack_loose().unwrap();
3791
        let b = s.put(ObjectType::Chunk, b"b").unwrap();
3792
        s.pack_loose().unwrap();
3793
        let idx = {
3794
            let packs = s.packs.borrow();
3795
            let (i, _) = s.find_packed(&a).unwrap();
3796
            format!("{}.idx", packs[i].key.trim_end_matches(".pack"))
3797
        };
3798
        *backend.doomed.lock().unwrap() = Some(idx);
3799
        let before = backend.lists.load(std::sync::atomic::Ordering::Relaxed);
3800
        let again = Store::open_on(backend.clone(), dir.path()).unwrap();
3801
        assert_eq!(again.pack_count(), 1, "the deleted one is passed over");
3802
        assert_eq!(
3803
            backend.lists.load(std::sync::atomic::Ordering::Relaxed) - before,
3804
            2,
3805
            "and the store listed again"
3806
        );
3807
        assert_eq!(again.get(&b).unwrap().1, b"b");
3808
        // A listing that names an index never there opens too.
3809
        let ghosted = racing(true);
3810
        let s = Store::open_on(ghosted.clone(), dir.path().join("g")).unwrap();
3811
        let c = s.put(ObjectType::Chunk, b"c").unwrap();
3812
        s.pack_loose().unwrap();
3813
        let again = Store::open_on(ghosted, dir.path().join("g")).unwrap();
3814
        assert_eq!(again.pack_count(), 1);
3815
        assert_eq!(again.get(&c).unwrap().1, b"c");
3816
    }
3817
}
3818
3819
#[cfg(test)]
3820
mod batch_tests {
3821
    use super::*;
3822
3823
    #[test]
3824
    fn a_large_batch_packs_every_object_and_indexes_them_all() {
3825
        let dir = tempfile::tempdir().unwrap();
3826
        let store = Store::open(dir.path()).unwrap();
3827
        let mut refs = Vec::new();
3828
        {
3829
            let batch = store.begin_batch().unwrap();
3830
            for i in 0..1000u32 {
3831
                let body = format!("object {i} {}", "x".repeat(20_000 + i as usize)).into_bytes();
3832
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
3833
            }
3834
            batch.finish().unwrap();
3835
        }
3836
        let again = Store::open(dir.path()).unwrap();
3837
        for r in &refs {
3838
            again
3839
                .get(r)
3840
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
3841
        }
3842
    }
3843
3844
    #[test]
3845
    fn a_pack_a_stopped_batch_left_is_adopted_and_its_torn_end_cut() {
3846
        // oxygen#79: a batch never finished (a process killed part way)
3847
        // leaves its objects in an unindexed .tmp pack.
3848
        let dir = tempfile::tempdir().unwrap();
3849
        let mut refs = Vec::new();
3850
        {
3851
            let store = Store::open(dir.path()).unwrap();
3852
            let batch = store.begin_batch().unwrap();
3853
            for i in 0..300u32 {
3854
                let body = format!("object {i} {}", "y".repeat(20_000 + i as usize)).into_bytes();
3855
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
3856
            }
3857
            std::mem::forget(batch);
3858
        }
3859
        let tmp: Vec<_> = fs::read_dir(dir.path())
3860
            .unwrap()
3861
            .flatten()
3862
            .filter(|e| e.file_name().to_string_lossy().starts_with(".tmp-"))
3863
            .map(|e| e.path())
3864
            .collect();
3865
        assert_eq!(tmp.len(), 1, "one unfinished pack");
3866
        // A torn last entry: a header and part of its body.
3867
        let mut f = fs::OpenOptions::new().append(true).open(&tmp[0]).unwrap();
3868
        f.write_all(&[0, 1, 0xff, 0, 0, 0, 9, 9, 0, 0]).unwrap();
3869
        f.write_all(&[7; 40]).unwrap();
3870
        drop(f);
3871
        let store = Store::open(dir.path()).unwrap();
3872
        assert!(store.get(&refs[0]).is_err(), "unreadable before");
3873
        let young = store
3874
            .adopt_unfinished_packs(std::time::Duration::from_secs(3600), &|_| {})
3875
            .unwrap();
3876
        assert_eq!(young, (0, 0), "a pack written to lately is left alone");
3877
        let (packs, objects) = store
3878
            .adopt_unfinished_packs(std::time::Duration::ZERO, &|_| {})
3879
            .unwrap();
3880
        assert_eq!((packs, objects), (1, 300));
3881
        for r in &refs {
3882
            store
3883
                .get(r)
3884
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
3885
        }
3886
        let again = Store::open(dir.path()).unwrap();
3887
        assert!(again.get(&refs[299]).is_ok(), "indexed for good");
3888
    }
3889
}
3890
3891
#[cfg(test)]
3892
mod size_tests {
3893
    use super::*;
3894
3895
    #[test]
3896
    fn an_object_over_the_limit_is_refused_and_nothing_is_written() {
3897
        let dir = tempfile::tempdir().unwrap();
3898
        let store = Store::open(dir.path()).unwrap();
3899
        let big = vec![b'x'; MAX_OBJECT_LEN as usize + 1];
3900
        {
3901
            let batch = store.begin_batch().unwrap();
3902
            store.put(ObjectType::Blob, b"small").unwrap();
3903
            let e = store.put(ObjectType::Change, &big).unwrap_err();
3904
            assert!(
3905
                matches!(e, StoreError::TooLarge { kind: "change", .. }),
3906
                "{e}"
3907
            );
3908
            assert!(e.to_string().contains("commit it in parts"), "{e}");
3909
            batch.finish().unwrap();
3910
        }
3911
        // The store still opens and reads what was written.
3912
        let again = Store::open(dir.path()).unwrap();
3913
        again.get(&hash_object(ObjectType::Blob, b"small")).unwrap();
3914
    }
3915
}
3916
3917
#[cfg(test)]
3918
mod text_tests {
3919
    use super::*;
3920
    use crate::objects::{AtomRef, Edit, CHANGE_VERSION};
3921
3922
    fn large_change() -> Change {
3923
        let mut c = crate::objects::tests::sample_change();
3924
        assert_eq!(c.version, CHANGE_VERSION);
3925
        c.edits[0] = Edit::Insert {
3926
            node: crate::ids::NodeId([7; 16]),
3927
            after: AtomRef::Start,
3928
            before: AtomRef::End,
3929
            lines: (0..2000)
3930
                .map(|i| format!("line {i}\n").into_bytes())
3931
                .collect(),
3932
        };
3933
        c
3934
    }
3935
3936
    #[test]
3937
    fn a_large_insert_is_stored_beside_its_change_and_read_back() {
3938
        let dir = tempfile::tempdir().unwrap();
3939
        let s = Store::open(dir.path()).unwrap();
3940
        let c = large_change();
3941
        let r = s.put_change(&c).unwrap();
3942
        assert_eq!(r, c.hash());
3943
        let (_, body) = s.get(&r).unwrap();
3944
        assert!(
3945
            body.len() < 1024,
3946
            "the change holds a reference, not {} bytes of lines",
3947
            body.len()
3948
        );
3949
        let (back, parts) = s.get_change_parts(&r).unwrap();
3950
        assert_eq!(back, c);
3951
        // The text blob and its chunk.
3952
        assert_eq!(parts.len(), 2);
3953
        for p in &parts {
3954
            assert!(s.has(p));
3955
        }
3956
        // Storing it again writes nothing new.
3957
        let n = s.list().unwrap().len();
3958
        s.put_change(&c).unwrap();
3959
        assert_eq!(s.list().unwrap().len(), n);
3960
    }
3961
3962
    #[test]
3963
    fn a_change_whose_text_blob_is_missing_is_reported_not_misread() {
3964
        if !super::files_on_disk() {
3965
            return;
3966
        }
3967
        let dir = tempfile::tempdir().unwrap();
3968
        let s = Store::open(dir.path()).unwrap();
3969
        let c = large_change();
3970
        let r = s.put_change(&c).unwrap();
3971
        let (_, parts) = s.get_change_parts(&r).unwrap();
3972
        fs::remove_file(s.path_for(&parts[0])).unwrap();
3973
        let e = s.get_change(&r).unwrap_err();
3974
        assert!(
3975
            matches!(&e, StoreError::Corrupt(h, why) if *h == r && why.contains("missing")),
3976
            "{e}"
3977
        );
3978
    }
3979
}
3980
3981
#[cfg(test)]
3982
mod put_many_tests {
3983
    use super::*;
3984
3985
    #[test]
3986
    fn many_objects_in_a_batch_are_compressed_together_and_all_read_back() {
3987
        let dir = tempfile::tempdir().unwrap();
3988
        let s = Store::open(dir.path()).unwrap();
3989
        let before = s.put(ObjectType::Chunk, b"already here").unwrap();
3990
        let bodies: Vec<Vec<u8>> = (0..200u32)
3991
            .map(|i| format!("chunk {i} ").repeat(6000).into_bytes())
3992
            .collect();
3993
        let mut objects: Vec<(ObjectType, &[u8])> = bodies
3994
            .iter()
3995
            .map(|b| (ObjectType::Chunk, b.as_slice()))
3996
            .collect();
3997
        objects.push((ObjectType::Chunk, bodies[0].as_slice()));
3998
        objects.push((ObjectType::Chunk, b"already here"));
3999
        let refs = {
4000
            let batch = s.begin_batch().unwrap();
4001
            let refs = s.put_many(&objects).unwrap();
4002
            batch.finish().unwrap();
4003
            refs
4004
        };
4005
        assert_eq!(refs.len(), objects.len());
4006
        assert_eq!(refs[200], refs[0], "a repeat is the same object");
4007
        assert_eq!(refs[201], before);
4008
        let again = Store::open(dir.path()).unwrap();
4009
        for (r, b) in refs.iter().zip(&objects) {
4010
            assert_eq!(again.get(r).unwrap().1, b.1);
4011
        }
4012
    }
4013
}
4014
4015
/// Signatures and index entries on a store that is not this disk's
4016
/// directory: the journal and the entry lists (spec §13.7, #116).
4017
#[cfg(test)]
4018
mod entry_tests {
4019
    use super::*;
4020
    use crate::identity::{Domain, SecretKey};
4021
    use crate::strata::Store as _;
4022
    use std::collections::HashMap;
4023
    use std::sync::atomic::{AtomicBool, Ordering};
4024
    use std::sync::Mutex;
4025
4026
    /// A store in memory that counts its requests by kind, as a bucket
4027
    /// store's /health counters do; it can also slip a competing journal
4028
    /// line in just before a write of the journal, as another process
4029
    /// would.
4030
    #[derive(Default)]
4031
    struct Counted {
4032
        inner: MemoryStore,
4033
        counts: Mutex<HashMap<&'static str, usize>>,
4034
        interfere: AtomicBool,
4035
    }
4036
4037
    impl Counted {
4038
        fn count(&self, kind: &'static str) {
4039
            *self.counts.lock().unwrap().entry(kind).or_default() += 1;
4040
        }
4041
        fn n(&self, kind: &str) -> usize {
4042
            self.counts.lock().unwrap().get(kind).copied().unwrap_or(0)
4043
        }
4044
        fn reset(&self) {
4045
            self.counts.lock().unwrap().clear();
4046
        }
4047
        fn total(&self) -> usize {
4048
            self.counts.lock().unwrap().values().sum()
4049
        }
4050
        fn summary(&self) -> String {
4051
            let c = self.counts.lock().unwrap();
4052
            let mut v: Vec<String> = c.iter().map(|(k, n)| format!("{k} {n}")).collect();
4053
            v.sort();
4054
            v.join(", ")
4055
        }
4056
    }
4057
4058
    impl crate::strata::Store for Counted {
4059
        fn describe(&self) -> String {
4060
            "counted".into()
4061
        }
4062
        fn has(&self, k: &str) -> crate::strata::Result<bool> {
4063
            self.count("has");
4064
            self.inner.has(k)
4065
        }
4066
        fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
4067
            self.count(if is_legacy(k) { "get_legacy" } else { "get" });
4068
            self.inner.get(k)
4069
        }
4070
        fn get_range(&self, k: &str, r: std::ops::Range<u64>) -> crate::strata::Result<Vec<u8>> {
4071
            self.count("range");
4072
            self.inner.get_range(k, r)
4073
        }
4074
        fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
4075
            self.count("put");
4076
            self.inner.put_new(k, b)
4077
        }
4078
        fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
4079
            self.count(if p == "pack-" {
4080
                "list_packs"
4081
            } else if is_legacy(p) {
4082
                "list_legacy"
4083
            } else {
4084
                "list"
4085
            });
4086
            self.inner.list(p)
4087
        }
4088
        fn delete(&self, k: &str) -> crate::strata::Result<()> {
4089
            self.count("delete");
4090
            self.inner.delete(k)
4091
        }
4092
        fn read_ref(
4093
            &self,
4094
            k: &str,
4095
        ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
4096
            self.count(if k == JOURNAL_KEY {
4097
                "read_journal"
4098
            } else {
4099
                "read_ref"
4100
            });
4101
            self.inner.read_ref(k)
4102
        }
4103
        fn write_ref(
4104
            &self,
4105
            k: &str,
4106
            b: &[u8],
4107
            e: Option<&crate::strata::Version>,
4108
        ) -> crate::strata::Result<crate::strata::Version> {
4109
            self.count("write_ref");
4110
            if k == JOURNAL_KEY && self.interfere.swap(false, Ordering::SeqCst) {
4111
                let (mut m, v) = match self.inner.read_ref(k)? {
4112
                    Some((bytes, v)) => (parse_retired(&bytes), Some(v)),
4113
                    None => (Default::default(), None),
4114
                };
4115
                m.insert(
4116
                    format!("seal/{}", "c".repeat(64)) + "/" + &"d".repeat(64),
4117
                    0,
4118
                );
4119
                self.inner.write_ref(k, &encode_retired(&m), v.as_ref())?;
4120
            }
4121
            self.inner.write_ref(k, b, e)
4122
        }
4123
    }
4124
4125
    /// Whether a key or prefix is under an older one-list layout.
4126
    fn is_legacy(k: &str) -> bool {
4127
        ENTRY_INDEXES
4128
            .iter()
4129
            .any(|(_, legacy)| k.starts_with(&format!("{legacy}/")))
4130
    }
4131
4132
    struct Fixture {
4133
        backend: Arc<Counted>,
4134
        dir: tempfile::TempDir,
4135
        handles: std::cell::Cell<usize>,
4136
    }
4137
4138
    impl Fixture {
4139
        fn new() -> Fixture {
4140
            Fixture {
4141
                backend: Arc::new(Counted::default()),
4142
                dir: tempfile::tempdir().unwrap(),
4143
                handles: Default::default(),
4144
            }
4145
        }
4146
        /// Another handle on the store, as another process opens it.
4147
        fn open(&self) -> Store {
4148
            let n = self.handles.get() + 1;
4149
            self.handles.set(n);
4150
            Store::open_on(self.backend.clone(), self.dir.path().join(format!("s{n}"))).unwrap()
4151
        }
4152
    }
4153
4154
    fn signer(n: u8) -> SecretKey {
4155
        SecretKey {
4156
            ed_seed: [n; 32],
4157
            ml_seed: [n + 100; 32],
4158
        }
4159
    }
4160
4161
    fn sign(s: &Store, r: &ObjectRef, n: u8) -> Signature {
4162
        let sk = signer(n);
4163
        let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap();
4164
        let sig = sk.sign(Domain::Change, &key, r);
4165
        s.put_signature(r, &sig).unwrap();
4166
        sig
4167
    }
4168
4169
    fn journal(s: &Store) -> std::collections::BTreeMap<String, u64> {
4170
        s.read_record(JOURNAL_KEY).unwrap().0
4171
    }
4172
4173
    fn fold(s: &Store) -> bool {
4174
        s.fold_entries(&mut silent()).unwrap()
4175
    }
4176
4177
    /// What 0.4.6 writes for a signature: its own key, and nothing else.
4178
    fn old_writer_signs(f: &Fixture, r: &ObjectRef, n: u8) -> Signature {
4179
        let sk = signer(n);
4180
        let sig = sk.sign(Domain::Change, &ObjectRef([n; 32]), r);
4181
        f.backend
4182
            .put_new(
4183
                &format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex()),
4184
                &Signature::encode_list(std::slice::from_ref(&sig)),
4185
            )
4186
            .unwrap();
4187
        sig
4188
    }
4189
4190
    /// After a fold, a signature or seal another handle writes is found
4191
    /// by a store opened later through the journal, with no listing; a
4192
    /// handle that already read the entries sees it once refreshed.
4193
    #[test]
4194
    fn an_entry_written_after_a_fold_is_found_through_the_journal() {
4195
        let f = Fixture::new();
4196
        let w = f.open();
4197
        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();
4198
        let s1 = sign(&w, &r1, 1);
4199
        assert!(fold(&w));
4200
        assert!(!fold(&w), "nothing new: no second list");
4201
4202
        let early = f.open();
4203
        assert_eq!(early.signatures(&r1).unwrap(), vec![s1.clone()]);
4204
4205
        let w2 = f.open();
4206
        let r2 = w2.put(ObjectType::Chunk, b"two").unwrap();
4207
        let s2 = sign(&w2, &r2, 2);
4208
        w2.index_seal(&r2, &ObjectRef([9; 32])).unwrap();
4209
        w2.index_receipt(&r2, &ObjectRef([8; 32])).unwrap();
4210
        // Its own writes, at once.
4211
        assert_eq!(w2.signatures(&r2).unwrap(), vec![s2.clone()]);
4212
4213
        let later = f.open();
4214
        f.backend.reset();
4215
        assert_eq!(later.signatures(&r1).unwrap(), vec![s1]);
4216
        assert_eq!(later.signatures(&r2).unwrap(), vec![s2.clone()]);
4217
        assert_eq!(later.seals_for(&r2).unwrap(), vec![ObjectRef([9; 32])]);
4218
        assert_eq!(later.receipts_for(&r2).unwrap(), vec![ObjectRef([8; 32])]);
4219
        assert_eq!(later.receipt_states().unwrap(), vec![r2]);
4220
        assert!(later.seals_for(&r1).unwrap().is_empty());
4221
        assert_eq!(f.backend.n("list"), 0, "{}", f.backend.summary());
4222
        assert_eq!(f.backend.n("read_journal"), 1, "{}", f.backend.summary());
4223
4224
        // The handle that read the entries before keeps its view until
4225
        // it is refreshed, as it keeps its operation head.
4226
        assert!(early.signatures(&r2).unwrap().is_empty());
4227
        early.refresh().unwrap();
4228
        assert_eq!(early.signatures(&r2).unwrap(), vec![s2]);
4229
    }
4230
4231
    /// A fold marks the journal's lines rather than dropping them, so a
4232
    /// reader whose listing predates it still finds every entry; an hour
4233
    /// on, the old list is deleted and the lines dropped, and readers
4234
    /// find everything in the new list.
4235
    #[test]
4236
    fn a_fold_keeps_the_journal_for_the_hour_readers_may_need_it() {
4237
        let f = Fixture::new();
4238
        let w = f.open();
4239
        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();
4240
        let s1 = sign(&w, &r1, 1);
4241
        assert!(fold(&w));
4242
        let r2 = w.put(ObjectType::Chunk, b"two").unwrap();
4243
        let s2 = sign(&w, &r2, 2);
4244
        assert!(journal(&w).values().any(|t| *t == 0));
4245
4246
        // Opened before the second fold: lists the first list only.
4247
        let stale = f.open();
4248
        let stale_after_expiry = f.open();
4249
        assert!(fold(&w));
4250
        assert!(
4251
            journal(&w).values().all(|t| *t != 0),
4252
            "every line is marked folded"
4253
        );
4254
        assert!(journal(&w).len() >= 2, "and kept");
4255
        assert_eq!(stale.signatures(&r1).unwrap(), vec![s1.clone()]);
4256
        assert_eq!(stale.signatures(&r2).unwrap(), vec![s2.clone()]);
4257
4258
        // An hour later: the first list goes with the next collection,
4259
        // and the next journal update drops the folded lines.
4260
        reader_tests::age_retired(&w, RETIRED_FOR * 2);
4261
        w.update_record(JOURNAL_KEY, "test", |m| {
4262
            for t in m.values_mut() {
4263
                *t = t.saturating_sub(2 * RETIRED_FOR.as_millis() as u64);
4264
            }
4265
        })
4266
        .unwrap();
4267
        w.expire_retired(RETIRED_FOR, &mut silent()).unwrap();
4268
        let lists: Vec<String> = f
4269
            .backend
4270
            .inner
4271
            .list("pack-")
4272
            .unwrap()
4273
            .into_iter()
4274
            .map(|e| e.key)
4275
            .filter(|k| is_entry_list(k))
4276
            .collect();
4277
        assert_eq!(lists.len(), 1, "the replaced list is deleted: {lists:?}");
4278
        let r3 = w.put(ObjectType::Chunk, b"three").unwrap();
4279
        let s3 = sign(&w, &r3, 3);
4280
        let j = journal(&w);
4281
        assert_eq!(j.len(), 1, "only the new line is left: {j:?}");
4282
        assert!(j.keys().all(|k| k.starts_with(&fanout_prefix("sig", &r3))));
4283
4284
        // A handle whose listing named the deleted list lists again.
4285
        assert_eq!(
4286
            stale_after_expiry.signatures(&r1).unwrap(),
4287
            vec![s1.clone()]
4288
        );
4289
        assert_eq!(
4290
            stale_after_expiry.signatures(&r3).unwrap(),
4291
            vec![s3.clone()]
4292
        );
4293
        let fresh = f.open();
4294
        f.backend.reset();
4295
        assert_eq!(fresh.signatures(&r1).unwrap(), vec![s1]);
4296
        assert_eq!(fresh.signatures(&r2).unwrap(), vec![s2]);
4297
        assert_eq!(fresh.signatures(&r3).unwrap(), vec![s3]);
4298
        assert_eq!(f.backend.n("list"), 0);
4299
    }
4300
4301
    /// A store no fold has touched (as 0.4.6 left it: per-object keys and
4302
    /// older one-list keys, no journal, no lists) reads by listing, and
4303
    /// reads the same once folded. An older writer's entry after the
4304
    /// fold is missed until the next fold: the trade-off of §13.7.
4305
    #[test]
4306
    fn an_old_layout_store_reads_before_and_after_its_first_fold() {
4307
        let f = Fixture::new();
4308
        let w = f.open();
4309
        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();
4310
        let r2 = w.put(ObjectType::Chunk, b"two").unwrap();
4311
        let s1 = old_writer_signs(&f, &r1, 1);
4312
        let s2 = SecretKey {
4313
            ed_seed: [2; 32],
4314
            ml_seed: [3; 32],
4315
        }
4316
        .sign(Domain::Change, &ObjectRef([2; 32]), &r2);
4317
        f.backend
4318
            .put_new(
4319
                &legacy_key("sigs", &r2),
4320
                &Signature::encode_list(std::slice::from_ref(&s2)),
4321
            )
4322
            .unwrap();
4323
        f.backend
4324
            .put_new(
4325
                &format!("{}{}", fanout_prefix("seal", &r1), "ab".repeat(32)),
4326
                b"",
4327
            )
4328
            .unwrap();
4329
        f.backend
4330
            .put_new(
4331
                &legacy_key("receipts", &r2),
4332
                format!("{}\n", ObjectRef([7; 32]).to_hex()).as_bytes(),
4333
            )
4334
            .unwrap();
4335
        let check = |s: &Store| {
4336
            assert_eq!(s.signatures(&r1).unwrap(), vec![s1.clone()]);
4337
            assert_eq!(s.signatures(&r2).unwrap(), vec![s2.clone()]);
4338
            assert_eq!(s.seals_for(&r1).unwrap(), vec![ObjectRef([0xab; 32])]);
4339
            assert_eq!(s.receipts_for(&r2).unwrap(), vec![ObjectRef([7; 32])]);
4340
            assert_eq!(s.receipt_states().unwrap(), vec![r2]);
4341
        };
4342
        let reader = f.open();
4343
        assert!(!reader.folded.get());
4344
        check(&reader);
4345
        assert!(f.backend.inner.read_ref(JOURNAL_KEY).unwrap().is_none());
4346
4347
        assert!(fold(&w));
4348
        let reader = f.open();
4349
        assert!(reader.folded.get());
4350
        check(&reader);
4351
4352
        // The older lists' entries are per-object entries now, and the
4353
        // marker is there.
4354
        assert!(f.backend.inner.has(LEGACY_FOLDED_KEY).unwrap());
4355
        assert!(f
4356
            .backend
4357
            .inner
4358
            .has(&format!("{}{}", fanout_prefix("sig", &r2), s2.key.to_hex()))
4359
            .unwrap());
4360
        assert!(f
4361
            .backend
4362
            .inner
4363
            .has(&format!(
4364
                "{}{}",
4365
                fanout_prefix("receipt", &r2),
4366
                ObjectRef([7; 32]).to_hex()
4367
            ))
4368
            .unwrap());
4369
4370
        // An older writer after the fold: missed until the next one.
4371
        let r3 = w.put(ObjectType::Chunk, b"three").unwrap();
4372
        let s3 = old_writer_signs(&f, &r3, 3);
4373
        assert!(f.open().signatures(&r3).unwrap().is_empty());
4374
        assert!(fold(&w));
4375
        assert_eq!(f.open().signatures(&r3).unwrap(), vec![s3]);
4376
    }
4377
4378
    /// Writers racing on the journal never lose a line: a lost
4379
    /// compare-and-swap reads the journal again and retries.
4380
    #[test]
4381
    fn concurrent_journal_appends_lose_nothing() {
4382
        let f = Fixture::new();
4383
        // One race, made to happen.
4384
        let w = f.open();
4385
        f.backend.interfere.store(true, Ordering::SeqCst);
4386
        w.index_seal(&ObjectRef([1; 32]), &ObjectRef([2; 32]))
4387
            .unwrap();
4388
        let j = journal(&w);
4389
        assert_eq!(j.len(), 2, "the competitor's line and ours: {j:?}");
4390
        // Many, from threads, each its own handle.
4391
        let backend: Arc<Backend> = f.backend.clone();
4392
        std::thread::scope(|scope| {
4393
            for t in 0..8u8 {
4394
                let backend = backend.clone();
4395
                let dir = f.dir.path().join(format!("t{t}"));
4396
                scope.spawn(move || {
4397
                    let s = Store::open_on(backend, dir).unwrap();
4398
                    for i in 0..25u8 {
4399
                        s.index_seal(&ObjectRef([t + 10; 32]), &ObjectRef([i; 32]))
4400
                            .unwrap();
4401
                    }
4402
                });
4403
            }
4404
        });
4405
        assert_eq!(journal(&w).len(), 2 + 8 * 25);
4406
        assert!(fold(&w));
4407
        let r = f.open();
4408
        for t in 0..8u8 {
4409
            assert_eq!(r.seals_for(&ObjectRef([t + 10; 32])).unwrap().len(), 25);
4410
        }
4411
    }
4412
4413
    /// A batch adds its entries to the journal in one update when it is
4414
    /// flushed; past the limit the writer folds them into a partial list.
4415
    #[test]
4416
    fn a_long_journal_is_folded_into_a_partial_list() {
4417
        let f = Fixture::new();
4418
        let w = f.open();
4419
        assert!(fold(&w), "folded, with nothing yet");
4420
        let key = ObjectRef([5; 32]);
4421
        f.backend.reset();
4422
        let batch = w.begin_batch().unwrap();
4423
        for i in 0..=JOURNAL_FOLD_AT as u32 {
4424
            let mut v = [0u8; 32];
4425
            v[..4].copy_from_slice(&i.to_le_bytes());
4426
            w.index_seal(&key, &ObjectRef(v)).unwrap();
4427
        }
4428
        assert_eq!(f.backend.n("write_ref"), 0, "nothing until the flush");
4429
        batch.finish().unwrap();
4430
        let parts = f
4431
            .backend
4432
            .inner
4433
            .list("pack-")
4434
            .unwrap()
4435
            .into_iter()
4436
            .filter(|e| e.key.ends_with(PART_SUFFIX))
4437
            .count();
4438
        assert_eq!(parts, 1);
4439
        assert!(journal(&w).values().all(|t| *t != 0), "all folded");
4440
        let r = f.open();
4441
        assert_eq!(r.seals_for(&key).unwrap().len(), JOURNAL_FOLD_AT + 1);
4442
        // The next complete fold takes the partial list in.
4443
        assert!(fold(&w));
4444
        let lists: Vec<String> = f.open().entry_lists.borrow().clone();
4445
        assert_eq!(lists.len(), 1);
4446
        assert!(lists[0].ends_with(LIST_SUFFIX));
4447
        assert_eq!(f.open().seals_for(&key).unwrap().len(), JOURNAL_FOLD_AT + 1);
4448
    }
4449
4450
    /// A collection that deletes an object takes its entries out of the
4451
    /// journal too, so the next fold leaves them out.
4452
    #[test]
4453
    fn pruned_objects_leave_the_journal() {
4454
        let f = Fixture::new();
4455
        let w = f.open();
4456
        let gone = w.put(ObjectType::Chunk, b"gone").unwrap();
4457
        let kept = w.put(ObjectType::Chunk, b"kept").unwrap();
4458
        sign(&w, &gone, 1);
4459
        let s = sign(&w, &kept, 2);
4460
        assert!(fold(&w));
4461
        let keep: std::collections::HashSet<ObjectRef> = w
4462
            .list()
4463
            .unwrap()
4464
            .into_iter()
4465
            .filter(|r| *r != gone)
4466
            .collect();
4467
        w.prune(&keep, std::time::Duration::ZERO).unwrap();
4468
        assert!(journal(&w).keys().all(|k| !k.contains(&gone.to_hex()[2..])));
4469
        fold(&w);
4470
        let r = f.open();
4471
        assert!(r.signatures(&gone).unwrap().is_empty());
4472
        assert_eq!(r.signatures(&kept).unwrap(), vec![s]);
4473
        let all = r.with_entries(|e| e.clone()).unwrap().unwrap();
4474
        assert!(
4475
            all.iter().all(|k| !k.contains(&gone.to_hex()[2..])),
4476
            "{all:?}"
4477
        );
4478
    }
4479
4480
    /// Requests per lookup, before and after the fold (#116): twenty
4481
    /// objects, five of them signed, each looked up for its signatures
4482
    /// and its seals, as a fetch does.
4483
    #[test]
4484
    fn a_folded_store_looks_up_entries_without_listing() {
4485
        let f = Fixture::new();
4486
        let w = f.open();
4487
        let refs: Vec<ObjectRef> = (0..20)
4488
            .map(|i| {
4489
                w.put(ObjectType::Chunk, format!("o {i}").as_bytes())
4490
                    .unwrap()
4491
            })
4492
            .collect();
4493
        for r in &refs[..5] {
4494
            sign(&w, r, 1);
4495
        }
4496
        let lookups = |s: &Store| {
4497
            for r in &refs {
4498
                s.signatures(r).unwrap();
4499
                s.seals_for(r).unwrap();
4500
            }
4501
        };
4502
        let before = f.open();
4503
        f.backend.reset();
4504
        lookups(&before);
4505
        let (before_total, before_list) = (f.backend.total(), f.backend.n("list"));
4506
        eprintln!("40 lookups before the fold: {}", f.backend.summary());
4507
4508
        assert!(fold(&w));
4509
        let after = f.open();
4510
        f.backend.reset();
4511
        lookups(&after);
4512
        eprintln!("40 lookups after the fold: {}", f.backend.summary());
4513
        assert_eq!(before_list, 40);
4514
        assert_eq!(f.backend.n("list"), 0);
4515
        assert_eq!(f.backend.n("get_legacy"), 0, "the older lists are folded");
4516
        assert_eq!(f.backend.n("read_journal"), 1);
4517
        assert!(
4518
            f.backend.total() < before_total / 10,
4519
            "{}",
4520
            f.backend.summary()
4521
        );
4522
    }
4523
4524
    /// Once a fold has copied the older lists into per-object entries and
4525
    /// written its marker, readers stop reading them, folded or not; the
4526
    /// answers are the same.
4527
    #[test]
4528
    fn the_legacy_marker_stops_reads_of_the_older_lists() {
4529
        let f = Fixture::new();
4530
        let w = f.open();
4531
        let r = w.put(ObjectType::Chunk, b"old").unwrap();
4532
        let sig = SecretKey {
4533
            ed_seed: [4; 32],
4534
            ml_seed: [5; 32],
4535
        }
4536
        .sign(Domain::Change, &ObjectRef([4; 32]), &r);
4537
        f.backend
4538
            .put_new(
4539
                &legacy_key("sigs", &r),
4540
                &Signature::encode_list(std::slice::from_ref(&sig)),
4541
            )
4542
            .unwrap();
4543
        for (legacy, v) in [("seals", 9u8), ("certs", 8), ("accepted-lists", 6)] {
4544
            f.backend
4545
                .put_new(
4546
                    &legacy_key(legacy, &r),
4547
                    format!("{}\n", ObjectRef([v; 32]).to_hex()).as_bytes(),
4548
                )
4549
                .unwrap();
4550
        }
4551
        let check = |s: &Store| {
4552
            assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()]);
4553
            assert_eq!(s.seals_for(&r).unwrap(), vec![ObjectRef([9; 32])]);
4554
            assert_eq!(
4555
                s.index_get("cert", "certs", &r).unwrap(),
4556
                vec![ObjectRef([8; 32])]
4557
            );
4558
            assert_eq!(
4559
                s.index_get_pub("accepted", &r).unwrap(),
4560
                vec![ObjectRef([6; 32])]
4561
            );
4562
        };
4563
        let before = f.open();
4564
        f.backend.reset();
4565
        check(&before);
4566
        assert_eq!(f.backend.n("get_legacy"), 4, "{}", f.backend.summary());
4567
4568
        assert!(fold(&w));
4569
        let folded = f.open();
4570
        f.backend.reset();
4571
        check(&folded);
4572
        assert_eq!(f.backend.n("get_legacy") + f.backend.n("list"), 0);
4573
4574
        // Not folded (its lists gone), but marked: per-object listings,
4575
        // still no older lists.
4576
        for e in f.backend.inner.list("pack-").unwrap() {
4577
            if is_entry_list(&e.key) {
4578
                f.backend.inner.delete(&e.key).unwrap();
4579
            }
4580
        }
4581
        let marked = f.open();
4582
        assert!(!marked.folded.get() && marked.legacy_folded.get());
4583
        f.backend.reset();
4584
        check(&marked);
4585
        assert_eq!(f.backend.n("get_legacy"), 0, "{}", f.backend.summary());
4586
        assert_eq!(f.backend.n("list"), 4);
4587
    }
4588
}