OXYGENHUB
◆ v0.4.6 oxygen / crates / oxy-core / src / store.rs
106.0 KiB · Rust · last changed The issuer countersigns when it accepted a certified update (keyless step 4c) 18 days ago source blame history raw
64818938 Andy McCall ↶ 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.
d02e545d Andy McCall ↶ 5
//!
64818938 Andy McCall ↶ 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
//!
d02e545d Andy McCall ↶ 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`.
64818938 Andy McCall ↶ 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.
ff0af1ee unknown 27
5b9c6ad4 Andy McCall ↶ 28
use crate::attest::Attest;
ff0af1ee unknown 29
use crate::hash::{hash_object, header, ObjectRef, ObjectType};
1983686f Andy McCall ↶ 30
use crate::identity::{Identity, Signature};
6f8aeb99 Andy McCall ↶ 31
use crate::objects::{Blob, Change, ObjectError, StateObject};
fc1a466a Andy McCall ↶ 32
use crate::receipt::Receipt;
64818938 Andy McCall ↶ 33
use crate::strata::{self, Error as BackendError, LocalStore, MemoryStore};
ff0af1ee unknown 34
use rand::RngCore;
d02e545d Andy McCall ↶ 35
use std::cell::RefCell;
ff0af1ee unknown 36
use std::fs;
64818938 Andy McCall ↶ 37
use std::io::{self, Write};
ff0af1ee unknown 38
use std::path::{Path, PathBuf};
64818938 Andy McCall ↶ 39
use std::sync::Arc;
ff0af1ee unknown 40
d02e545d Andy McCall ↶ 41
const PACK_MAGIC: &[u8; 8] = b"OXYPACK1";
42
const IDX_MAGIC: &[u8; 8] = b"OXYIDX01";
43
const IDX_ENTRY: usize = 32 + 8 + 4 + 4 + 1 + 1;
44
const FLAG_ZSTD: u8 = 1;
a17fc622 Andy McCall ↶ 45
/// Compressed with the pack's trained dictionary (`pack-ID.dict`), which a
46
/// full repack makes from a sample of the objects it packs: small files
47
/// share most of their vocabulary, which each one alone cannot use.
48
const FLAG_DICT: u8 = 2;
49
/// The size of a pack's dictionary, and the least a repack must pack for
50
/// one to pay.
51
const DICT_BYTES: usize = 110 << 10;
52
const DICT_MIN_OBJECTS: usize = 1000;
53
/// Objects up to this size are compressed with the dictionary.
54
const DICT_OBJECT_MAX: usize = 64 << 10;
d02e545d Andy McCall ↶ 55
const ZSTD_LEVEL: i32 = 3;
da652d63 Andy McCall ↶ 56
/// The level of a full repack (`oxy gc`): slower, once, for a smaller store.
57
pub const REPACK_ZSTD_LEVEL: i32 = 9;
2b872d07 Andy McCall ↶ 58
/// Largest object a pack entry may claim; anything bigger is corruption,
59
/// and refusing it keeps a bad index from driving huge allocations.
60
const MAX_OBJECT_LEN: u32 = 1 << 30;
d02e545d Andy McCall ↶ 61
afaba5f1 Andy McCall ↶ 62
/// An object read from the store: its type and body.
63
pub type Object = (ObjectType, Vec<u8>);
64
9ae4d15a Andy McCall ↶ 65
/// Where a long operation reports what it is doing (oxygen#66): a phase,
66
/// or a count within one, as a line for a spinner or a log.
67
pub type Progress<'a> = &'a mut dyn FnMut(&str);
64818938 Andy McCall ↶ 68
69
/// The storage under a store: objects by key, see [`crate::strata`].
70
pub type Backend = dyn strata::Store;
9ae4d15a Andy McCall ↶ 71
72
fn silent() -> impl FnMut(&str) {
73
    |_| {}
64818938 Andy McCall ↶ 74
}
75
76
/// A backend error as a store error: what is not found or already there
77
/// is handled where it matters; the rest is an I/O failure.
78
fn backend_err(e: BackendError) -> StoreError {
79
    match e {
80
        BackendError::Io(e) => StoreError::Io(e),
81
        other => StoreError::Io(io::Error::other(other.to_string())),
82
    }
83
}
84
85
/// `put_new` where an object already there is what we wanted.
86
fn put_new_or_existing(backend: &Backend, key: &str, bytes: &[u8]) -> Result<()> {
87
    match backend.put_new(key, bytes) {
88
        Ok(()) | Err(BackendError::Exists(_)) => Ok(()),
89
        Err(e) => Err(backend_err(e)),
90
    }
91
}
92
93
/// `get` as an option: `None` when there is no such key.
94
fn get_optional(backend: &Backend, key: &str) -> Result<Option<Vec<u8>>> {
95
    match backend.get(key) {
96
        Ok(b) => Ok(Some(b)),
97
        Err(BackendError::NotFound(_)) => Ok(None),
98
        Err(e) => Err(backend_err(e)),
99
    }
100
}
101
102
fn delete(backend: &Backend, key: &str) -> Result<()> {
103
    backend.delete(key).map_err(backend_err)
104
}
105
106
/// `loose/<hh>/<rest>`.
107
fn loose_key(r: &ObjectRef) -> String {
108
    let h = r.to_hex();
109
    format!("loose/{}/{}", &h[..2], &h[2..])
110
}
111
112
/// The object a loose key names, if it is one.
113
fn loose_ref(key: &str) -> Option<ObjectRef> {
114
    let rest = key.strip_prefix("loose/")?;
115
    let (hh, tail) = rest.split_once('/')?;
116
    if hh.len() != 2 || tail.contains('/') {
117
        return None;
118
    }
119
    ObjectRef::from_hex(&format!("{hh}{tail}"))
120
}
121
122
/// `<index>/<hh>/<rest>/` for the entries kept under an object.
123
fn fanout_prefix(index: &str, r: &ObjectRef) -> String {
124
    let h = r.to_hex();
125
    format!("{index}/{}/{}/", &h[..2], &h[2..])
126
}
127
128
/// `<index>/<hh>/<rest>`: the one-file-per-object layout of older stores.
129
fn legacy_key(index: &str, r: &ObjectRef) -> String {
130
    let h = r.to_hex();
131
    format!("{index}/{}/{}", &h[..2], &h[2..])
9ae4d15a Andy McCall ↶ 132
}
133
64818938 Andy McCall ↶ 134
fn now_ms() -> u64 {
135
    std::time::SystemTime::now()
136
        .duration_since(std::time::UNIX_EPOCH)
137
        .map(|d| d.as_millis() as u64)
138
        .unwrap_or(0)
139
}
140
d02e545d Andy McCall ↶ 141
#[derive(Clone, Copy, Debug)]
142
struct PackEntry {
143
    offset: u64,
144
    stored_len: u32,
145
    raw_len: u32,
146
    ty: ObjectType,
147
    flags: u8,
148
}
149
150
/// One pack's index, fully in memory and sorted by ObjectRef.
151
struct PackIndex {
64818938 Andy McCall ↶ 152
    /// The pack's key, `pack-<id>.pack`.
153
    key: String,
d02e545d Andy McCall ↶ 154
    refs: Vec<ObjectRef>,
155
    entries: Vec<PackEntry>,
a17fc622 Andy McCall ↶ 156
    /// The pack's dictionary, prepared once, when it has one.
64818938 Andy McCall ↶ 157
    dict: Option<Arc<zstd::dict::DecoderDictionary<'static>>>,
d02e545d Andy McCall ↶ 158
}
159
a17fc622 Andy McCall ↶ 160
impl std::fmt::Debug for PackIndex {
161
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
162
        f.debug_struct("PackIndex")
64818938 Andy McCall ↶ 163
            .field("key", &self.key)
a17fc622 Andy McCall ↶ 164
            .field("objects", &self.refs.len())
165
            .field("dict", &self.dict.is_some())
166
            .finish()
64818938 Andy McCall ↶ 167
    }
168
}
169
170
/// Decode a stored pack entry and check it against its reference.
171
fn decode_entry(
172
    r: &ObjectRef,
173
    e: PackEntry,
174
    stored: Vec<u8>,
175
    dict: Option<&Arc<zstd::dict::DecoderDictionary<'static>>>,
176
) -> Result<Object> {
177
    let body = if e.flags & FLAG_DICT != 0 {
178
        let dict = dict.ok_or_else(|| {
179
            StoreError::Corrupt(*r, "packed with a dictionary the pack lacks".into())
180
        })?;
181
        zstd::bulk::Decompressor::with_prepared_dictionary(dict)
182
            .and_then(|mut d| d.decompress(&stored, e.raw_len as usize))
183
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
184
    } else if e.flags & FLAG_ZSTD != 0 {
185
        zstd::bulk::decompress(&stored, e.raw_len as usize)
186
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
187
    } else {
188
        stored
189
    };
190
    if body.len() != e.raw_len as usize || hash_object(e.ty, &body) != *r {
191
        return Err(StoreError::Corrupt(
192
            *r,
193
            "packed object does not match its reference".into(),
194
        ));
a17fc622 Andy McCall ↶ 195
    }
64818938 Andy McCall ↶ 196
    Ok((e.ty, body))
a17fc622 Andy McCall ↶ 197
}
198
d02e545d Andy McCall ↶ 199
impl PackIndex {
64818938 Andy McCall ↶ 200
    fn load(backend: &Backend, idx_key: &str, pack_key: &str) -> Result<PackIndex> {
201
        let data = backend.get(idx_key).map_err(backend_err)?;
202
        let bad = |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{idx_key}: {m}"));
d02e545d Andy McCall ↶ 203
        if data.len() < 12 || &data[..8] != IDX_MAGIC {
204
            return Err(bad("bad index header"));
205
        }
206
        let count = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize;
207
        if data.len() != 12 + count * IDX_ENTRY {
208
            return Err(bad("index length mismatch"));
209
        }
64818938 Andy McCall ↶ 210
        let pack_len = backend
211
            .stat(pack_key)
212
            .map_err(backend_err)?
213
            .ok_or_else(|| bad("the index names a pack that is not there"))?
214
            .size;
d02e545d Andy McCall ↶ 215
        let mut refs = Vec::with_capacity(count);
216
        let mut entries = Vec::with_capacity(count);
217
        for i in 0..count {
218
            let e = &data[12 + i * IDX_ENTRY..12 + (i + 1) * IDX_ENTRY];
219
            let r = ObjectRef::from_slice(&e[..32]).unwrap();
220
            if let Some(prev) = refs.last() {
221
                if r <= *prev {
222
                    return Err(bad("index not sorted"));
223
                }
224
            }
225
            refs.push(r);
2b872d07 Andy McCall ↶ 226
            let entry = PackEntry {
d02e545d Andy McCall ↶ 227
                offset: u64::from_le_bytes(e[32..40].try_into().unwrap()),
228
                stored_len: u32::from_le_bytes(e[40..44].try_into().unwrap()),
229
                raw_len: u32::from_le_bytes(e[44..48].try_into().unwrap()),
75167672 Andy McCall 230
                ty: ObjectType::ALL
231
                    .get(e[48] as usize)
232
                    .copied()
233
                    .ok_or_else(|| bad("unknown object type"))?,
d02e545d Andy McCall ↶ 234
                flags: e[49],
2b872d07 Andy McCall ↶ 235
            };
a9f686ed Andy McCall ↶ 236
            if entry.stored_len > MAX_OBJECT_LEN || entry.raw_len > MAX_OBJECT_LEN {
237
                return Err(StoreError::Corrupt(
238
                    r,
239
                    format!(
64818938 Andy McCall ↶ 240
                        "{idx_key}: the index claims an object of {} bytes ({} stored), more than the {MAX_OBJECT_LEN} an object may be",
241
                        entry.raw_len, entry.stored_len
a9f686ed Andy McCall ↶ 242
                    ),
243
                ));
244
            }
245
            if entry.offset.saturating_add(entry.stored_len as u64) > pack_len {
2b872d07 Andy McCall ↶ 246
                return Err(bad("index entry points outside the pack"));
247
            }
248
            entries.push(entry);
d02e545d Andy McCall ↶ 249
        }
64818938 Andy McCall ↶ 250
        let dict_key = format!("{}.dict", pack_key.trim_end_matches(".pack"));
251
        let dict = get_optional(backend, &dict_key)?
252
            .map(|d| Arc::new(zstd::dict::DecoderDictionary::copy(&d)));
75167672 Andy McCall 253
        Ok(PackIndex {
64818938 Andy McCall ↶ 254
            key: pack_key.to_string(),
75167672 Andy McCall 255
            refs,
256
            entries,
a17fc622 Andy McCall ↶ 257
            dict,
75167672 Andy McCall 258
        })
d02e545d Andy McCall ↶ 259
    }
260
261
    fn find(&self, r: &ObjectRef) -> Option<PackEntry> {
262
        self.refs.binary_search(r).ok().map(|i| self.entries[i])
263
    }
264
64818938 Andy McCall ↶ 265
    /// Read one entry from the pack: one ranged read, then decode.
266
    fn fetch(&self, backend: &Backend, r: &ObjectRef, e: PackEntry) -> Result<Object> {
267
        let stored = backend
268
            .get_range(&self.key, e.offset..e.offset + e.stored_len as u64)
269
            .map_err(|err| match err {
270
                BackendError::NotFound(_) => StoreError::NotFound(*r),
271
                other => backend_err(other),
a17fc622 Andy McCall ↶ 272
            })?;
64818938 Andy McCall ↶ 273
        decode_entry(r, e, stored, self.dict.as_ref())
d02e545d Andy McCall ↶ 274
    }
275
}
276
277
fn type_index(ty: ObjectType) -> u8 {
278
    ObjectType::ALL.iter().position(|t| *t == ty).unwrap() as u8
279
}
280
281
/// What `pack_loose` did.
282
#[derive(Debug, Clone, Default, PartialEq, Eq)]
283
pub struct PackReport {
284
    pub objects: usize,
285
    pub raw_bytes: u64,
286
    pub packed_bytes: u64,
287
}
288
135212de Andy McCall ↶ 289
/// Packs tolerated before `merge_packs` folds them into one (spec §13.4).
290
pub const MAX_PACKS: usize = 16;
291
ba160f60 Andy McCall ↶ 292
/// What `prune` did.
293
#[derive(Debug, Clone, Default, PartialEq, Eq)]
294
pub struct PruneReport {
295
    pub objects: usize,
296
    pub bytes: u64,
297
    /// Unreachable objects left alone because they are younger than the grace period.
298
    pub deferred: usize,
299
}
300
ff0af1ee unknown 301
#[derive(Debug, thiserror::Error)]
302
pub enum StoreError {
303
    #[error("io: {0}")]
304
    Io(#[from] io::Error),
305
    #[error("corrupt object {0}: {1}")]
306
    Corrupt(ObjectRef, String),
307
    #[error("object {0}")]
308
    Object(#[from] ObjectError),
309
    #[error("object {0} is a {1}, expected {2}")]
310
    WrongType(ObjectRef, &'static str, &'static str),
311
    #[error("object {0} not found")]
312
    NotFound(ObjectRef),
6f8aeb99 Andy McCall ↶ 313
    #[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)")]
a9f686ed Andy McCall ↶ 314
    TooLarge {
315
        kind: &'static str,
316
        size: usize,
317
        limit: u32,
318
    },
ff0af1ee unknown 319
}
320
321
pub type Result<T> = std::result::Result<T, StoreError>;
322
323
pub struct Store {
64818938 Andy McCall ↶ 324
    backend: Arc<Backend>,
325
    /// Temporary files (a pack being streamed) and derived caches (`metas`).
326
    /// For a store in a directory it is that directory.
327
    scratch: PathBuf,
328
    /// The backend is that directory, so the old file paths still hold.
329
    local: bool,
d02e545d Andy McCall ↶ 330
    packs: RefCell<Vec<PackIndex>>,
8f10a07a Andy McCall ↶ 331
    /// An open write batch (spec §13.4): objects go here instead of to loose
332
    /// files, and are written out with one `fsync` when the batch finishes.
333
    pending: RefCell<Option<Pending>>,
334
    batch_depth: std::cell::Cell<u32>,
ce14f0d6 Andy McCall ↶ 335
    /// Change metadata already read, from the `metas` cache file.
336
    metas: RefCell<Option<crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta>>>,
8f10a07a Andy McCall ↶ 337
}
338
64818938 Andy McCall ↶ 339
impl std::fmt::Debug for Store {
340
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
341
        f.debug_struct("Store")
342
            .field("backend", &self.backend.describe())
343
            .field("scratch", &self.scratch)
344
            .field("packs", &self.packs.borrow().len())
345
            .finish()
346
    }
347
}
348
8f10a07a Andy McCall ↶ 349
/// Objects written since `begin_batch`. Small batches stay in memory and
350
/// become loose objects; once `SPILL_OBJECTS` or `SPILL_BYTES` is exceeded
351
/// the batch streams into a pack file instead.
352
#[derive(Debug)]
353
struct Pending {
354
    buffered: Vec<(ObjectType, Vec<u8>)>,
355
    buffered_bytes: usize,
356
    lookup: std::collections::HashMap<ObjectRef, PendingEntry>,
357
    pack: Option<PendingPack>,
358
    report: PackReport,
ff0af1ee unknown 359
}
360
8f10a07a Andy McCall ↶ 361
#[derive(Debug, Clone, Copy)]
362
enum PendingEntry {
363
    Buffered(usize),
364
    Packed(PackEntry),
365
}
366
367
#[derive(Debug)]
368
struct PendingPack {
369
    id: String,
370
    tmp_path: PathBuf,
371
    writer: io::BufWriter<fs::File>,
372
    offset: u64,
373
    index: Vec<(ObjectRef, PackEntry)>,
374
}
375
376
const SPILL_OBJECTS: usize = 256;
377
const SPILL_BYTES: usize = 8 << 20;
378
379
/// Finishes the enclosing batch when dropped; call `finish` to see errors.
380
pub struct Batch<'a> {
381
    store: &'a Store,
382
    done: bool,
383
}
384
385
impl Batch<'_> {
386
    pub fn finish(mut self) -> Result<()> {
387
        self.done = true;
388
        self.store.finish_batch()
389
    }
390
}
391
392
impl Drop for Batch<'_> {
393
    fn drop(&mut self) {
394
        if !self.done {
395
            let _ = self.store.finish_batch();
396
        }
397
    }
398
}
399
192fad1e Andy McCall ↶ 400
/// Append one stored object to a pending pack and index it.
401
fn write_pack_entry(
402
    pack: &mut PendingPack,
403
    r: &ObjectRef,
404
    ty: ObjectType,
405
    raw_len: usize,
406
    flags: u8,
407
    stored: &[u8],
408
    report: &mut PackReport,
409
) -> Result<PackEntry> {
410
    let header_len = 1 + 1 + 4 + 4 + 32;
411
    pack.writer.write_all(&[type_index(ty), flags])?;
412
    pack.writer
413
        .write_all(&(stored.len() as u32).to_le_bytes())?;
414
    pack.writer.write_all(&(raw_len as u32).to_le_bytes())?;
415
    pack.writer.write_all(&r.0)?;
416
    pack.writer.write_all(stored)?;
417
    let entry = PackEntry {
418
        offset: pack.offset + header_len,
419
        stored_len: stored.len() as u32,
420
        raw_len: raw_len as u32,
421
        ty,
422
        flags,
423
    };
424
    pack.offset += header_len + stored.len() as u64;
425
    pack.index.push((*r, entry));
426
    report.objects += 1;
427
    report.raw_bytes += raw_len as u64;
428
    report.packed_bytes += stored.len() as u64;
429
    Ok(entry)
64818938 Andy McCall ↶ 430
}
431
432
/// A random name for a temporary file or a pack.
433
fn fresh_id() -> String {
434
    let mut nonce = [0u8; 8];
435
    rand::rngs::OsRng.fill_bytes(&mut nonce);
436
    hex::encode(nonce)
437
}
438
439
/// Under `OXY_STORE_BACKEND=memory` every store opens on a store in
440
/// memory shared by directory, so the whole test suite can run without a
441
/// directory store, as it would over a bucket. Objects live for the
442
/// process.
443
fn memory_backend_for(dir: &Path) -> Option<Arc<Backend>> {
444
    use std::collections::HashMap;
445
    use std::sync::{Mutex, OnceLock};
446
    if std::env::var_os("OXY_STORE_BACKEND").as_deref() != Some(std::ffi::OsStr::new("memory")) {
447
        return None;
448
    }
449
    static STORES: OnceLock<Mutex<HashMap<PathBuf, Arc<MemoryStore>>>> = OnceLock::new();
450
    let key = dir.canonicalize().unwrap_or_else(|_| dir.to_path_buf());
451
    let mut map = STORES
452
        .get_or_init(Default::default)
453
        .lock()
454
        .unwrap_or_else(|p| p.into_inner());
455
    let store = map
456
        .entry(key)
457
        .or_insert_with(|| Arc::new(MemoryStore::new()))
458
        .clone();
459
    Some(store)
192fad1e Andy McCall ↶ 460
}
461
dd21fd82 Andy McCall ↶ 462
#[cfg(test)]
463
thread_local! {
464
    /// A backend every store opened on this thread uses instead of its
465
    /// directory: tests of a repository kept on a store that is not this
466
    /// disk's directory, with several handles on one store.
467
    pub(crate) static TEST_BACKEND: RefCell<Option<Arc<Backend>>> = const { RefCell::new(None) };
468
}
469
ff0af1ee unknown 470
impl Store {
64818938 Andy McCall ↶ 471
    /// Open the store in `dir`, which is also its scratch directory.
ff0af1ee unknown 472
    pub fn open(dir: impl Into<PathBuf>) -> io::Result<Store> {
473
        let dir = dir.into();
474
        fs::create_dir_all(dir.join("loose"))?;
dd21fd82 Andy McCall ↶ 475
        #[cfg(test)]
476
        if let Some(backend) = TEST_BACKEND.with(|t| t.borrow().clone()) {
477
            return Store::open_on(backend, dir);
478
        }
64818938 Andy McCall ↶ 479
        if let Some(backend) = memory_backend_for(&dir) {
480
            return Store::open_on(backend, dir);
481
        }
482
        let backend: Arc<Backend> =
483
            Arc::new(LocalStore::open(&dir).map_err(|e| io::Error::other(e.to_string()))?);
484
        let mut store = Store::open_on(backend, dir)?;
485
        store.local = true;
486
        Ok(store)
487
    }
488
489
    /// Open the store kept on `backend`, with `scratch` for what must be
490
    /// on this disk: a pack while it is being written, and caches.
491
    pub fn open_on(backend: Arc<Backend>, scratch: impl Into<PathBuf>) -> io::Result<Store> {
492
        let scratch = scratch.into();
493
        fs::create_dir_all(&scratch)?;
75167672 Andy McCall 494
        let store = Store {
64818938 Andy McCall ↶ 495
            backend,
496
            scratch,
497
            local: false,
75167672 Andy McCall 498
            packs: RefCell::new(Vec::new()),
499
            pending: RefCell::new(None),
500
            batch_depth: std::cell::Cell::new(0),
ce14f0d6 Andy McCall ↶ 501
            metas: RefCell::new(None),
75167672 Andy McCall 502
        };
503
        store
504
            .reload_packs()
505
            .map_err(|e| io::Error::other(e.to_string()))?;
d02e545d Andy McCall ↶ 506
        Ok(store)
64818938 Andy McCall ↶ 507
    }
508
509
    /// The storage under this store.
510
    pub fn backend(&self) -> &Arc<Backend> {
511
        &self.backend
d02e545d Andy McCall ↶ 512
    }
513
64818938 Andy McCall ↶ 514
    /// Whether the store is the directory `dir()` names, where every key
515
    /// is a file.
516
    pub fn is_local(&self) -> bool {
517
        self.local
518
    }
519
d02e545d Andy McCall ↶ 520
    fn reload_packs(&self) -> Result<()> {
64818938 Andy McCall ↶ 521
        let listed = self.backend.list("pack-").map_err(backend_err)?;
522
        let mut idx_keys: Vec<&str> = listed
523
            .iter()
524
            .map(|e| e.key.as_str())
525
            .filter(|k| k.ends_with(".idx"))
d02e545d Andy McCall ↶ 526
            .collect();
64818938 Andy McCall ↶ 527
        idx_keys.sort();
528
        let mut packs = Vec::new();
529
        for idx in idx_keys {
530
            let pack = format!("{}.pack", idx.trim_end_matches(".idx"));
531
            if listed.iter().any(|e| e.key == pack) {
532
                packs.push(PackIndex::load(self.backend.as_ref(), idx, &pack)?);
d02e545d Andy McCall ↶ 533
            }
534
        }
535
        *self.packs.borrow_mut() = packs;
536
        Ok(())
537
    }
538
539
    pub fn pack_count(&self) -> usize {
540
        self.packs.borrow().len()
541
    }
542
543
    fn find_packed(&self, r: &ObjectRef) -> Option<(usize, PackEntry)> {
75167672 Andy McCall 544
        self.packs
545
            .borrow()
546
            .iter()
547
            .enumerate()
548
            .find_map(|(i, p)| p.find(r).map(|e| (i, e)))
d02e545d Andy McCall ↶ 549
    }
550
64818938 Andy McCall ↶ 551
    /// Loose objects, with their keys and ages.
552
    fn loose_objects(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> {
553
        let mut out: Vec<(ObjectRef, strata::Entry)> = self
554
            .backend
555
            .list("loose/")
556
            .map_err(backend_err)?
557
            .into_iter()
558
            .filter_map(|e| loose_ref(&e.key).map(|r| (r, e)))
559
            .collect();
560
        out.sort_by_key(|(r, _)| *r);
d02e545d Andy McCall ↶ 561
        Ok(out)
562
    }
563
564
    pub fn loose_count(&self) -> io::Result<usize> {
64818938 Andy McCall ↶ 565
        self.loose_objects()
566
            .map(|l| l.len())
567
            .map_err(|e| io::Error::other(e.to_string()))
d02e545d Andy McCall ↶ 568
    }
569
64818938 Andy McCall ↶ 570
    /// Move every loose object into a new pack (spec §13.4). Loose objects
571
    /// are deleted only after the pack and its index are durable.
d02e545d Andy McCall ↶ 572
    pub fn pack_loose(&self) -> Result<PackReport> {
573
        let loose = self.loose_objects()?;
75167672 Andy McCall 574
        self.write_pack(
575
            &loose.iter().map(|(r, _)| *r).collect::<Vec<_>>(),
64818938 Andy McCall ↶ 576
            &loose.iter().map(|(_, e)| e.key.clone()).collect::<Vec<_>>(),
75167672 Andy McCall 577
            &[],
da652d63 Andy McCall ↶ 578
            ZSTD_LEVEL,
9ae4d15a Andy McCall ↶ 579
            &mut silent(),
75167672 Andy McCall 580
        )
da652d63 Andy McCall ↶ 581
    }
582
583
    /// Fold every pack and loose object into one pack compressed at
584
    /// `level` (spec §13.4): what an explicit `oxy gc` does, as `git gc`
585
    /// repacks. Returns the report and how many packs were folded in.
586
    pub fn repack_all(&self, level: i32) -> Result<(PackReport, usize)> {
9ae4d15a Andy McCall ↶ 587
        self.repack_all_with(level, &mut silent())
588
    }
589
590
    /// [`Store::repack_all`], reporting progress.
591
    pub fn repack_all_with(&self, level: i32, progress: Progress) -> Result<(PackReport, usize)> {
da652d63 Andy McCall ↶ 592
        let loose = self.loose_objects()?;
a67914bc Andy McCall ↶ 593
        let (mut refs, keys, trained): (Vec<ObjectRef>, Vec<String>, bool) = {
da652d63 Andy McCall ↶ 594
            let packs = self.packs.borrow();
595
            (
596
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
64818938 Andy McCall ↶ 597
                packs.iter().map(|p| p.key.clone()).collect(),
a67914bc Andy McCall ↶ 598
                packs.first().is_some_and(|p| p.dict.is_some()),
da652d63 Andy McCall ↶ 599
            )
600
        };
a67914bc Andy McCall ↶ 601
        // Nothing loose and one pack already made by a full repack (only
602
        // such a pack has a trained dictionary): rewriting it would give
603
        // the same pack. A hub's nightly gc rewrote andymccall/oxygen's
604
        // 48 000 objects (37 s under its write lock) every time.
605
        if loose.is_empty() && keys.len() <= 1 && (level == ZSTD_LEVEL || trained) {
da652d63 Andy McCall ↶ 606
            return Ok((PackReport::default(), 0));
607
        }
608
        refs.extend(loose.iter().map(|(r, _)| *r));
64818938 Andy McCall ↶ 609
        let loose_keys: Vec<String> = loose.into_iter().map(|(_, e)| e.key).collect();
610
        let n = keys.len();
611
        let report = self.write_pack(&refs, &loose_keys, &keys, level, progress)?;
da652d63 Andy McCall ↶ 612
        Ok((report, n))
ba160f60 Andy McCall ↶ 613
    }
614
135212de Andy McCall ↶ 615
    /// Fold every pack into one when there are more than `MAX_PACKS`
616
    /// (spec §13.4). Returns how many packs were merged.
617
    pub fn merge_packs(&self) -> Result<usize> {
64818938 Andy McCall ↶ 618
        let (refs, keys): (Vec<ObjectRef>, Vec<String>) = {
135212de Andy McCall ↶ 619
            let packs = self.packs.borrow();
620
            if packs.len() <= MAX_PACKS {
621
                return Ok(0);
622
            }
75167672 Andy McCall 623
            (
624
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
64818938 Andy McCall ↶ 625
                packs.iter().map(|p| p.key.clone()).collect(),
75167672 Andy McCall 626
            )
135212de Andy McCall ↶ 627
        };
64818938 Andy McCall ↶ 628
        let n = keys.len();
629
        self.write_pack(&refs, &[], &keys, ZSTD_LEVEL, &mut silent())?;
135212de Andy McCall ↶ 630
        Ok(n)
631
    }
632
ba160f60 Andy McCall ↶ 633
    /// Delete every object not in `keep` (spec §21). Loose objects and packs
634
    /// younger than `grace` are left alone, since a concurrent writer may not
635
    /// have recorded its references yet. Packs holding unreachable objects
64818938 Andy McCall ↶ 636
    /// are rewritten without them. Signatures, seal and receipt entries of
637
    /// deleted objects go too.
75167672 Andy McCall 638
    pub fn prune(
9ae4d15a Andy McCall ↶ 639
        &self,
640
        keep: &std::collections::HashSet<ObjectRef>,
641
        grace: std::time::Duration,
642
    ) -> Result<PruneReport> {
643
        self.prune_with(keep, grace, &mut silent())
644
    }
645
646
    /// [`Store::prune`], reporting progress.
647
    pub fn prune_with(
75167672 Andy McCall 648
        &self,
649
        keep: &std::collections::HashSet<ObjectRef>,
650
        grace: std::time::Duration,
9ae4d15a Andy McCall ↶ 651
        progress: Progress,
75167672 Andy McCall 652
    ) -> Result<PruneReport> {
9ae4d15a Andy McCall ↶ 653
        progress("pruning loose objects");
64818938 Andy McCall ↶ 654
        let now = now_ms();
655
        let grace_ms = grace.as_millis() as u64;
656
        // An object whose age the store does not know is old enough.
657
        let old_enough =
658
            |modified: Option<u64>| modified.is_none_or(|t| now.saturating_sub(t) >= grace_ms);
ba160f60 Andy McCall ↶ 659
        let mut report = PruneReport::default();
660
        // Loose objects.
64818938 Andy McCall ↶ 661
        for (r, entry) in self.loose_objects()? {
ba160f60 Andy McCall ↶ 662
            if keep.contains(&r) {
663
                continue;
664
            }
64818938 Andy McCall ↶ 665
            if !old_enough(entry.modified_ms) {
ba160f60 Andy McCall ↶ 666
                report.deferred += 1;
667
                continue;
668
            }
64818938 Andy McCall ↶ 669
            report.bytes += entry.size;
ba160f60 Andy McCall ↶ 670
            report.objects += 1;
64818938 Andy McCall ↶ 671
            delete(self.backend.as_ref(), &entry.key)?;
ba160f60 Andy McCall ↶ 672
            self.remove_sidecars(&r)?;
673
        }
674
        // Packs: rewrite those that hold unreachable objects.
675
        let mut keep_refs: Vec<ObjectRef> = Vec::new();
64818938 Andy McCall ↶ 676
        let mut old_keys: Vec<String> = Vec::new();
ba160f60 Andy McCall ↶ 677
        let mut rewrite = false;
9ae4d15a Andy McCall ↶ 678
        progress("pruning packs");
ba160f60 Andy McCall ↶ 679
        for p in self.packs.borrow().iter() {
680
            let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count();
681
            if unreachable == 0 {
682
                continue;
683
            }
64818938 Andy McCall ↶ 684
            let modified = self
685
                .backend
686
                .stat(&p.key)
687
                .map_err(backend_err)?
688
                .and_then(|e| e.modified_ms);
689
            if !old_enough(modified) {
ba160f60 Andy McCall ↶ 690
                report.deferred += unreachable;
691
                continue;
692
            }
693
            rewrite = true;
64818938 Andy McCall ↶ 694
            old_keys.push(p.key.clone());
ba160f60 Andy McCall ↶ 695
            for (r, e) in p.refs.iter().zip(p.entries.iter()) {
696
                if keep.contains(r) {
697
                    keep_refs.push(*r);
698
                } else {
699
                    report.objects += 1;
700
                    report.bytes += e.stored_len as u64;
701
                }
702
            }
703
        }
704
        if rewrite {
705
            let dropped: Vec<ObjectRef> = {
706
                let packs = self.packs.borrow();
75167672 Andy McCall 707
                packs
708
                    .iter()
64818938 Andy McCall ↶ 709
                    .filter(|p| old_keys.contains(&p.key))
75167672 Andy McCall 710
                    .flat_map(|p| p.refs.iter().copied())
711
                    .filter(|r| !keep.contains(r))
712
                    .collect()
ba160f60 Andy McCall ↶ 713
            };
64818938 Andy McCall ↶ 714
            self.write_pack(&keep_refs, &[], &old_keys, ZSTD_LEVEL, progress)?;
ba160f60 Andy McCall ↶ 715
            for r in dropped {
716
                self.remove_sidecars(&r)?;
717
            }
718
        }
719
        Ok(report)
720
    }
721
64818938 Andy McCall ↶ 722
    /// Everything kept beside an object: its signatures and the seal and
723
    /// receipt entries under it, in both layouts.
ba160f60 Andy McCall ↶ 724
    fn remove_sidecars(&self, r: &ObjectRef) -> Result<()> {
64818938 Andy McCall ↶ 725
        for index in ["sigs", "seals", "receipts"] {
726
            delete(self.backend.as_ref(), &legacy_key(index, r))?;
727
        }
728
        for index in ["sig", "seal", "receipt"] {
729
            for e in self
730
                .backend
731
                .list(&fanout_prefix(index, r))
732
                .map_err(backend_err)?
733
            {
734
                delete(self.backend.as_ref(), &e.key)?;
ba160f60 Andy McCall ↶ 735
            }
736
        }
737
        Ok(())
738
    }
739
64818938 Andy McCall ↶ 740
    /// Put a complete, synced temporary pack file and its index into the
741
    /// store; `index` must be sorted by ObjectRef. The pack goes first: an
742
    /// index never names a pack that is not there.
75167672 Andy McCall 743
    fn finish_pack_files(
744
        &self,
745
        id: &str,
746
        tmp_pack: &Path,
747
        index: &[(ObjectRef, PackEntry)],
748
    ) -> Result<()> {
64818938 Andy McCall ↶ 749
        let mut idx = Vec::with_capacity(12 + index.len() * IDX_ENTRY);
750
        idx.extend_from_slice(IDX_MAGIC);
751
        idx.extend_from_slice(&(index.len() as u32).to_le_bytes());
752
        for (r, e) in index {
753
            idx.extend_from_slice(&r.0);
754
            idx.extend_from_slice(&e.offset.to_le_bytes());
755
            idx.extend_from_slice(&e.stored_len.to_le_bytes());
756
            idx.extend_from_slice(&e.raw_len.to_le_bytes());
757
            idx.extend_from_slice(&[type_index(e.ty), e.flags]);
8f10a07a Andy McCall ↶ 758
        }
64818938 Andy McCall ↶ 759
        let put = self
760
            .backend
761
            .put_new_file(&format!("pack-{id}.pack"), tmp_pack)
762
            .map_err(backend_err);
763
        // A store on another disk copied it; the one in this directory
764
        // moved it and this finds nothing.
765
        let _ = fs::remove_file(tmp_pack);
766
        put?;
767
        self.backend
768
            .put_new(&format!("pack-{id}.idx"), &idx)
769
            .map_err(backend_err)?;
8f10a07a Andy McCall ↶ 770
        self.reload_packs()
64818938 Andy McCall ↶ 771
    }
772
773
    /// Remove a pack with its index and dictionary.
774
    fn remove_pack(&self, pack_key: &str) -> Result<()> {
775
        let stem = pack_key.trim_end_matches(".pack");
776
        delete(self.backend.as_ref(), pack_key)?;
777
        delete(self.backend.as_ref(), &format!("{stem}.idx"))?;
778
        delete(self.backend.as_ref(), &format!("{stem}.dict"))
8f10a07a Andy McCall ↶ 779
    }
780
64818938 Andy McCall ↶ 781
    /// Write `refs` into a new pack, then delete `loose_keys` and `old_packs`.
75167672 Andy McCall 782
    fn write_pack(
783
        &self,
784
        refs: &[ObjectRef],
64818938 Andy McCall ↶ 785
        loose_keys: &[String],
786
        old_packs: &[String],
da652d63 Andy McCall ↶ 787
        level: i32,
9ae4d15a Andy McCall ↶ 788
        progress: Progress,
75167672 Andy McCall 789
    ) -> Result<PackReport> {
ba160f60 Andy McCall ↶ 790
        if refs.is_empty() {
791
            for p in old_packs {
64818938 Andy McCall ↶ 792
                self.remove_pack(p)?;
ba160f60 Andy McCall ↶ 793
            }
794
            self.reload_packs()?;
d02e545d Andy McCall ↶ 795
            return Ok(PackReport::default());
796
        }
ba160f60 Andy McCall ↶ 797
        let mut refs: Vec<ObjectRef> = refs.to_vec();
798
        refs.sort();
799
        refs.dedup();
64818938 Andy McCall ↶ 800
        let id = fresh_id();
801
        let tmp_pack = self.scratch.join(format!(".tmp-{id}.pack"));
d02e545d Andy McCall ↶ 802
        let mut report = PackReport::default();
ba160f60 Andy McCall ↶ 803
        let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len());
a17fc622 Andy McCall ↶ 804
        // A full repack of enough objects trains a dictionary for them.
805
        let dict = if level == REPACK_ZSTD_LEVEL && refs.len() >= DICT_MIN_OBJECTS {
806
            let _t = crate::timing::phase("pack: train a dictionary");
9ae4d15a Andy McCall ↶ 807
            progress(&format!("training a dictionary on {} objects", refs.len()));
a17fc622 Andy McCall ↶ 808
            self.train_dictionary(&refs)
809
        } else {
810
            None
811
        };
812
        let encoder = dict
813
            .as_ref()
814
            .map(|d| zstd::dict::EncoderDictionary::copy(d, level));
64818938 Andy McCall ↶ 815
        let nocompress = self.nocompress_set()?;
d02e545d Andy McCall ↶ 816
        {
817
            let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?);
818
            f.write_all(PACK_MAGIC)?;
819
            let mut offset = PACK_MAGIC.len() as u64;
da652d63 Andy McCall ↶ 820
            // Read a batch (the store is not shared between threads),
821
            // compress it on every core, write it in order.
822
            let mut batches = refs.chunks(1024);
823
            let mut compressed_batch = Vec::new();
9ae4d15a Andy McCall ↶ 824
            let total = refs.len();
825
            let mut reported = usize::MAX;
da652d63 Andy McCall ↶ 826
            loop {
827
                if compressed_batch.is_empty() {
828
                    let Some(batch) = batches.next() else { break };
9ae4d15a Andy McCall ↶ 829
                    // Every tenth, and the first batch.
830
                    let tenth = report.objects * 10 / total.max(1);
831
                    if tenth != reported {
832
                        reported = tenth;
833
                        progress(&format!("packing {} of {total} objects", report.objects));
834
                    }
da652d63 Andy McCall ↶ 835
                    let mut read: Vec<(ObjectRef, ObjectType, Vec<u8>, bool)> =
836
                        Vec::with_capacity(batch.len());
837
                    for r in batch {
838
                        let (ty, body) = self.get(r)?;
64818938 Andy McCall ↶ 839
                        read.push((*r, ty, body, nocompress.contains(r)));
75167672 Andy McCall 840
                    }
da652d63 Andy McCall ↶ 841
                    use rayon::prelude::*;
842
                    compressed_batch = read
843
                        .into_par_iter()
a17fc622 Andy McCall ↶ 844
                        .map_init(
845
                            || {
846
                                encoder.as_ref().and_then(|e| {
847
                                    zstd::bulk::Compressor::with_prepared_dictionary(e).ok()
848
                                })
849
                            },
850
                            |with_dict, (r, ty, body, raw)| {
851
                                let packed = if raw {
852
                                    None
853
                                } else {
854
                                    // Small objects with the dictionary;
855
                                    // a large one has context enough.
856
                                    let dicted = with_dict
857
                                        .as_mut()
858
                                        .filter(|_| body.len() <= DICT_OBJECT_MAX)
859
                                        .and_then(|c| c.compress(&body).ok())
860
                                        .map(|c| (FLAG_ZSTD | FLAG_DICT, c));
861
                                    dicted
862
                                        .or_else(|| {
863
                                            zstd::bulk::compress(&body, level)
864
                                                .ok()
865
                                                .map(|c| (FLAG_ZSTD, c))
866
                                        })
867
                                        .filter(|(_, c)| c.len() < body.len())
868
                                };
869
                                (r, ty, body, packed)
870
                            },
871
                        )
da652d63 Andy McCall ↶ 872
                        .collect::<Vec<_>>();
873
                    compressed_batch.reverse();
874
                }
875
                let Some((r, ty, body, packed)) = compressed_batch.pop() else {
876
                    continue;
877
                };
878
                let r = &r;
879
                let (flags, stored) = match packed {
a17fc622 Andy McCall ↶ 880
                    Some(c) => c,
da652d63 Andy McCall ↶ 881
                    None => (0, body.clone()),
cdb5beba Andy McCall ↶ 882
                };
d02e545d Andy McCall ↶ 883
                let header_len = 1 + 1 + 4 + 4 + 32;
884
                f.write_all(&[type_index(ty), flags])?;
885
                f.write_all(&(stored.len() as u32).to_le_bytes())?;
886
                f.write_all(&(body.len() as u32).to_le_bytes())?;
887
                f.write_all(&r.0)?;
888
                f.write_all(&stored)?;
75167672 Andy McCall 889
                index.push((
890
                    *r,
891
                    PackEntry {
892
                        offset: offset + header_len,
893
                        stored_len: stored.len() as u32,
894
                        raw_len: body.len() as u32,
895
                        ty,
896
                        flags,
897
                    },
898
                ));
d02e545d Andy McCall ↶ 899
                offset += header_len + stored.len() as u64;
900
                report.objects += 1;
901
                report.raw_bytes += body.len() as u64;
902
                report.packed_bytes += stored.len() as u64;
903
            }
904
            f.flush()?;
905
            f.get_ref().sync_all()?;
a17fc622 Andy McCall ↶ 906
        }
907
        // The dictionary before the pack that needs it becomes visible.
908
        if let Some(d) = &dict {
64818938 Andy McCall ↶ 909
            self.backend
910
                .put_new(&format!("pack-{id}.dict"), d)
911
                .map_err(backend_err)?;
d02e545d Andy McCall ↶ 912
        }
9ae4d15a Andy McCall ↶ 913
        progress(&format!("writing the index of {} objects", index.len()));
8f10a07a Andy McCall ↶ 914
        self.finish_pack_files(&id, &tmp_pack, &index)?;
ba160f60 Andy McCall ↶ 915
        for p in old_packs {
64818938 Andy McCall ↶ 916
            self.remove_pack(p)?;
ba160f60 Andy McCall ↶ 917
        }
d02e545d Andy McCall ↶ 918
        self.reload_packs()?;
64818938 Andy McCall ↶ 919
        for key in loose_keys {
920
            delete(self.backend.as_ref(), key)?;
d02e545d Andy McCall ↶ 921
        }
922
        Ok(report)
ff0af1ee unknown 923
    }
924
a17fc622 Andy McCall ↶ 925
    /// A dictionary for these objects, trained on a sample of them (at
926
    /// most 10 000, 32 MB); `None` when training fails.
927
    fn train_dictionary(&self, refs: &[ObjectRef]) -> Option<Vec<u8>> {
928
        let step = (refs.len() / 10_000).max(1);
929
        let mut samples: Vec<Vec<u8>> = Vec::new();
930
        let mut total = 0usize;
931
        for r in refs.iter().step_by(step) {
932
            if total > 32 << 20 {
933
                break;
934
            }
935
            let Ok((ty, body)) = self.get(r) else {
936
                continue;
937
            };
938
            if matches!(
939
                ty,
940
                ObjectType::Chunk | ObjectType::Blob | ObjectType::Change
941
            ) && !body.is_empty()
942
                && body.len() < 1 << 20
943
            {
944
                total += body.len();
945
                samples.push(body);
946
            }
947
        }
948
        zstd::dict::from_samples(&samples, DICT_BYTES).ok()
949
    }
950
64818938 Andy McCall ↶ 951
    /// The scratch directory: for a store in a directory, that directory.
ff0af1ee unknown 952
    pub fn dir(&self) -> &Path {
64818938 Andy McCall ↶ 953
        &self.scratch
ff0af1ee unknown 954
    }
955
64818938 Andy McCall ↶ 956
    /// Where a loose object is kept, for a store in a directory (tests
957
    /// tamper with and remove objects there).
958
    #[cfg(test)]
387335ff Andy McCall ↶ 959
    pub(crate) fn path_for(&self, r: &ObjectRef) -> PathBuf {
64818938 Andy McCall ↶ 960
        self.scratch.join(loose_key(r))
961
    }
962
963
    /// Remove a loose object, whatever the store is on (tests make an
964
    /// object go missing).
965
    #[cfg(test)]
966
    pub(crate) fn remove_loose_for_test(&self, r: &ObjectRef) -> Result<()> {
967
        delete(self.backend.as_ref(), &loose_key(r))
ff0af1ee unknown 968
    }
969
970
    pub fn has(&self, r: &ObjectRef) -> bool {
64818938 Andy McCall ↶ 971
        self.backend.has(&loose_key(r)).unwrap_or(false)
75167672 Andy McCall 972
            || self.find_packed(r).is_some()
973
            || self
974
                .pending
975
                .borrow()
976
                .as_ref()
977
                .is_some_and(|p| p.lookup.contains_key(r))
ff0af1ee unknown 978
    }
979
8f10a07a Andy McCall ↶ 980
    /// Start a write batch: until the matching `finish`, `put` collects
981
    /// objects and writes them out together, with one `fsync` for a large
982
    /// batch (a pack) instead of one per object. Batches nest; only the
983
    /// outermost does the work. Objects are readable while pending.
984
    pub fn begin_batch(&self) -> Result<Batch<'_>> {
985
        let depth = self.batch_depth.get();
986
        if depth == 0 {
75167672 Andy McCall 987
            *self.pending.borrow_mut() = Some(Pending {
988
                buffered: vec![],
989
                buffered_bytes: 0,
990
                lookup: Default::default(),
991
                pack: None,
992
                report: PackReport::default(),
993
            });
8f10a07a Andy McCall ↶ 994
        }
995
        self.batch_depth.set(depth + 1);
75167672 Andy McCall 996
        Ok(Batch {
997
            store: self,
998
            done: false,
999
        })
8f10a07a Andy McCall ↶ 1000
    }
1001
1002
    fn finish_batch(&self) -> Result<()> {
1003
        let depth = self.batch_depth.get();
1004
        if depth == 0 {
1005
            return Ok(());
1006
        }
1007
        self.batch_depth.set(depth - 1);
1008
        if depth > 1 {
1009
            return Ok(());
1010
        }
75167672 Andy McCall 1011
        let Some(pending) = self.pending.borrow_mut().take() else {
1012
            return Ok(());
1013
        };
4dcb8c09 Andy McCall ↶ 1014
        self.write_pending(pending)
1015
    }
1016
1017
    /// Make every object written so far durable, even inside an open
1018
    /// batch, which stays open. Call before a reference that names them
1019
    /// is moved (spec §13.6): the operation head, a bookmark pointer.
1020
    pub fn flush(&self) -> Result<()> {
1021
        let Some(pending) = self.pending.borrow_mut().take() else {
1022
            return Ok(());
1023
        };
1024
        self.write_pending(pending)?;
1025
        *self.pending.borrow_mut() = Some(Pending {
1026
            buffered: vec![],
1027
            buffered_bytes: 0,
1028
            lookup: Default::default(),
1029
            pack: None,
1030
            report: PackReport::default(),
1031
        });
1032
        Ok(())
1033
    }
1034
1035
    fn write_pending(&self, pending: Pending) -> Result<()> {
8f10a07a Andy McCall ↶ 1036
        match pending.pack {
1037
            None => {
b30a880f Andy McCall ↶ 1038
                let objects: Vec<(ObjectType, &[u8])> = pending
1039
                    .buffered
1040
                    .iter()
1041
                    .map(|(ty, body)| (*ty, body.as_slice()))
1042
                    .collect();
1043
                self.put_loose_many(&objects)
8f10a07a Andy McCall ↶ 1044
            }
1045
            Some(mut pack) => {
1046
                pack.writer.flush()?;
1047
                pack.writer.get_ref().sync_all()?;
1048
                drop(pack.writer);
1049
                pack.index.sort_by_key(|(r, _)| *r);
1050
                self.finish_pack_files(&pack.id, &pack.tmp_path, &pack.index)?;
1051
                Ok(())
1052
            }
1053
        }
1054
    }
1055
1056
    /// Whether a batch is open and holds `r`.
64818938 Andy McCall ↶ 1057
    fn pending_get(&self, r: &ObjectRef) -> Option<Result<Object>> {
8f10a07a Andy McCall ↶ 1058
        let mut guard = self.pending.borrow_mut();
1059
        let pending = guard.as_mut()?;
1060
        let entry = *pending.lookup.get(r)?;
1061
        Some(match entry {
1062
            PendingEntry::Buffered(i) => {
1063
                let (ty, body) = &pending.buffered[i];
1064
                Ok((*ty, body.clone()))
1065
            }
1066
            PendingEntry::Packed(e) => {
75167672 Andy McCall 1067
                let Some(pack) = pending.pack.as_mut() else {
1068
                    return Some(Err(StoreError::NotFound(*r)));
1069
                };
64818938 Andy McCall ↶ 1070
                // Read from the pack file being written, on this disk.
1071
                (|| {
1072
                    pack.writer.flush()?;
1073
                    use std::io::{Read, Seek, SeekFrom};
1074
                    let mut f = fs::File::open(&pack.tmp_path)?;
1075
                    f.seek(SeekFrom::Start(e.offset))?;
1076
                    let mut stored = vec![0u8; e.stored_len as usize];
1077
                    f.read_exact(&mut stored)?;
1078
                    decode_entry(r, e, stored, None)
1079
                })()
8f10a07a Andy McCall ↶ 1080
            }
1081
        })
1082
    }
1083
1084
    /// Append one object to the pending pack file.
75167672 Andy McCall 1085
    fn pending_pack_write(
1086
        &self,
1087
        pack: &mut PendingPack,
1088
        r: &ObjectRef,
1089
        ty: ObjectType,
1090
        body: &[u8],
1091
        report: &mut PackReport,
1092
    ) -> Result<PackEntry> {
8f10a07a Andy McCall ↶ 1093
        let (flags, stored) = if self.is_nocompress(r) {
1094
            (0, body.to_vec())
1095
        } else {
75167672 Andy McCall 1096
            let compressed = zstd::bulk::compress(body, ZSTD_LEVEL)
1097
                .map_err(|e| StoreError::Corrupt(*r, format!("zstd: {e}")))?;
1098
            if compressed.len() < body.len() {
1099
                (FLAG_ZSTD, compressed)
1100
            } else {
1101
                (0, body.to_vec())
1102
            }
8f10a07a Andy McCall ↶ 1103
        };
192fad1e Andy McCall ↶ 1104
        write_pack_entry(pack, r, ty, body.len(), flags, &stored, report)
8f10a07a Andy McCall ↶ 1105
    }
1106
1107
    fn put_pending(&self, r: ObjectRef, ty: ObjectType, body: &[u8]) -> Result<()> {
1108
        let mut guard = self.pending.borrow_mut();
1109
        let pending = guard.as_mut().expect("batch open");
1110
        if pending.lookup.contains_key(&r) {
1111
            return Ok(());
1112
        }
75167672 Andy McCall 1113
        if pending.pack.is_none()
1114
            && pending.buffered.len() < SPILL_OBJECTS
1115
            && pending.buffered_bytes + body.len() <= SPILL_BYTES
1116
        {
8f10a07a Andy McCall ↶ 1117
            pending.buffered.push((ty, body.to_vec()));
1118
            pending.buffered_bytes += body.len();
75167672 Andy McCall 1119
            pending
1120
                .lookup
1121
                .insert(r, PendingEntry::Buffered(pending.buffered.len() - 1));
8f10a07a Andy McCall ↶ 1122
            return Ok(());
1123
        }
192fad1e Andy McCall ↶ 1124
        self.ensure_pack(pending)?;
1125
        let mut report = std::mem::take(&mut pending.report);
1126
        let pack = pending.pack.as_mut().unwrap();
1127
        let e = self.pending_pack_write(pack, &r, ty, body, &mut report)?;
1128
        pending.report = report;
1129
        pending.lookup.insert(r, PendingEntry::Packed(e));
1130
        Ok(())
1131
    }
1132
1133
    /// Open the pending batch's pack, moving the objects buffered so far
1134
    /// into it (a spill).
1135
    fn ensure_pack(&self, pending: &mut Pending) -> Result<()> {
8f10a07a Andy McCall ↶ 1136
        if pending.pack.is_none() {
1137
            // Spill: open the pack and move the buffered objects into it.
64818938 Andy McCall ↶ 1138
            let id = fresh_id();
1139
            let tmp_path = self.scratch.join(format!(".tmp-{id}.pack"));
8f10a07a Andy McCall ↶ 1140
            let mut writer = io::BufWriter::new(fs::File::create(&tmp_path)?);
1141
            writer.write_all(PACK_MAGIC)?;
75167672 Andy McCall 1142
            let mut pack = PendingPack {
1143
                id,
1144
                tmp_path,
1145
                writer,
1146
                offset: PACK_MAGIC.len() as u64,
1147
                index: vec![],
1148
            };
8f10a07a Andy McCall ↶ 1149
            let buffered = std::mem::take(&mut pending.buffered);
1150
            pending.buffered_bytes = 0;
1151
            let mut report = std::mem::take(&mut pending.report);
1152
            for (bty, bbody) in &buffered {
1153
                let br = hash_object(*bty, bbody);
1154
                let e = self.pending_pack_write(&mut pack, &br, *bty, bbody, &mut report)?;
1155
                pending.lookup.insert(br, PendingEntry::Packed(e));
1156
            }
1157
            pending.report = report;
1158
            pending.pack = Some(pack);
192fad1e Andy McCall ↶ 1159
        }
1160
        Ok(())
1161
    }
1162
1163
    /// Store many objects at once. Inside a large write batch they are
1164
    /// compressed on every core and appended in order; otherwise each is
1165
    /// stored as [`Store::put`] does.
1166
    pub fn put_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<Vec<ObjectRef>> {
1167
        use rayon::prelude::*;
1168
        let total: usize = objects.iter().map(|(_, b)| b.len()).sum();
1169
        if self.pending.borrow().is_none() || total < SPILL_BYTES {
1170
            return objects.iter().map(|(ty, b)| self.put(*ty, b)).collect();
1171
        }
1172
        for (ty, b) in objects {
1173
            if b.len() > MAX_OBJECT_LEN as usize {
1174
                return Err(StoreError::TooLarge {
1175
                    kind: ty.name(),
1176
                    size: b.len(),
1177
                    limit: MAX_OBJECT_LEN,
1178
                });
1179
            }
8f10a07a Andy McCall ↶ 1180
        }
192fad1e Andy McCall ↶ 1181
        let refs: Vec<ObjectRef> = objects
1182
            .par_iter()
1183
            .map(|(ty, b)| hash_object(*ty, b))
1184
            .collect();
1185
        // What is new: not stored already, not earlier in this call.
1186
        let mut seen: std::collections::HashSet<ObjectRef> = std::collections::HashSet::new();
1187
        let mut todo: Vec<usize> = Vec::new();
1188
        for (i, r) in refs.iter().enumerate() {
1189
            let known = self
1190
                .pending
1191
                .borrow()
1192
                .as_ref()
1193
                .is_some_and(|p| p.lookup.contains_key(r));
1194
            if seen.insert(*r)
1195
                && !known
1196
                && self.find_packed(r).is_none()
64818938 Andy McCall ↶ 1197
                && !self.backend.has(&loose_key(r)).unwrap_or(false)
192fad1e Andy McCall ↶ 1198
            {
1199
                todo.push(i);
1200
            }
1201
        }
64818938 Andy McCall ↶ 1202
        let raw = self.nocompress_set()?;
192fad1e Andy McCall ↶ 1203
        let packed: Vec<(u8, Vec<u8>)> = todo
1204
            .par_iter()
1205
            .map(|i| {
1206
                let body = objects[*i].1;
1207
                if raw.contains(&refs[*i]) {
1208
                    return (0, body.to_vec());
1209
                }
1210
                match zstd::bulk::compress(body, ZSTD_LEVEL) {
1211
                    Ok(c) if c.len() < body.len() => (FLAG_ZSTD, c),
1212
                    _ => (0, body.to_vec()),
1213
                }
1214
            })
1215
            .collect();
1216
        let mut guard = self.pending.borrow_mut();
1217
        let pending = guard.as_mut().expect("batch open");
1218
        self.ensure_pack(pending)?;
8f10a07a Andy McCall ↶ 1219
        let mut report = std::mem::take(&mut pending.report);
1220
        let pack = pending.pack.as_mut().unwrap();
192fad1e Andy McCall ↶ 1221
        for (i, (flags, stored)) in todo.iter().zip(packed) {
1222
            let (ty, body) = objects[*i];
1223
            let e = write_pack_entry(pack, &refs[*i], ty, body.len(), flags, &stored, &mut report)?;
1224
            pending.lookup.insert(refs[*i], PendingEntry::Packed(e));
1225
        }
8f10a07a Andy McCall ↶ 1226
        pending.report = report;
192fad1e Andy McCall ↶ 1227
        Ok(refs)
8f10a07a Andy McCall ↶ 1228
    }
1229
ff0af1ee unknown 1230
    /// Store `body` as an object of type `ty`; idempotent.
1231
    pub fn put(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
a9f686ed Andy McCall ↶ 1232
        // Refuse what the store could not read back (spec §13.2), before
1233
        // anything is written: a pack must never hold an entry its index
1234
        // would reject.
1235
        if body.len() > MAX_OBJECT_LEN as usize {
1236
            return Err(StoreError::TooLarge {
1237
                kind: ty.name(),
1238
                size: body.len(),
1239
                limit: MAX_OBJECT_LEN,
1240
            });
1241
        }
ff0af1ee unknown 1242
        let r = hash_object(ty, body);
64818938 Andy McCall ↶ 1243
        if self.find_packed(&r).is_some() || self.backend.has(&loose_key(&r)).unwrap_or(false) {
ff0af1ee unknown 1244
            return Ok(r);
1245
        }
8f10a07a Andy McCall ↶ 1246
        if self.pending.borrow().is_some() {
1247
            self.put_pending(r, ty, body)?;
1248
            return Ok(r);
1249
        }
1250
        self.put_loose(ty, body)?;
1251
        Ok(r)
1252
    }
1253
64818938 Andy McCall ↶ 1254
    /// Write one loose object, durable when this returns.
8f10a07a Andy McCall ↶ 1255
    fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
1256
        let r = hash_object(ty, body);
64818938 Andy McCall ↶ 1257
        let mut bytes = header(ty, body.len());
1258
        bytes.extend_from_slice(body);
4dcb8c09 Andy McCall ↶ 1259
        crate::durable::step("write loose object")?;
64818938 Andy McCall ↶ 1260
        put_new_or_existing(self.backend.as_ref(), &loose_key(&r), &bytes)?;
ff0af1ee unknown 1261
        Ok(r)
b30a880f Andy McCall ↶ 1262
    }
1263
1264
    /// Write several loose objects with as few waits on the disk as the
64818938 Andy McCall ↶ 1265
    /// guarantees allow (a directory store writes them all, then syncs
1266
    /// each directory once). Durable when this returns, as `put_loose`
1267
    /// one at a time would be (spec §13.6).
b30a880f Andy McCall ↶ 1268
    fn put_loose_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<()> {
1269
        if objects.len() <= 1 {
1270
            for (ty, body) in objects {
1271
                self.put_loose(*ty, body)?;
1272
            }
1273
            return Ok(());
1274
        }
1275
        let mut seen = std::collections::HashSet::new();
64818938 Andy McCall ↶ 1276
        let items: Vec<(String, Vec<u8>)> = objects
b30a880f Andy McCall ↶ 1277
            .iter()
1278
            .filter_map(|(ty, body)| {
1279
                let r = hash_object(*ty, body);
64818938 Andy McCall ↶ 1280
                seen.insert(r).then(|| {
1281
                    let mut bytes = header(*ty, body.len());
1282
                    bytes.extend_from_slice(body);
1283
                    (loose_key(&r), bytes)
b30a880f Andy McCall ↶ 1284
                })
1285
            })
1286
            .collect();
64818938 Andy McCall ↶ 1287
        self.backend.put_many_new(&items).map_err(backend_err)
ff0af1ee unknown 1288
    }
1289
09245702 Andy McCall ↶ 1290
    /// Read and verify an object, packed, loose, or pending in a batch.
1291
    /// The packs' indexes are in memory, so they are asked first: asking
1292
    /// for a loose copy of a packed object costs a request that finds
1293
    /// nothing on a store in a bucket, every time (oxygenhub#228).
64818938 Andy McCall ↶ 1294
    pub fn get(&self, r: &ObjectRef) -> Result<Object> {
09245702 Andy McCall ↶ 1295
        if let Some((i, e)) = self.find_packed(r) {
1296
            return self.packs.borrow()[i].fetch(self.backend.as_ref(), r, e);
1297
        }
64818938 Andy McCall ↶ 1298
        match get_optional(self.backend.as_ref(), &loose_key(r))? {
1299
            Some(bytes) => parse_loose(&bytes, r),
09245702 Andy McCall ↶ 1300
            None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))),
afaba5f1 Andy McCall ↶ 1301
        }
1302
    }
1303
09245702 Andy McCall ↶ 1304
    /// Many objects at once, read and verified in parallel: packed ones
1305
    /// from their packs, the others loose (as [`Store::get`], packs first),
1306
    /// then what is pending in a batch. Results are in the order of `refs`.
afaba5f1 Andy McCall ↶ 1307
    pub fn get_many(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
1308
        use rayon::prelude::*;
64818938 Andy McCall ↶ 1309
        let backend = self.backend.as_ref();
afaba5f1 Andy McCall ↶ 1310
        let packs = self.packs.borrow();
09245702 Andy McCall ↶ 1311
        let place: Vec<Option<(usize, PackEntry)>> = refs
75167672 Andy McCall 1312
            .iter()
09245702 Andy McCall ↶ 1313
            .map(|r| {
afaba5f1 Andy McCall ↶ 1314
                packs
1315
                    .iter()
1316
                    .enumerate()
09245702 Andy McCall ↶ 1317
                    .find_map(|(p, pack)| pack.find(r).map(|e| (p, e)))
afaba5f1 Andy McCall ↶ 1318
            })
1319
            .collect();
1320
        let indexes: &[PackIndex] = &packs;
09245702 Andy McCall ↶ 1321
        let found: Vec<Option<Result<Object>>> = refs
afaba5f1 Andy McCall ↶ 1322
            .par_iter()
09245702 Andy McCall ↶ 1323
            .zip(place.par_iter())
1324
            .map(|(r, at)| match at {
1325
                Some((p, e)) => Some(indexes[*p].fetch(backend, r, *e)),
1326
                None => match get_optional(backend, &loose_key(r)) {
1327
                    Ok(Some(bytes)) => Some(parse_loose(&bytes, r)),
1328
                    Ok(None) => None,
1329
                    Err(e) => Some(Err(e)),
1330
                },
1331
            })
afaba5f1 Andy McCall ↶ 1332
            .collect();
1333
        drop(packs);
1334
        found
1335
            .into_iter()
1336
            .zip(refs)
09245702 Andy McCall ↶ 1337
            .map(|(f, r)| {
1338
                f.unwrap_or_else(|| self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))))
afaba5f1 Andy McCall ↶ 1339
            })
1340
            .collect()
ff0af1ee unknown 1341
    }
1342
1343
    pub fn get_typed(&self, r: &ObjectRef, want: ObjectType) -> Result<Vec<u8>> {
1344
        let (ty, body) = self.get(r)?;
1345
        if ty != want {
1346
            return Err(StoreError::WrongType(*r, ty.name(), want.name()));
1347
        }
1348
        Ok(body)
1349
    }
1350
6f8aeb99 Andy McCall ↶ 1351
    /// Store a change and the text blobs (and their chunks) its large
1352
    /// inserts are stored as (spec §6.1).
ff0af1ee unknown 1353
    pub fn put_change(&self, c: &Change) -> Result<ObjectRef> {
5b4c5130 Andy McCall ↶ 1354
        let encoding = crate::timing::phase("store: encode the change and its texts");
17f94b13 Andy McCall ↶ 1355
        let bytes = c.encode_streaming(|batch| self.put_texts(&batch))?;
5b4c5130 Andy McCall ↶ 1356
        drop(encoding);
1357
        let _t = crate::timing::phase("store: put the encoding");
17f94b13 Andy McCall ↶ 1358
        self.put(ObjectType::Change, &bytes)
6f8aeb99 Andy McCall ↶ 1359
    }
1360
17f94b13 Andy McCall ↶ 1361
    /// Store text blobs and their chunks: every chunk at once (compressed
1362
    /// on every core in a large batch), then the blobs.
1363
    pub fn put_texts(&self, texts: &[crate::objects::StoredText]) -> Result<()> {
192fad1e Andy McCall ↶ 1364
        let mut chunks: Vec<(ObjectType, &[u8])> = Vec::new();
1365
        for t in texts {
1366
            if !self.has(&t.blob.hash()) {
1367
                chunks.extend(
1368
                    crate::chunk::chunks(&t.content)
1369
                        .into_iter()
1370
                        .map(|c| (ObjectType::Chunk, c)),
1371
                );
1372
            }
6f8aeb99 Andy McCall ↶ 1373
        }
192fad1e Andy McCall ↶ 1374
        self.put_many(&chunks)?;
1375
        for t in texts {
1376
            self.put(ObjectType::Blob, &t.blob.encode())?;
6f8aeb99 Andy McCall ↶ 1377
        }
17f94b13 Andy McCall ↶ 1378
        Ok(())
ff0af1ee unknown 1379
    }
1380
1381
    pub fn get_change(&self, r: &ObjectRef) -> Result<Change> {
6f8aeb99 Andy McCall ↶ 1382
        Ok(self.get_change_parts(r)?.0)
1383
    }
1384
1385
    /// A change and the objects its stored inserts are made of: each text
1386
    /// blob and its chunks, which travel and are kept with it.
1387
    pub fn get_change_parts(&self, r: &ObjectRef) -> Result<(Change, Vec<ObjectRef>)> {
1388
        let bytes = self.get_typed(r, ObjectType::Change)?;
1389
        let mut parts = Vec::new();
1390
        let mut failed = None;
0f93f49c Andy McCall ↶ 1391
        let decoded = Change::decode_with(&bytes, &mut |text, count, form| match self
1392
            .read_text(text)
1393
        {
6f8aeb99 Andy McCall ↶ 1394
            Ok((blob, content)) => {
1395
                parts.push(*text);
1396
                parts.extend(blob.chunks);
0f93f49c Andy McCall ↶ 1397
                crate::objects::text_lines(&content, count, form)
6f8aeb99 Andy McCall ↶ 1398
            }
1399
            Err(e) => {
1400
                let why = e.to_string();
1401
                failed = Some(e);
1402
                Err(ObjectError::Schema(format!(
1403
                    "text blob {}: {why}",
1404
                    text.short(12)
1405
                )))
1406
            }
1407
        });
1408
        match (decoded, failed) {
1409
            (Ok(c), _) => Ok((c, parts)),
1410
            (Err(_), Some(StoreError::NotFound(t))) => Err(StoreError::Corrupt(
1411
                *r,
1412
                format!("its text blob {} is missing", t.short(12)),
1413
            )),
1414
            (Err(e), _) => Err(StoreError::Corrupt(*r, e.to_string())),
1415
        }
1416
    }
1417
a4fad5b0 Andy McCall ↶ 1418
    /// A change with every insert left without its lines (empty
1419
    /// placeholders), and where each one's lines are, in order: for
1420
    /// history-only states, which read text only for the graphs they build
1421
    /// (see [`Change::decode_skeleton`]).
1422
    pub fn get_change_lazy(&self, r: &ObjectRef) -> Result<(Change, Vec<crate::objects::TextRef>)> {
86e1824f Andy McCall ↶ 1423
        let bytes = self.get_typed(r, ObjectType::Change)?;
a4fad5b0 Andy McCall ↶ 1424
        Change::decode_skeleton(&bytes, *r).map_err(|e| StoreError::Corrupt(*r, e.to_string()))
86e1824f Andy McCall ↶ 1425
    }
1426
69816d68 Andy McCall ↶ 1427
    /// A change's metadata (id, dependencies, author, clock, message, kind)
1428
    /// without its contents: read from its skeleton, so neither its text
1429
    /// blobs nor its inline lines are read (spec §11).
ce14f0d6 Andy McCall ↶ 1430
    ///
1431
    /// Served from the `metas` cache when this store read it before (a
1432
    /// change object can be as large as the tree it adds), as long as the
1433
    /// store still has the change.
69816d68 Andy McCall ↶ 1434
    pub fn get_change_meta(&self, r: &ObjectRef) -> Result<crate::state::ChangeMeta> {
ce14f0d6 Andy McCall ↶ 1435
        if self.metas.borrow().is_none() {
1436
            *self.metas.borrow_mut() = Some(self.read_metas_cache());
1437
        }
1438
        let cached = self.metas.borrow().as_ref().and_then(|m| m.get(r).cloned());
1439
        if let Some(meta) = cached {
1440
            if self.has(r) {
1441
                #[cfg(test)]
1442
                assert_eq!(
1443
                    meta,
1444
                    Change::decode_meta(&self.get_typed(r, ObjectType::Change)?).unwrap(),
1445
                    "the metadata cache holds a change's metadata"
1446
                );
1447
                return Ok(meta);
1448
            }
1449
        }
4d6c9c1b Andy McCall ↶ 1450
        let bytes = self.get_typed(r, ObjectType::Change)?;
ce14f0d6 Andy McCall ↶ 1451
        let meta =
1452
            Change::decode_meta(&bytes).map_err(|e| StoreError::Corrupt(*r, e.to_string()))?;
1453
        self.remember_meta(r, &meta);
1454
        Ok(meta)
1455
    }
1456
1457
    /// The metadata cache: records of a change hash, its metadata in the
1458
    /// flat layout, and the first 8 bytes of BLAKE3 over both, each
1459
    /// appended in one write. Derived only: a record that does not check
64818938 Andy McCall ↶ 1460
    /// ends the reading (a torn write), and the cache can be deleted. It
1461
    /// lives in the scratch directory, on this disk.
ce14f0d6 Andy McCall ↶ 1462
    fn read_metas_cache(&self) -> crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta> {
1463
        let mut out = crate::fasthash::FastMap::default();
64818938 Andy McCall ↶ 1464
        let Ok(bytes) = fs::read(self.scratch.join("metas")) else {
ce14f0d6 Andy McCall ↶ 1465
            return out;
1466
        };
1467
        let mut r = crate::flat::Reader::new(&bytes);
1468
        let mut good = 0;
1469
        while !r.at_end() {
1470
            let mut record = || -> std::result::Result<_, ObjectError> {
1471
                let h = r.object()?;
1472
                let body = r.bytes()?;
1473
                let check = r.fixed::<8>()?;
1474
                Ok((h, body, check))
1475
            };
1476
            let Ok((h, body, check)) = record() else {
1477
                break;
1478
            };
1479
            let mut sum = blake3::Hasher::new();
1480
            sum.update(&h.0).update(body);
1481
            if sum.finalize().as_bytes()[..8] != check {
1482
                break;
1483
            }
1484
            match crate::state::ChangeMeta::read_flat(&mut crate::flat::Reader::new(body)) {
1485
                Ok(m) => out.insert(h, m),
1486
                Err(_) => break,
1487
            };
1488
            good += 32 + 4 + body.len() + 8;
1489
        }
1490
        // Cut a bad tail, so records appended later are read.
1491
        if good < bytes.len() {
1492
            let _ = fs::OpenOptions::new()
1493
                .write(true)
64818938 Andy McCall ↶ 1494
                .open(self.scratch.join("metas"))
ce14f0d6 Andy McCall ↶ 1495
                .and_then(|f| f.set_len(good as u64));
1496
        }
1497
        out
1498
    }
1499
1500
    /// Add a change's metadata to the cache; quietly nothing when it
1501
    /// cannot be written (it is only a cache).
1502
    fn remember_meta(&self, r: &ObjectRef, meta: &crate::state::ChangeMeta) {
1503
        let mut body = crate::flat::Writer::default();
1504
        meta.write_flat(&mut body);
1505
        let mut w = crate::flat::Writer::default();
1506
        w.object(r);
1507
        w.bytes(&body.0);
1508
        let mut sum = blake3::Hasher::new();
1509
        sum.update(&r.0).update(&body.0);
1510
        w.fixed(&sum.finalize().as_bytes()[..8]);
1511
        let _ = fs::OpenOptions::new()
1512
            .create(true)
1513
            .append(true)
64818938 Andy McCall ↶ 1514
            .open(self.scratch.join("metas"))
ce14f0d6 Andy McCall ↶ 1515
            .and_then(|mut f| f.write_all(&w.0));
1516
        if let Some(m) = self.metas.borrow_mut().as_mut() {
1517
            m.insert(*r, meta.clone());
1518
        }
69816d68 Andy McCall ↶ 1519
    }
1520
6f8aeb99 Andy McCall ↶ 1521
    fn read_text(&self, r: &ObjectRef) -> Result<(Blob, Vec<u8>)> {
1522
        let blob = Blob::decode(&self.get_typed(r, ObjectType::Blob)?)?;
1523
        let mut content = Vec::with_capacity(blob.length as usize);
1524
        for c in &blob.chunks {
1525
            content.extend(self.get_typed(c, ObjectType::Chunk)?);
1526
        }
1527
        Ok((blob, content))
ff0af1ee unknown 1528
    }
1529
1530
    pub fn put_chunk(&self, bytes: &[u8]) -> Result<ObjectRef> {
1531
        self.put(ObjectType::Chunk, bytes)
1532
    }
1533
cdb5beba Andy McCall ↶ 1534
    /// Store a chunk that packs will keep uncompressed (`nocompress`, spec §13.3).
1535
    pub fn put_chunk_nocompress(&self, bytes: &[u8]) -> Result<ObjectRef> {
8f10a07a Andy McCall ↶ 1536
        let r = hash_object(ObjectType::Chunk, bytes);
64818938 Andy McCall ↶ 1537
        put_new_or_existing(
1538
            self.backend.as_ref(),
1539
            &format!("nocompress/{}", r.to_hex()),
1540
            b"",
1541
        )?;
8f10a07a Andy McCall ↶ 1542
        self.put(ObjectType::Chunk, bytes)
cdb5beba Andy McCall ↶ 1543
    }
1544
1545
    fn is_nocompress(&self, r: &ObjectRef) -> bool {
64818938 Andy McCall ↶ 1546
        self.backend
1547
            .has(&format!("nocompress/{}", r.to_hex()))
1548
            .unwrap_or(false)
1549
    }
1550
1551
    /// Every chunk marked to be kept raw.
1552
    fn nocompress_set(&self) -> Result<std::collections::HashSet<ObjectRef>> {
1553
        Ok(self
1554
            .backend
1555
            .list("nocompress/")
1556
            .map_err(backend_err)?
1557
            .into_iter()
1558
            .filter_map(|e| ObjectRef::from_hex(e.key.trim_start_matches("nocompress/")))
1559
            .collect())
cdb5beba Andy McCall ↶ 1560
    }
1561
86e682c2 Andy McCall ↶ 1562
    pub fn put_state(&self, st: &StateObject) -> Result<ObjectRef> {
1563
        self.put(ObjectType::State, &st.encode())
1564
    }
1565
1566
    pub fn get_state(&self, r: &ObjectRef) -> Result<StateObject> {
1567
        Ok(StateObject::decode(&self.get_typed(r, ObjectType::State)?)?)
1568
    }
1569
5b9c6ad4 Andy McCall ↶ 1570
    pub fn put_attest(&self, a: &Attest) -> Result<ObjectRef> {
1571
        self.put(ObjectType::Attest, &a.encode())
1572
    }
1573
1574
    pub fn get_attest(&self, r: &ObjectRef) -> Result<Attest> {
1575
        Ok(Attest::decode(&self.get_typed(r, ObjectType::Attest)?)?)
fc1a466a Andy McCall ↶ 1576
    }
1577
1578
    pub fn put_receipt(&self, r: &Receipt) -> Result<ObjectRef> {
1579
        self.put(ObjectType::Receipt, &r.encode())
5b9c6ad4 Andy McCall ↶ 1580
    }
1581
fc1a466a Andy McCall ↶ 1582
    pub fn get_receipt(&self, r: &ObjectRef) -> Result<Receipt> {
1583
        Ok(Receipt::decode(&self.get_typed(r, ObjectType::Receipt)?)?)
5b9c6ad4 Andy McCall ↶ 1584
    }
1585
64818938 Andy McCall ↶ 1586
    /// Record `value` under `key` in `index`: one write-once entry, so two
1587
    /// writers of a shared store never lose each other's.
1588
    fn index_add(&self, index: &str, key: &ObjectRef, value: &ObjectRef) -> Result<()> {
1589
        let entry = format!("{}{}", fanout_prefix(index, key), value.to_hex());
1590
        put_new_or_existing(self.backend.as_ref(), &entry, b"")
fc1a466a Andy McCall ↶ 1591
    }
1592
64818938 Andy McCall ↶ 1593
    /// The values recorded under `key` in `index`, with those an older
1594
    /// store kept as one list in `legacy`, sorted.
1595
    fn index_get(&self, index: &str, legacy: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
1596
        let mut values: Vec<ObjectRef> = self
1597
            .backend
1598
            .list(&fanout_prefix(index, key))
1599
            .map_err(backend_err)?
1600
            .into_iter()
1601
            .filter_map(|e| e.key.rsplit('/').next().and_then(ObjectRef::from_hex))
1602
            .collect();
1603
        if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key(legacy, key))? {
1604
            values.extend(
1605
                String::from_utf8_lossy(&bytes)
1606
                    .lines()
1607
                    .filter_map(ObjectRef::from_hex),
1608
            );
5b9c6ad4 Andy McCall ↶ 1609
        }
fc1a466a Andy McCall ↶ 1610
        values.sort();
64818938 Andy McCall ↶ 1611
        values.dedup();
1612
        Ok(values)
5b9c6ad4 Andy McCall ↶ 1613
    }
1614
fc1a466a Andy McCall ↶ 1615
    /// Record that seal attestation `attest` covers `change`.
1616
    pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> {
64818938 Andy McCall ↶ 1617
        self.index_add("seal", change, attest)
fc1a466a Andy McCall ↶ 1618
    }
1619
1620
    /// Seal attestations known to cover `change`.
1621
    pub fn seals_for(&self, change: &ObjectRef) -> Result<Vec<ObjectRef>> {
64818938 Andy McCall ↶ 1622
        self.index_get("seal", "seals", change)
fc1a466a Andy McCall ↶ 1623
    }
1624
1625
    /// Record receipt attestation `attest` for `state` (spec §14.6).
c63fed2e Andy McCall ↶ 1626
    /// Keep a short-lived key's certificate (spec §14.7), with its issuer's
1627
    /// signature, indexed by the key it certifies.
1628
    pub fn put_certificate(&self, c: &crate::cert::SignedCertificate) -> Result<ObjectRef> {
1629
        let r = self.put(ObjectType::Certificate, &c.cert.encode())?;
1630
        self.put_signature(&r, &c.sig)?;
1631
        self.index_add("cert", &c.cert.key_id(), &r)?;
1632
        Ok(r)
1633
    }
1634
1635
    /// A kept certificate with its issuer's signature.
1636
    pub fn get_certificate(&self, r: &ObjectRef) -> Result<crate::cert::SignedCertificate> {
1637
        let cert = crate::cert::Certificate::decode(&self.get_typed(r, ObjectType::Certificate)?)?;
1638
        let sig = self
1639
            .signatures(r)?
1640
            .into_iter()
1641
            .find(|s| s.key == cert.issuer)
1642
            .ok_or(StoreError::NotFound(*r))?;
1643
        Ok(crate::cert::SignedCertificate { cert, sig })
1644
    }
1645
1646
    /// The certificate kept for short-lived key `key`, if any.
1647
    pub fn certificate_of(&self, key: &ObjectRef) -> Result<Option<ObjectRef>> {
1648
        Ok(self
1649
            .index_get("cert", "certs", key)?
1650
            .into_iter()
1651
            .find(|r| self.has(r)))
1652
    }
1653
f53cbe7a Andy McCall ↶ 1654
    /// Record `value` under `key` in index `index` (for repository code).
1655
    pub(crate) fn index_add_pub(
1656
        &self,
1657
        index: &str,
1658
        key: &ObjectRef,
1659
        value: &ObjectRef,
1660
    ) -> Result<()> {
1661
        self.index_add(index, key, value)
1662
    }
1663
1664
    /// The values recorded under `key` in index `index` (which has no
1665
    /// older one-list form: the legacy name only has to differ from it).
1666
    pub(crate) fn index_get_pub(&self, index: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
1667
        self.index_get(index, &format!("{index}-lists"), key)
1668
    }
1669
fc1a466a Andy McCall ↶ 1670
    pub fn index_receipt(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
64818938 Andy McCall ↶ 1671
        self.index_add("receipt", state, attest)
fc1a466a Andy McCall ↶ 1672
    }
1673
1674
    /// Receipt attestations recorded for `state`.
1675
    pub fn receipts_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
64818938 Andy McCall ↶ 1676
        self.index_get("receipt", "receipts", state)
fc1a466a Andy McCall ↶ 1677
    }
1678
3a4b3e01 Andy McCall ↶ 1679
    pub fn index_approval(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
1680
        self.index_add("approval", state, attest)
1681
    }
1682
1683
    /// Approval attestations recorded for `state` (OxygenFlow approval
1684
    /// gates, oxygenhub#235).
1685
    pub fn approvals_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
1686
        self.index_get("approval", "approvals", state)
1687
    }
1688
1689
    /// Every state with an approval, for garbage collection.
1690
    pub fn approval_states(&self) -> Result<Vec<ObjectRef>> {
1691
        self.indexed_states("approval", "approvals")
1692
    }
1693
fc1a466a Andy McCall ↶ 1694
    /// Every state with a receipt, for garbage collection.
1695
    pub fn receipt_states(&self) -> Result<Vec<ObjectRef>> {
3a4b3e01 Andy McCall ↶ 1696
        self.indexed_states("receipt", "receipts")
1697
    }
1698
1699
    /// The states an index (`one/<hh>/<rest>/<attest>` entries, and the
1700
    /// packed `many/<hh>/<rest>` lists) holds attestations for.
1701
    fn indexed_states(&self, one: &str, many: &str) -> Result<Vec<ObjectRef>> {
64818938 Andy McCall ↶ 1702
        let mut out: Vec<ObjectRef> = Vec::new();
3a4b3e01 Andy McCall ↶ 1703
        for e in self.backend.list(&format!("{one}/")).map_err(backend_err)? {
64818938 Andy McCall ↶ 1704
            // receipt/<hh>/<rest>/<attest>
1705
            let mut parts = e.key.splitn(4, '/');
1706
            let (Some(_), Some(hh), Some(rest), Some(_)) =
1707
                (parts.next(), parts.next(), parts.next(), parts.next())
1708
            else {
1709
                continue;
1710
            };
1711
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
fc1a466a Andy McCall ↶ 1712
        }
3a4b3e01 Andy McCall ↶ 1713
        for e in self
1714
            .backend
1715
            .list(&format!("{many}/"))
1716
            .map_err(backend_err)?
1717
        {
64818938 Andy McCall ↶ 1718
            let mut parts = e.key.splitn(3, '/');
1719
            let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next()) else {
1720
                continue;
1721
            };
1722
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
1723
        }
fc1a466a Andy McCall ↶ 1724
        out.sort();
64818938 Andy McCall ↶ 1725
        out.dedup();
fc1a466a Andy McCall ↶ 1726
        Ok(out)
5b9c6ad4 Andy McCall ↶ 1727
    }
1728
1983686f Andy McCall ↶ 1729
    pub fn put_identity(&self, id: &Identity) -> Result<ObjectRef> {
1730
        self.put(ObjectType::Identity, &id.encode())
1731
    }
1732
1733
    pub fn get_identity(&self, r: &ObjectRef) -> Result<Identity> {
1734
        Ok(Identity::decode(&self.get_typed(r, ObjectType::Identity)?)?)
1735
    }
1736
64818938 Andy McCall ↶ 1737
    /// Signatures stored alongside object `r` (spec §5.4), one per signing
1738
    /// key, sorted by key. Empty if none.
1983686f Andy McCall ↶ 1739
    pub fn signatures(&self, r: &ObjectRef) -> Result<Vec<Signature>> {
64818938 Andy McCall ↶ 1740
        let mut sigs: Vec<Signature> = Vec::new();
1741
        for e in self
1742
            .backend
1743
            .list(&fanout_prefix("sig", r))
1744
            .map_err(backend_err)?
1745
        {
1746
            let bytes = self.backend.get(&e.key).map_err(backend_err)?;
1747
            sigs.extend(Signature::decode_list(&bytes)?);
1983686f Andy McCall ↶ 1748
        }
64818938 Andy McCall ↶ 1749
        // An older store's one list per object; a signer kept there and
1750
        // here counts once, as written here.
1751
        if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key("sigs", r))? {
1752
            for sig in Signature::decode_list(&bytes)? {
1753
                if !sigs.iter().any(|s| s.key == sig.key) {
1754
                    sigs.push(sig);
1755
                }
1756
            }
1757
        }
1758
        sigs.sort_by_key(|s| s.key);
1759
        sigs.dedup_by_key(|s| s.key);
1760
        Ok(sigs)
1983686f Andy McCall ↶ 1761
    }
1762
64818938 Andy McCall ↶ 1763
    /// Attach a signature to object `r`, replacing any earlier one by the
1764
    /// same key: one entry per signer, so the latest by a key is the one
1765
    /// kept, and two signers never touch each other's.
1983686f Andy McCall ↶ 1766
    pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> {
64818938 Andy McCall ↶ 1767
        let key = format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex());
1768
        let bytes = Signature::encode_list(std::slice::from_ref(sig));
1769
        if get_optional(self.backend.as_ref(), &key)?.as_deref() == Some(bytes.as_slice()) {
1770
            return Ok(());
1983686f Andy McCall ↶ 1771
        }
64818938 Andy McCall ↶ 1772
        delete(self.backend.as_ref(), &key)?;
1773
        put_new_or_existing(self.backend.as_ref(), &key, &bytes)
1983686f Andy McCall ↶ 1774
    }
1775
54e2e630 Andy McCall ↶ 1776
    /// Finish packs a writer left unfinished: a batch writes its objects to
1777
    /// `.tmp-ID.pack` and only indexes it at the end, so a process that
1778
    /// stopped part way (an import killed after hours) leaves every object
1779
    /// in a pack nothing can read. Each such pack untouched for `min_age`
1780
    /// is read through: every whole entry checked against its reference,
1781
    /// a torn last one cut off, then indexed as a pack of the store.
1782
    /// Returns the packs and objects adopted.
1783
    pub fn adopt_unfinished_packs(
1784
        &self,
1785
        min_age: std::time::Duration,
1786
        progress: &dyn Fn(&str),
1787
    ) -> Result<(usize, usize)> {
1788
        use std::io::{BufReader, Read};
1789
        let mut packs = 0;
1790
        let mut objects = 0;
1791
        let Ok(dir) = fs::read_dir(&self.scratch) else {
1792
            return Ok((0, 0));
1793
        };
1794
        for entry in dir.flatten() {
1795
            let name = entry.file_name().to_string_lossy().to_string();
1796
            let Some(id) = name
1797
                .strip_prefix(".tmp-")
1798
                .and_then(|n| n.strip_suffix(".pack"))
1799
            else {
1800
                continue;
1801
            };
1802
            let path = entry.path();
1803
            let meta = fs::metadata(&path)?;
1804
            let age = meta
1805
                .modified()
1806
                .ok()
1807
                .and_then(|m| m.elapsed().ok())
1808
                .unwrap_or_default();
1809
            if age < min_age {
1810
                progress(&format!(
1811
                    "{name}: written to {}s ago, left alone",
1812
                    age.as_secs()
1813
                ));
1814
                continue;
1815
            }
1816
            let mut f = BufReader::with_capacity(1 << 20, fs::File::open(&path)?);
1817
            let mut magic = [0u8; 8];
1818
            if f.read_exact(&mut magic).is_err() || &magic != PACK_MAGIC {
1819
                progress(&format!("{name}: not a pack, left alone"));
1820
                continue;
1821
            }
1822
            let mut offset = PACK_MAGIC.len() as u64;
1823
            let mut index: Vec<(ObjectRef, PackEntry)> = Vec::new();
1824
            loop {
1825
                let mut head = [0u8; 42];
1826
                if f.read_exact(&mut head).is_err() {
1827
                    break;
1828
                }
1829
                let Some(ty) = ObjectType::ALL.get(head[0] as usize).copied() else {
1830
                    break;
1831
                };
1832
                let flags = head[1];
1833
                let stored_len = u32::from_le_bytes(head[2..6].try_into().unwrap());
1834
                let raw_len = u32::from_le_bytes(head[6..10].try_into().unwrap());
1835
                let r = ObjectRef(head[10..42].try_into().unwrap());
1836
                let mut stored = vec![0u8; stored_len as usize];
1837
                if flags & FLAG_DICT != 0 || f.read_exact(&mut stored).is_err() {
1838
                    break;
1839
                }
1840
                let e = PackEntry {
1841
                    offset: offset + 42,
1842
                    stored_len,
1843
                    raw_len,
1844
                    ty,
1845
                    flags,
1846
                };
1847
                if decode_entry(&r, e, stored, None).is_err() {
1848
                    break;
1849
                }
1850
                index.push((r, e));
1851
                offset += 42 + stored_len as u64;
1852
                if index.len().is_multiple_of(200_000) {
1853
                    progress(&format!(
1854
                        "{name}: {} objects checked, {} MiB",
1855
                        index.len(),
1856
                        offset >> 20
1857
                    ));
1858
                }
1859
            }
1860
            if index.is_empty() {
1861
                progress(&format!("{name}: no whole object, left alone"));
1862
                continue;
1863
            }
1864
            if offset < meta.len() {
1865
                progress(&format!(
1866
                    "{name}: {} torn bytes at the end cut off",
1867
                    meta.len() - offset
1868
                ));
1869
                fs::OpenOptions::new()
1870
                    .write(true)
1871
                    .open(&path)?
1872
                    .set_len(offset)?;
1873
            }
1874
            index.sort_by_key(|(r, _)| *r);
1875
            index.dedup_by_key(|(r, _)| *r);
1876
            objects += index.len();
1877
            packs += 1;
1878
            progress(&format!("{name}: {} objects indexed", index.len()));
1879
            self.finish_pack_files(id, &path, &index)?;
1880
        }
1881
        Ok((packs, objects))
1882
    }
1883
d02e545d Andy McCall ↶ 1884
    /// Every object reference in the store, loose and packed (slow; for gc and tests).
ff0af1ee unknown 1885
    pub fn list(&self) -> io::Result<Vec<ObjectRef>> {
64818938 Andy McCall ↶ 1886
        let mut out: Vec<ObjectRef> = self
1887
            .loose_objects()
1888
            .map_err(|e| io::Error::other(e.to_string()))?
1889
            .into_iter()
1890
            .map(|(r, _)| r)
1891
            .collect();
d02e545d Andy McCall ↶ 1892
        for p in self.packs.borrow().iter() {
1893
            out.extend(p.refs.iter().copied());
ff0af1ee unknown 1894
        }
1895
        out.sort();
d02e545d Andy McCall ↶ 1896
        out.dedup();
ff0af1ee unknown 1897
        Ok(out)
1898
    }
afaba5f1 Andy McCall ↶ 1899
}
1900
64818938 Andy McCall ↶ 1901
/// A loose object's bytes, verified against `r`.
1902
fn parse_loose(data: &[u8], r: &ObjectRef) -> Result<Object> {
1903
    let nul = data
1904
        .iter()
1905
        .position(|&b| b == 0)
1906
        .ok_or_else(|| StoreError::Corrupt(*r, "no header".into()))?;
1907
    let head = std::str::from_utf8(&data[..nul])
1908
        .map_err(|_| StoreError::Corrupt(*r, "header not UTF-8".into()))?;
1909
    let mut parts = head.split(' ');
1910
    let (Some("oxy1"), Some(ty), Some(len), None) =
1911
        (parts.next(), parts.next(), parts.next(), parts.next())
1912
    else {
1913
        return Err(StoreError::Corrupt(*r, format!("bad header {head:?}")));
afaba5f1 Andy McCall ↶ 1914
    };
64818938 Andy McCall ↶ 1915
    let ty = ObjectType::from_name(ty)
1916
        .ok_or_else(|| StoreError::Corrupt(*r, format!("unknown type {ty}")))?;
1917
    let len: usize = len
1918
        .parse()
1919
        .map_err(|_| StoreError::Corrupt(*r, "bad length".into()))?;
1920
    let body = &data[nul + 1..];
1921
    if body.len() != len {
1922
        return Err(StoreError::Corrupt(
1923
            *r,
1924
            format!("length {} != header {len}", body.len()),
1925
        ));
1926
    }
1927
    if hash_object(ty, body) != *r {
1928
        return Err(StoreError::Corrupt(*r, "hash mismatch".into()));
1929
    }
1930
    Ok((ty, body.to_vec()))
1931
}
1932
1933
/// Whether the objects are files that a test may tamper with: not under
1934
/// `OXY_STORE_BACKEND=memory`.
1935
#[cfg(test)]
1936
fn files_on_disk() -> bool {
1937
    std::env::var_os("OXY_STORE_BACKEND").is_none()
ff0af1ee unknown 1938
}
1939
1940
#[cfg(test)]
1941
mod tests {
1942
    use super::*;
64818938 Andy McCall ↶ 1943
1944
    /// The store under test twice: in a directory, and on a store in
1945
    /// memory with a scratch directory, as over a bucket.
1946
    fn each_store(dir: &Path, f: impl Fn(&str, Store)) {
1947
        f("directory", Store::open(dir.join("objects")).unwrap());
1948
        f(
1949
            "memory",
1950
            Store::open_on(Arc::new(MemoryStore::new()), dir.join("scratch")).unwrap(),
1951
        );
1952
    }
ff0af1ee unknown 1953
1954
    #[test]
1955
    fn put_get_verify() {
64818938 Andy McCall ↶ 1956
        if !super::files_on_disk() {
1957
            return;
1958
        }
ff0af1ee unknown 1959
        let dir = tempfile::tempdir().unwrap();
1960
        let s = Store::open(dir.path().join("objects")).unwrap();
1961
        let r = s.put(ObjectType::Chunk, b"hello").unwrap();
1962
        assert!(s.has(&r));
1963
        assert_eq!(s.get(&r).unwrap(), (ObjectType::Chunk, b"hello".to_vec()));
1964
        assert_eq!(s.put(ObjectType::Chunk, b"hello").unwrap(), r);
1965
        assert_eq!(s.list().unwrap(), vec![r]);
75167672 Andy McCall 1966
        assert!(matches!(
1967
            s.get_typed(&r, ObjectType::Blob),
1968
            Err(StoreError::WrongType(..))
1969
        ));
1970
        assert!(matches!(
1971
            s.get(&ObjectRef([9; 32])),
1972
            Err(StoreError::NotFound(_))
1973
        ));
ff0af1ee unknown 1974
1975
        // Tamper: flip a byte, read must fail.
1976
        let path = s.path_for(&r);
1977
        let mut data = fs::read(&path).unwrap();
1978
        let last = data.len() - 1;
1979
        data[last] ^= 1;
1980
        fs::write(&path, data).unwrap();
1981
        assert!(matches!(s.get(&r), Err(StoreError::Corrupt(..))));
1982
    }
1983
1984
    #[test]
d02e545d Andy McCall ↶ 1985
    fn packing_keeps_every_object_readable() {
64818938 Andy McCall ↶ 1986
        if !super::files_on_disk() {
1987
            return;
1988
        }
d02e545d Andy McCall ↶ 1989
        let dir = tempfile::tempdir().unwrap();
1990
        let s = Store::open(dir.path()).unwrap();
1991
        let mut refs = Vec::new();
1992
        for i in 0..300u32 {
1993
            let body = format!("object {i} ").repeat(1 + (i % 7) as usize);
1994
            refs.push((s.put(ObjectType::Chunk, body.as_bytes()).unwrap(), body));
1995
        }
1996
        let c = crate::objects::tests::sample_change();
1997
        let cr = s.put_change(&c).unwrap();
1998
        assert_eq!(s.loose_count().unwrap(), 301);
1999
        let report = s.pack_loose().unwrap();
2000
        assert_eq!(report.objects, 301);
2001
        assert!(report.packed_bytes < report.raw_bytes, "{report:?}");
2002
        assert_eq!(s.loose_count().unwrap(), 0);
2003
        assert_eq!(s.pack_count(), 1);
2004
        for (r, body) in &refs {
2005
            assert!(s.has(r));
75167672 Andy McCall 2006
            assert_eq!(
2007
                s.get(r).unwrap(),
2008
                (ObjectType::Chunk, body.as_bytes().to_vec())
2009
            );
d02e545d Andy McCall ↶ 2010
        }
2011
        assert_eq!(s.get_change(&cr).unwrap(), c);
2012
        assert_eq!(s.list().unwrap().len(), 301);
2013
        // Re-putting a packed object is a no-op; new objects go loose and a
2014
        // second pack coexists with the first.
75167672 Andy McCall 2015
        assert_eq!(
2016
            s.put(ObjectType::Chunk, refs[0].1.as_bytes()).unwrap(),
2017
            refs[0].0
2018
        );
d02e545d Andy McCall ↶ 2019
        assert_eq!(s.loose_count().unwrap(), 0);
2020
        let extra = s.put(ObjectType::Chunk, b"after packing").unwrap();
2021
        s.pack_loose().unwrap();
2022
        assert_eq!(s.pack_count(), 2);
2023
        assert_eq!(s.get(&extra).unwrap().1, b"after packing");
2024
        // A reopened store sees the packs.
2025
        let s2 = Store::open(dir.path()).unwrap();
2026
        assert_eq!(s2.pack_count(), 2);
2027
        assert_eq!(s2.get(&refs[5].0).unwrap().1, refs[5].1.as_bytes());
75167672 Andy McCall 2028
        assert!(matches!(
2029
            s2.get(&ObjectRef([9; 32])),
2030
            Err(StoreError::NotFound(_))
2031
        ));
d02e545d Andy McCall ↶ 2032
        // Corrupting an object's body inside the pack is detected on read.
2033
        let target = refs[5].0;
2034
        let (pack_path, offset) = {
2035
            let packs = s2.packs.borrow();
2036
            let (i, e) = s2.find_packed(&target).unwrap();
64818938 Andy McCall ↶ 2037
            (dir.path().join(&packs[i].key), e.offset as usize)
d02e545d Andy McCall ↶ 2038
        };
2039
        let mut data = fs::read(&pack_path).unwrap();
2040
        data[offset] ^= 0xff;
2041
        fs::write(&pack_path, data).unwrap();
2042
        let s3 = Store::open(dir.path()).unwrap();
2043
        assert!(matches!(s3.get(&target), Err(StoreError::Corrupt(..))));
75167672 Andy McCall 2044
        assert_eq!(
2045
            s3.get(&refs[6].0).unwrap().1,
2046
            refs[6].1.as_bytes(),
2047
            "other objects are unaffected"
2048
        );
a17fc622 Andy McCall ↶ 2049
    }
2050
2051
    #[test]
2052
    fn a_full_repack_trains_a_dictionary_and_reads_back() {
64818938 Andy McCall ↶ 2053
        if !super::files_on_disk() {
2054
            return;
2055
        }
a17fc622 Andy McCall ↶ 2056
        let dir = tempfile::tempdir().unwrap();
2057
        let s = Store::open(dir.path()).unwrap();
2058
        // Many small, alike files: what a dictionary is for.
2059
        let bodies: Vec<Vec<u8>> = (0..1500)
2060
            .map(|i| {
2061
                format!(
2062
                    "// SPDX-License-Identifier: GPL-2.0\n#include <linux/module.h>\n\
2063
                     static int probe_{i}(struct device *dev)\n{{\n\treturn {i};\n}}\n"
2064
                )
2065
                .into_bytes()
2066
            })
2067
            .collect();
2068
        let refs: Vec<ObjectRef> = bodies
2069
            .iter()
2070
            .map(|b| s.put(ObjectType::Chunk, b).unwrap())
2071
            .collect();
2072
        s.repack_all(REPACK_ZSTD_LEVEL).unwrap();
2073
        let dicts = fs::read_dir(dir.path())
2074
            .unwrap()
2075
            .flatten()
2076
            .filter(|e| e.path().extension().is_some_and(|x| x == "dict"))
2077
            .count();
2078
        assert_eq!(dicts, 1, "the pack has a dictionary");
2079
        let with_dict = refs
2080
            .iter()
2081
            .filter(|r| s.find_packed(r).unwrap().1.flags & FLAG_DICT != 0)
2082
            .count();
2083
        assert!(with_dict > 1000, "most use it: {with_dict}");
2084
        let again = Store::open(dir.path()).unwrap();
2085
        for (r, b) in refs.iter().zip(&bodies) {
2086
            assert_eq!(&again.get(r).unwrap().1, b);
2087
        }
a67914bc Andy McCall ↶ 2088
        let packs = |dir: &Path| {
2089
            let mut v: Vec<String> = fs::read_dir(dir)
2090
                .unwrap()
2091
                .flatten()
2092
                .map(|e| e.file_name().to_string_lossy().into_owned())
2093
                .filter(|n| n.starts_with("pack-"))
2094
                .collect();
2095
            v.sort();
2096
            v
2097
        };
2098
        // With nothing new, a repack leaves that pack as it is: it would
2099
        // only write the same again.
2100
        let before = packs(dir.path());
2101
        let (report, merged) = again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
2102
        assert_eq!((report.objects, merged), (0, 0));
2103
        assert_eq!(packs(dir.path()), before);
2104
        // One new object: a repack replaces the pack and its dictionary.
2105
        let extra = again.put(ObjectType::Chunk, b"one more\n").unwrap();
a17fc622 Andy McCall ↶ 2106
        again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
a67914bc Andy McCall ↶ 2107
        let files = packs(dir.path());
a17fc622 Andy McCall ↶ 2108
        assert_eq!(files.len(), 3, "one pack, index and dictionary: {files:?}");
a67914bc Andy McCall ↶ 2109
        assert_ne!(files, before);
a17fc622 Andy McCall ↶ 2110
        assert_eq!(&again.get(&refs[7]).unwrap().1, &bodies[7]);
a67914bc Andy McCall ↶ 2111
        assert_eq!(again.get(&extra).unwrap().1, b"one more\n");
d02e545d Andy McCall ↶ 2112
    }
2113
2114
    #[test]
cdb5beba Andy McCall ↶ 2115
    fn nocompress_chunks_pack_raw() {
2116
        let dir = tempfile::tempdir().unwrap();
2117
        let s = Store::open(dir.path()).unwrap();
2118
        let body = vec![b'a'; 10_000];
2119
        let raw = s.put_chunk_nocompress(&body).unwrap();
2120
        let compressible = s.put(ObjectType::Chunk, &vec![b'b'; 10_000]).unwrap();
2121
        s.pack_loose().unwrap();
2122
        let packs = s.packs.borrow();
2123
        let (i, e) = s.find_packed(&raw).unwrap();
2124
        assert_eq!((e.flags, e.stored_len as usize), (0, 10_000));
2125
        let (_, c) = s.find_packed(&compressible).unwrap();
2126
        assert_eq!(c.flags, FLAG_ZSTD);
2127
        assert!(c.stored_len < 10_000);
64818938 Andy McCall ↶ 2128
        assert_eq!(packs[i].fetch(s.backend.as_ref(), &raw, e).unwrap().1, body);
cdb5beba Andy McCall ↶ 2129
    }
2130
2131
    #[test]
135212de Andy McCall ↶ 2132
    fn many_packs_merge_into_one() {
2133
        let dir = tempfile::tempdir().unwrap();
64818938 Andy McCall ↶ 2134
        each_store(dir.path(), |which, s| {
2135
            let mut refs = Vec::new();
2136
            for i in 0..(MAX_PACKS + 2) {
2137
                refs.push(
2138
                    s.put(ObjectType::Chunk, format!("pack {i}").as_bytes())
2139
                        .unwrap(),
2140
                );
2141
                s.pack_loose().unwrap();
2142
            }
2143
            assert_eq!(s.pack_count(), MAX_PACKS + 2, "{which}");
2144
            assert_eq!(s.merge_packs().unwrap(), MAX_PACKS + 2);
2145
            assert_eq!(s.pack_count(), 1);
2146
            assert_eq!(s.merge_packs().unwrap(), 0);
2147
            for (i, r) in refs.iter().enumerate() {
2148
                assert_eq!(s.get(r).unwrap().1, format!("pack {i}").as_bytes());
2149
            }
2150
            let again = Store::open_on(s.backend().clone(), s.dir()).unwrap();
2151
            assert_eq!(again.pack_count(), 1, "{which}: reopened");
2152
        });
135212de Andy McCall ↶ 2153
    }
2154
2155
    #[test]
ba160f60 Andy McCall ↶ 2156
    fn prune_removes_unreachable_loose_and_packed_objects() {
2157
        use std::collections::HashSet;
2158
        let dir = tempfile::tempdir().unwrap();
64818938 Andy McCall ↶ 2159
        each_store(dir.path(), |which, s| {
2160
            let keep_a = s.put(ObjectType::Chunk, b"keep a").unwrap();
2161
            let drop_a = s.put(ObjectType::Chunk, b"drop a").unwrap();
2162
            s.pack_loose().unwrap();
2163
            let keep_b = s.put(ObjectType::Chunk, b"keep b").unwrap();
2164
            let drop_b = s.put(ObjectType::Chunk, b"drop b").unwrap();
2165
            let sk = crate::identity::SecretKey {
2166
                ed_seed: [1; 32],
2167
                ml_seed: [2; 32],
2168
            };
2169
            let id = sk.identity("a", "b", 0);
2170
            let key = s.put_identity(&id).unwrap();
2171
            s.put_signature(
2172
                &drop_b,
2173
                &sk.sign(crate::identity::Domain::Change, &key, &drop_b),
2174
            )
75167672 Andy McCall 2175
            .unwrap();
64818938 Andy McCall ↶ 2176
            s.index_seal(&drop_a, &ObjectRef([3; 32])).unwrap();
2177
            let keep: HashSet<ObjectRef> = [keep_a, keep_b, key].into_iter().collect();
2178
            // Everything is too young under a grace period.
2179
            let r = s
2180
                .prune(&keep, std::time::Duration::from_secs(3600))
2181
                .unwrap();
2182
            assert_eq!((r.objects, r.deferred), (0, 2), "{which}");
2183
            let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
2184
            assert_eq!(r.objects, 2, "{which}: {r:?}");
2185
            assert!(s.has(&keep_a) && s.has(&keep_b) && s.has(&key));
2186
            assert!(!s.has(&drop_a) && !s.has(&drop_b));
2187
            assert!(s.signatures(&drop_b).unwrap().is_empty());
2188
            assert!(s.seals_for(&drop_a).unwrap().is_empty());
2189
            assert_eq!(s.pack_count(), 1);
2190
            assert_eq!(s.get(&keep_a).unwrap().1, b"keep a");
2191
            assert_eq!(s.list().unwrap().len(), 3);
2192
        });
ba160f60 Andy McCall ↶ 2193
    }
2194
2195
    #[test]
1983686f Andy McCall ↶ 2196
    fn signatures_are_stored_beside_objects() {
64818938 Andy McCall ↶ 2197
        use crate::identity::{Domain, SecretKey};
2198
        let dir = tempfile::tempdir().unwrap();
2199
        each_store(dir.path(), |which, s| {
2200
            let r = s.put(ObjectType::Chunk, b"x").unwrap();
2201
            assert!(s.signatures(&r).unwrap().is_empty());
2202
            let sk = SecretKey {
2203
                ed_seed: [1; 32],
2204
                ml_seed: [2; 32],
2205
            };
2206
            let id = sk.identity("a", "b", 0);
2207
            let key = s.put_identity(&id).unwrap();
2208
            assert_eq!(s.get_identity(&key).unwrap(), id);
2209
            let sig = sk.sign(Domain::Change, &key, &r);
2210
            s.put_signature(&r, &sig).unwrap();
2211
            s.put_signature(&r, &sig).unwrap();
2212
            assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()], "{which}");
2213
            // A second signer is kept beside the first, sorted by key.
2214
            let sk2 = SecretKey {
2215
                ed_seed: [3; 32],
2216
                ml_seed: [4; 32],
2217
            };
2218
            let key2 = s.put_identity(&sk2.identity("c", "d", 0)).unwrap();
2219
            let sig2 = sk2.sign(Domain::Change, &key2, &r);
2220
            s.put_signature(&r, &sig2).unwrap();
2221
            let mut both = vec![sig.clone(), sig2.clone()];
2222
            both.sort_by_key(|s| s.key);
2223
            assert_eq!(s.signatures(&r).unwrap(), both);
2224
            // The object itself is untouched by its signatures.
2225
            assert_eq!(s.get(&r).unwrap().1, b"x");
2226
            // Seal and receipt entries, and the states with receipts.
2227
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
2228
            s.index_seal(&r, &ObjectRef([4; 32])).unwrap();
2229
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
2230
            assert_eq!(
2231
                s.seals_for(&r).unwrap(),
2232
                vec![ObjectRef([4; 32]), ObjectRef([5; 32])]
2233
            );
2234
            let st = ObjectRef([6; 32]);
2235
            s.index_receipt(&st, &ObjectRef([7; 32])).unwrap();
2236
            assert_eq!(s.receipts_for(&st).unwrap(), vec![ObjectRef([7; 32])]);
2237
            assert_eq!(s.receipt_states().unwrap(), vec![st]);
2238
        });
2239
    }
2240
2241
    /// A store written before signatures, seals and receipts were one
2242
    /// entry each (`sigs/`, `seals/`, `receipts/` lists) still reads, and
2243
    /// what is written now sits beside what was there.
2244
    #[test]
2245
    fn an_older_store_s_sidecar_lists_are_still_read() {
2246
        if !super::files_on_disk() {
2247
            return;
2248
        }
1983686f Andy McCall ↶ 2249
        use crate::identity::{Domain, SecretKey};
2250
        let dir = tempfile::tempdir().unwrap();
2251
        let s = Store::open(dir.path()).unwrap();
64818938 Andy McCall ↶ 2252
        let r = s.put(ObjectType::Chunk, b"old").unwrap();
2253
        let h = r.to_hex();
75167672 Andy McCall 2254
        let sk = SecretKey {
2255
            ed_seed: [1; 32],
2256
            ml_seed: [2; 32],
2257
        };
64818938 Andy McCall ↶ 2258
        let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap();
1983686f Andy McCall ↶ 2259
        let sig = sk.sign(Domain::Change, &key, &r);
64818938 Andy McCall ↶ 2260
        let legacy = dir.path().join("sigs").join(&h[..2]);
2261
        fs::create_dir_all(&legacy).unwrap();
2262
        fs::write(
2263
            legacy.join(&h[2..]),
2264
            Signature::encode_list(std::slice::from_ref(&sig)),
2265
        )
2266
        .unwrap();
2267
        let seals = dir.path().join("seals").join(&h[..2]);
2268
        fs::create_dir_all(&seals).unwrap();
2269
        fs::write(
2270
            seals.join(&h[2..]),
2271
            format!("{}\n", ObjectRef([9; 32]).to_hex()),
2272
        )
2273
        .unwrap();
2274
        let receipts = dir.path().join("receipts").join(&h[..2]);
2275
        fs::create_dir_all(&receipts).unwrap();
2276
        fs::write(
2277
            receipts.join(&h[2..]),
2278
            format!("{}\n", ObjectRef([8; 32]).to_hex()),
2279
        )
2280
        .unwrap();
2281
2282
        assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()]);
2283
        assert_eq!(s.seals_for(&r).unwrap(), vec![ObjectRef([9; 32])]);
2284
        assert_eq!(s.receipts_for(&r).unwrap(), vec![ObjectRef([8; 32])]);
2285
        assert_eq!(s.receipt_states().unwrap(), vec![r]);
2286
        // New entries join the old ones; the same signer is not doubled.
1983686f Andy McCall ↶ 2287
        s.put_signature(&r, &sig).unwrap();
64818938 Andy McCall ↶ 2288
        s.index_seal(&r, &ObjectRef([1; 32])).unwrap();
2289
        assert_eq!(s.signatures(&r).unwrap().len(), 1);
2290
        assert_eq!(
2291
            s.seals_for(&r).unwrap(),
2292
            vec![ObjectRef([1; 32]), ObjectRef([9; 32])]
2293
        );
2294
        // Pruning the object takes both layouts with it.
2295
        let r2 = s.put(ObjectType::Chunk, b"kept").unwrap();
2296
        let keep = [r2, key].into_iter().collect();
2297
        s.prune(&keep, std::time::Duration::ZERO).unwrap();
2298
        assert!(s.signatures(&r).unwrap().is_empty());
2299
        assert!(s.seals_for(&r).unwrap().is_empty());
2300
        assert!(s.receipt_states().unwrap().is_empty());
1983686f Andy McCall ↶ 2301
    }
2302
2303
    #[test]
ff0af1ee unknown 2304
    fn change_round_trip_through_store() {
2305
        let dir = tempfile::tempdir().unwrap();
64818938 Andy McCall ↶ 2306
        each_store(dir.path(), |which, s| {
2307
            let c = crate::objects::tests::sample_change();
2308
            let r = s.put_change(&c).unwrap();
2309
            assert_eq!(r, c.hash(), "{which}");
2310
            assert_eq!(s.get_change(&r).unwrap(), c);
2311
            assert_eq!(s.get_change_meta(&r).unwrap().message, c.message);
2312
        });
ff0af1ee unknown 2313
    }
2b872d07 Andy McCall ↶ 2314
2315
    fn xorshift(state: &mut u64) -> u64 {
2316
        *state ^= *state << 13;
2317
        *state ^= *state >> 7;
2318
        *state ^= *state << 17;
2319
        *state
2320
    }
2321
2322
    /// Random corruption of pack and index files never panics: reads fail
2323
    /// with an error or the store refuses to open, and untouched objects
2324
    /// stay readable when the index survives.
2325
    #[test]
2326
    fn corrupted_packs_and_indexes_fail_cleanly() {
64818938 Andy McCall ↶ 2327
        if !super::files_on_disk() {
2328
            return;
2329
        }
2b872d07 Andy McCall ↶ 2330
        let dir = tempfile::tempdir().unwrap();
2331
        let objects = dir.path().join("objects");
2332
        let s = Store::open(&objects).unwrap();
2333
        let mut refs = Vec::new();
2334
        let mut seed = 0x9e3779b97f4a7c15u64;
2335
        for i in 0..40u32 {
2336
            let len = (xorshift(&mut seed) % 5000) as usize;
2337
            let mut body = vec![0u8; len];
2338
            for b in body.iter_mut() {
75167672 Andy McCall 2339
                *b = if i % 2 == 0 {
2340
                    (xorshift(&mut seed) & 0xff) as u8
2341
                } else {
2342
                    b'a' + (i % 26) as u8
2343
                };
2b872d07 Andy McCall ↶ 2344
            }
2345
            refs.push(s.put(ObjectType::Chunk, &body).unwrap());
2346
        }
2347
        s.pack_loose().unwrap();
75167672 Andy McCall 2348
        let pack = fs::read_dir(&objects)
2349
            .unwrap()
2350
            .filter_map(|e| e.ok().map(|e| e.path()))
2351
            .find(|p| p.extension().is_some_and(|x| x == "pack"))
2352
            .unwrap();
2b872d07 Andy McCall ↶ 2353
        let idx = pack.with_extension("idx");
2354
        let pristine_pack = fs::read(&pack).unwrap();
2355
        let pristine_idx = fs::read(&idx).unwrap();
2356
2357
        for round in 0..300u32 {
75167672 Andy McCall 2358
            let (path, pristine) = if round % 2 == 0 {
2359
                (&pack, &pristine_pack)
2360
            } else {
2361
                (&idx, &pristine_idx)
2362
            };
2b872d07 Andy McCall ↶ 2363
            let mut data = pristine.clone();
2364
            match xorshift(&mut seed) % 4 {
2365
                0 => {
2366
                    let at = (xorshift(&mut seed) as usize) % data.len();
2367
                    data[at] ^= 1 << (xorshift(&mut seed) % 8);
2368
                }
2369
                1 => {
2370
                    let at = (xorshift(&mut seed) as usize) % data.len();
2371
                    data[at] = (xorshift(&mut seed) & 0xff) as u8;
2372
                }
2373
                2 => {
2374
                    let keep = (xorshift(&mut seed) as usize) % data.len();
2375
                    data.truncate(keep);
2376
                }
2377
                _ => {
2378
                    let at = (xorshift(&mut seed) as usize) % data.len();
2379
                    let n = ((xorshift(&mut seed) % 8) + 1) as usize;
2380
                    for k in 0..n.min(data.len() - at) {
2381
                        data[at + k] = 0xff;
2382
                    }
2383
                }
2384
            }
2385
            fs::write(path, &data).unwrap();
2386
            // Opening may refuse a broken index; a store that opens must
2387
            // either return each object intact or report it corrupt/missing.
2388
            if let Ok(s2) = Store::open(&objects) {
2389
                for r in &refs {
2390
                    match s2.get(r) {
75167672 Andy McCall 2391
                        Ok((ty, body)) => assert_eq!(
2392
                            hash_object(ty, &body),
2393
                            *r,
2394
                            "a returned object always matches its reference"
2395
                        ),
2396
                        Err(StoreError::Corrupt(..))
2397
                        | Err(StoreError::NotFound(_))
2398
                        | Err(StoreError::Io(_)) => {}
2b872d07 Andy McCall ↶ 2399
                        Err(e) => panic!("unexpected error {e}"),
2400
                    }
2401
                }
2402
            }
2403
            fs::write(path, pristine).unwrap();
2404
        }
2405
        // Restored files read back in full.
2406
        let s3 = Store::open(&objects).unwrap();
2407
        for r in &refs {
2408
            let (ty, body) = s3.get(r).unwrap();
2409
            assert_eq!(hash_object(ty, &body), *r);
2410
        }
2411
    }
8f10a07a Andy McCall ↶ 2412
2413
    /// A small batch becomes loose objects; a large one becomes a single
2414
    /// pack. Objects are readable while pending, and batches nest.
2415
    #[test]
2416
    fn batches_write_loose_or_one_pack() {
2417
        let dir = tempfile::tempdir().unwrap();
64818938 Andy McCall ↶ 2418
        each_store(dir.path(), |which, s| {
2419
            let outer = s.begin_batch().unwrap();
2420
            let a = s.put(ObjectType::Chunk, b"small a").unwrap();
2421
            {
2422
                let inner = s.begin_batch().unwrap();
2423
                let b = s.put(ObjectType::Chunk, b"small b").unwrap();
2424
                assert!(s.has(&b));
2425
                assert_eq!(s.get(&b).unwrap().1, b"small b");
2426
                inner.finish().unwrap();
2427
                assert_eq!(
2428
                    s.loose_count().unwrap(),
2429
                    0,
2430
                    "{which}: the inner batch defers to the outer one"
2431
                );
2432
            }
2433
            assert_eq!(s.get(&a).unwrap().1, b"small a");
75167672 Andy McCall 2434
            assert_eq!(
64818938 Andy McCall ↶ 2435
                s.put(ObjectType::Chunk, b"small a").unwrap(),
2436
                a,
2437
                "idempotent while pending"
75167672 Andy McCall 2438
            );
64818938 Andy McCall ↶ 2439
            outer.finish().unwrap();
2440
            assert_eq!(s.loose_count().unwrap(), 2);
2441
            assert_eq!(s.pack_count(), 0);
2442
            assert_eq!(s.get(&a).unwrap().1, b"small a");
8f10a07a Andy McCall ↶ 2443
64818938 Andy McCall ↶ 2444
            // Large: spills into one pack, readable during and after.
2445
            let big = s.begin_batch().unwrap();
2446
            let mut refs = Vec::new();
2447
            let mut seed = 7u64;
2448
            for i in 0..300u32 {
2449
                let mut body = vec![b'z'; 40_000];
2450
                if i % 3 == 0 {
2451
                    for b in body.iter_mut() {
2452
                        *b = (xorshift(&mut seed) & 0xff) as u8;
2453
                    }
8f10a07a Andy McCall ↶ 2454
                }
64818938 Andy McCall ↶ 2455
                body[..4].copy_from_slice(&i.to_le_bytes());
2456
                let r = if i % 5 == 0 {
2457
                    s.put_chunk_nocompress(&body).unwrap()
2458
                } else {
2459
                    s.put(ObjectType::Chunk, &body).unwrap()
2460
                };
2461
                assert_eq!(
2462
                    s.get(&r).unwrap().1,
2463
                    body,
2464
                    "readable while pending (object {i})"
2465
                );
2466
                refs.push((r, body));
8f10a07a Andy McCall ↶ 2467
            }
64818938 Andy McCall ↶ 2468
            assert!(s.has(&refs[0].0));
2469
            big.finish().unwrap();
2470
            assert_eq!(s.pack_count(), 1);
75167672 Andy McCall 2471
            assert_eq!(
64818938 Andy McCall ↶ 2472
                s.loose_count().unwrap(),
2473
                2,
2474
                "nothing from the big batch is loose"
75167672 Andy McCall 2475
            );
64818938 Andy McCall ↶ 2476
            assert!(fs::read_dir(s.dir())
75167672 Andy McCall 2477
                .unwrap()
64818938 Andy McCall ↶ 2478
                .flatten()
2479
                .all(|e| !e.file_name().to_string_lossy().contains("tmp-")));
2480
            let s2 = Store::open_on(s.backend().clone(), s.dir()).unwrap();
2481
            for (r, body) in &refs {
2482
                assert_eq!(s2.get(r).unwrap().1, *body);
2483
            }
2484
            // A dropped batch still writes its objects.
2485
            {
2486
                let _b = s2.begin_batch().unwrap();
2487
                s2.put(ObjectType::Chunk, b"dropped").unwrap();
2488
            }
2489
            assert_eq!(
2490
                s2.get(&hash_object(ObjectType::Chunk, b"dropped"))
2491
                    .unwrap()
2492
                    .1,
2493
                b"dropped"
2494
            );
2495
        });
afaba5f1 Andy McCall ↶ 2496
    }
2497
2498
    #[test]
2499
    fn get_many_reads_loose_and_packed_objects_in_order() {
2500
        let dir = tempfile::tempdir().unwrap();
64818938 Andy McCall ↶ 2501
        each_store(dir.path(), |which, s| {
2502
            let packed: Vec<ObjectRef> = (0..20)
2503
                .map(|i| {
2504
                    s.put(ObjectType::Chunk, format!("packed {i}").as_bytes())
2505
                        .unwrap()
2506
                })
2507
                .collect();
2508
            s.pack_loose().unwrap();
2509
            let loose: Vec<ObjectRef> = (0..20)
2510
                .map(|i| {
2511
                    s.put(ObjectType::Chunk, format!("loose {i}").as_bytes())
2512
                        .unwrap()
2513
                })
2514
                .collect();
2515
            let missing = hash_object(ObjectType::Chunk, b"never stored");
2516
            let mut refs: Vec<ObjectRef> = Vec::new();
2517
            for i in 0..20 {
2518
                refs.push(loose[i]);
2519
                refs.push(packed[i]);
2520
            }
2521
            refs.push(missing);
2522
            let got = s.get_many(&refs);
2523
            assert_eq!(got.len(), refs.len());
2524
            for (i, r) in refs.iter().enumerate().take(40) {
2525
                let want = if i % 2 == 0 {
2526
                    format!("loose {}", i / 2)
2527
                } else {
2528
                    format!("packed {}", i / 2)
2529
                };
2530
                let (ty, body) = got[i].as_ref().unwrap();
2531
                assert_eq!(
2532
                    (*ty, body.as_slice()),
2533
                    (ObjectType::Chunk, want.as_bytes()),
2534
                    "{which}: {r:?}"
2535
                );
2536
            }
2537
            assert!(matches!(got[40], Err(StoreError::NotFound(_))));
2538
        });
09245702 Andy McCall ↶ 2539
    }
2540
2541
    /// A packed object is read from its pack without first asking the
2542
    /// backend for a loose copy (oxygenhub#228: on a bucket each such ask
2543
    /// was a request that found nothing, every time).
2544
    #[test]
2545
    fn packed_objects_cost_no_loose_lookups() {
2546
        use std::sync::atomic::{AtomicUsize, Ordering};
2547
        struct Counting(MemoryStore, AtomicUsize);
2548
        impl crate::strata::Store for Counting {
2549
            fn describe(&self) -> String {
2550
                "counting".into()
2551
            }
2552
            fn has(&self, k: &str) -> crate::strata::Result<bool> {
2553
                self.0.has(k)
2554
            }
2555
            fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
2556
                if k.starts_with("loose/") {
2557
                    self.1.fetch_add(1, Ordering::Relaxed);
2558
                }
2559
                self.0.get(k)
2560
            }
2561
            fn get_range(
2562
                &self,
2563
                k: &str,
2564
                r: std::ops::Range<u64>,
2565
            ) -> crate::strata::Result<Vec<u8>> {
2566
                self.0.get_range(k, r)
2567
            }
2568
            fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
2569
                self.0.put_new(k, b)
2570
            }
2571
            fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
2572
                self.0.list(p)
2573
            }
2574
            fn delete(&self, k: &str) -> crate::strata::Result<()> {
2575
                self.0.delete(k)
2576
            }
2577
            fn read_ref(
2578
                &self,
2579
                k: &str,
2580
            ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
2581
                self.0.read_ref(k)
2582
            }
2583
            fn write_ref(
2584
                &self,
2585
                k: &str,
2586
                b: &[u8],
2587
                e: Option<&crate::strata::Version>,
2588
            ) -> crate::strata::Result<crate::strata::Version> {
2589
                self.0.write_ref(k, b, e)
2590
            }
2591
        }
2592
        let dir = tempfile::tempdir().unwrap();
2593
        let backend = Arc::new(Counting(MemoryStore::new(), AtomicUsize::new(0)));
2594
        let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap();
2595
        let refs: Vec<ObjectRef> = (0..10)
2596
            .map(|i| {
2597
                s.put(ObjectType::Chunk, format!("p {i}").as_bytes())
2598
                    .unwrap()
2599
            })
2600
            .collect();
2601
        s.pack_loose().unwrap();
2602
        let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap();
2603
        backend.1.store(0, Ordering::Relaxed);
2604
        for r in &refs {
2605
            assert!(s.get(r).is_ok());
2606
        }
2607
        assert!(s.get_many(&refs).iter().all(|g| g.is_ok()));
2608
        assert_eq!(backend.1.load(Ordering::Relaxed), 0, "no loose lookups");
2609
        // An object no pack holds is still looked for loose.
2610
        let missing = hash_object(ObjectType::Chunk, b"never stored");
2611
        assert!(matches!(s.get(&missing), Err(StoreError::NotFound(_))));
2612
        assert!(backend.1.load(Ordering::Relaxed) > 0);
a9f686ed Andy McCall ↶ 2613
    }
2614
}
2615
2616
#[cfg(test)]
2617
mod batch_tests {
2618
    use super::*;
2619
2620
    #[test]
2621
    fn a_large_batch_packs_every_object_and_indexes_them_all() {
2622
        let dir = tempfile::tempdir().unwrap();
2623
        let store = Store::open(dir.path()).unwrap();
2624
        let mut refs = Vec::new();
2625
        {
2626
            let batch = store.begin_batch().unwrap();
2627
            for i in 0..1000u32 {
2628
                let body = format!("object {i} {}", "x".repeat(20_000 + i as usize)).into_bytes();
2629
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
2630
            }
2631
            batch.finish().unwrap();
2632
        }
2633
        let again = Store::open(dir.path()).unwrap();
2634
        for r in &refs {
2635
            again
54e2e630 Andy McCall ↶ 2636
                .get(r)
2637
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
2638
        }
2639
    }
2640
2641
    #[test]
2642
    fn a_pack_a_stopped_batch_left_is_adopted_and_its_torn_end_cut() {
2643
        // oxygen#79: a batch never finished (a process killed part way)
2644
        // leaves its objects in an unindexed .tmp pack.
2645
        let dir = tempfile::tempdir().unwrap();
2646
        let mut refs = Vec::new();
2647
        {
2648
            let store = Store::open(dir.path()).unwrap();
2649
            let batch = store.begin_batch().unwrap();
2650
            for i in 0..300u32 {
2651
                let body = format!("object {i} {}", "y".repeat(20_000 + i as usize)).into_bytes();
2652
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
2653
            }
2654
            std::mem::forget(batch);
2655
        }
2656
        let tmp: Vec<_> = fs::read_dir(dir.path())
2657
            .unwrap()
2658
            .flatten()
2659
            .filter(|e| e.file_name().to_string_lossy().starts_with(".tmp-"))
2660
            .map(|e| e.path())
2661
            .collect();
2662
        assert_eq!(tmp.len(), 1, "one unfinished pack");
2663
        // A torn last entry: a header and part of its body.
2664
        let mut f = fs::OpenOptions::new().append(true).open(&tmp[0]).unwrap();
2665
        f.write_all(&[0, 1, 0xff, 0, 0, 0, 9, 9, 0, 0]).unwrap();
2666
        f.write_all(&[7; 40]).unwrap();
2667
        drop(f);
2668
        let store = Store::open(dir.path()).unwrap();
2669
        assert!(store.get(&refs[0]).is_err(), "unreadable before");
2670
        let young = store
2671
            .adopt_unfinished_packs(std::time::Duration::from_secs(3600), &|_| {})
2672
            .unwrap();
2673
        assert_eq!(young, (0, 0), "a pack written to lately is left alone");
2674
        let (packs, objects) = store
2675
            .adopt_unfinished_packs(std::time::Duration::ZERO, &|_| {})
2676
            .unwrap();
2677
        assert_eq!((packs, objects), (1, 300));
2678
        for r in &refs {
2679
            store
a9f686ed Andy McCall ↶ 2680
                .get(r)
2681
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
2682
        }
54e2e630 Andy McCall ↶ 2683
        let again = Store::open(dir.path()).unwrap();
2684
        assert!(again.get(&refs[299]).is_ok(), "indexed for good");
a9f686ed Andy McCall ↶ 2685
    }
2686
}
2687
2688
#[cfg(test)]
2689
mod size_tests {
2690
    use super::*;
2691
2692
    #[test]
2693
    fn an_object_over_the_limit_is_refused_and_nothing_is_written() {
2694
        let dir = tempfile::tempdir().unwrap();
2695
        let store = Store::open(dir.path()).unwrap();
2696
        let big = vec![b'x'; MAX_OBJECT_LEN as usize + 1];
2697
        {
2698
            let batch = store.begin_batch().unwrap();
2699
            store.put(ObjectType::Blob, b"small").unwrap();
2700
            let e = store.put(ObjectType::Change, &big).unwrap_err();
2701
            assert!(
2702
                matches!(e, StoreError::TooLarge { kind: "change", .. }),
2703
                "{e}"
2704
            );
2705
            assert!(e.to_string().contains("commit it in parts"), "{e}");
2706
            batch.finish().unwrap();
2707
        }
2708
        // The store still opens and reads what was written.
2709
        let again = Store::open(dir.path()).unwrap();
2710
        again.get(&hash_object(ObjectType::Blob, b"small")).unwrap();
6f8aeb99 Andy McCall ↶ 2711
    }
2712
}
2713
2714
#[cfg(test)]
2715
mod text_tests {
2716
    use super::*;
2717
    use crate::objects::{AtomRef, Edit, CHANGE_VERSION};
2718
2719
    fn large_change() -> Change {
2720
        let mut c = crate::objects::tests::sample_change();
2721
        assert_eq!(c.version, CHANGE_VERSION);
2722
        c.edits[0] = Edit::Insert {
2723
            node: crate::ids::NodeId([7; 16]),
2724
            after: AtomRef::Start,
2725
            before: AtomRef::End,
2726
            lines: (0..2000)
2727
                .map(|i| format!("line {i}\n").into_bytes())
2728
                .collect(),
2729
        };
2730
        c
2731
    }
2732
2733
    #[test]
2734
    fn a_large_insert_is_stored_beside_its_change_and_read_back() {
2735
        let dir = tempfile::tempdir().unwrap();
2736
        let s = Store::open(dir.path()).unwrap();
2737
        let c = large_change();
2738
        let r = s.put_change(&c).unwrap();
2739
        assert_eq!(r, c.hash());
2740
        let (_, body) = s.get(&r).unwrap();
2741
        assert!(
2742
            body.len() < 1024,
2743
            "the change holds a reference, not {} bytes of lines",
2744
            body.len()
2745
        );
2746
        let (back, parts) = s.get_change_parts(&r).unwrap();
2747
        assert_eq!(back, c);
2748
        // The text blob and its chunk.
2749
        assert_eq!(parts.len(), 2);
2750
        for p in &parts {
2751
            assert!(s.has(p));
2752
        }
2753
        // Storing it again writes nothing new.
2754
        let n = s.list().unwrap().len();
2755
        s.put_change(&c).unwrap();
2756
        assert_eq!(s.list().unwrap().len(), n);
2757
    }
2758
2759
    #[test]
2760
    fn a_change_whose_text_blob_is_missing_is_reported_not_misread() {
64818938 Andy McCall ↶ 2761
        if !super::files_on_disk() {
2762
            return;
2763
        }
6f8aeb99 Andy McCall ↶ 2764
        let dir = tempfile::tempdir().unwrap();
2765
        let s = Store::open(dir.path()).unwrap();
2766
        let c = large_change();
2767
        let r = s.put_change(&c).unwrap();
2768
        let (_, parts) = s.get_change_parts(&r).unwrap();
2769
        fs::remove_file(s.path_for(&parts[0])).unwrap();
2770
        let e = s.get_change(&r).unwrap_err();
2771
        assert!(
2772
            matches!(&e, StoreError::Corrupt(h, why) if *h == r && why.contains("missing")),
2773
            "{e}"
2774
        );
192fad1e Andy McCall ↶ 2775
    }
2776
}
2777
2778
#[cfg(test)]
2779
mod put_many_tests {
2780
    use super::*;
2781
2782
    #[test]
2783
    fn many_objects_in_a_batch_are_compressed_together_and_all_read_back() {
2784
        let dir = tempfile::tempdir().unwrap();
2785
        let s = Store::open(dir.path()).unwrap();
2786
        let before = s.put(ObjectType::Chunk, b"already here").unwrap();
2787
        let bodies: Vec<Vec<u8>> = (0..200u32)
2788
            .map(|i| format!("chunk {i} ").repeat(6000).into_bytes())
2789
            .collect();
2790
        let mut objects: Vec<(ObjectType, &[u8])> = bodies
2791
            .iter()
2792
            .map(|b| (ObjectType::Chunk, b.as_slice()))
2793
            .collect();
2794
        objects.push((ObjectType::Chunk, bodies[0].as_slice()));
2795
        objects.push((ObjectType::Chunk, b"already here"));
2796
        let refs = {
2797
            let batch = s.begin_batch().unwrap();
2798
            let refs = s.put_many(&objects).unwrap();
2799
            batch.finish().unwrap();
2800
            refs
2801
        };
2802
        assert_eq!(refs.len(), objects.len());
2803
        assert_eq!(refs[200], refs[0], "a repeat is the same object");
2804
        assert_eq!(refs[201], before);
2805
        let again = Store::open(dir.path()).unwrap();
2806
        for (r, b) in refs.iter().zip(&objects) {
2807
            assert_eq!(again.get(r).unwrap().1, b.1);
2808
        }
8f10a07a Andy McCall ↶ 2809
    }
ff0af1ee unknown 2810
}