Signatures and index entries from a journal and entry lists, not a LIST per lookup ef7d4ea6f33b

- **A repository kept in a bucket finds signatures and seals without a LIST each (#116).** Every lookup of an object's signatures, seals, receipts, approvals or certificate listed a prefix in the bucket, and most of these lookups found nothing. Writers now also add each such entry to a journal (`pack-entries`, updated by compare-and-swap), and `oxy gc` folds them all into one list. Once a store has been folded, a reader reads the lists and the journal once per open and looks everything up from them: 40 lookups went from 40 LISTs to one journal read. Older readers are unaffected, because every entry still has its own key. Entries written by oxy 0.4.6 or older after the fold are missed until the next `oxy gc`, so upgrade every client that writes to the bucket first, then run `oxy gc` once. A server's `/health` counts journal reads as `read_journal`. A repository's own directory works as before.

    /// turn, for the next collection. A store that is not this disk's

    /// directory then folds its signatures and index entries (§13.7).

        // Last, on a store that is not this disk's directory, every

        // signature and index entry into one list (spec §13.7, #116).

        self.store.fold_entries(progress)?;

//!

//! A store that is not this disk's directory, where finding an object's

//! entries is a LIST request, also keeps every entry key in the

//! `pack-entries` journal and, once a collection folds them, in entry

//! lists (`pack-<id>.entries`), read once per open (spec §13.7, #116).

/// The journal of signatures and index entries (spec §13.7): every entry

/// key a writer on a store that is not this disk's directory wrote since

/// the last fold, one line each, `<folded at, wall ms; 0 until folded>

/// <key>`, as the retired record's lines are. A reference record updated

/// by compare-and-swap, so concurrent writers never lose each other's

/// lines; read once per open, instead of a listing per lookup.

pub const JOURNAL_KEY: &str = "pack-entries";

/// A complete entry list (spec §13.7): every entry a fold found.

const LIST_SUFFIX: &str = ".entries";

/// A partial entry list: journal lines a writer folded out.

const PART_SUFFIX: &str = ".entries-part";

/// The first line of an entry list.

const LIST_MAGIC: &str = "oxy-entries 1";

/// Written once a collection has copied the older one-list keys into

/// per-object entries: readers stop looking for them.

const LEGACY_FOLDED_KEY: &str = "pack-legacy-folded";

/// Unfolded journal lines a writer lets the journal hold before it folds

/// them into a partial list.

const JOURNAL_FOLD_AT: usize = 1000;

/// The per-object entry indexes, each with the one-list name older

/// stores kept it under.

const ENTRY_INDEXES: [(&str, &str); 6] = [

    ("sig", "sigs"),

    ("seal", "seals"),

    ("receipt", "receipts"),

    ("approval", "approvals"),

    ("cert", "certs"),

    ("accepted", "accepted-lists"),

];

}

/// Whether a key names an entry list (spec §13.7), complete or partial.

fn is_entry_list(key: &str) -> bool {

    key.starts_with("pack-") && (key.ends_with(LIST_SUFFIX) || key.ends_with(PART_SUFFIX))

}

/// An entry list's bytes: zstd-compressed text, [`LIST_MAGIC`] and then

/// one key per line, sorted.

fn encode_entry_list<'a>(keys: impl Iterator<Item = &'a String>) -> Result<Vec<u8>> {

    let mut text = String::from(LIST_MAGIC);

    text.push('\n');

    for k in keys {

        text.push_str(k);

        text.push('\n');

    }

    Ok(zstd::bulk::compress(text.as_bytes(), ZSTD_LEVEL)?)

}

/// The keys an entry list holds.

fn decode_entry_list(key: &str, bytes: &[u8]) -> Result<Vec<String>> {

    let bad = |why: &str| StoreError::Io(io::Error::other(format!("{key}: {why}")));

    let text = zstd::stream::decode_all(bytes).map_err(|_| bad("not zstd"))?;

    let text = String::from_utf8(text).map_err(|_| bad("not UTF-8"))?;

    let mut lines = text.lines();

    if lines.next() != Some(LIST_MAGIC) {

        return Err(bad("not an entry list"));

    }

    Ok(lines

        .filter(|l| !l.is_empty())

        .map(str::to_string)

        .collect())

/// Drop journal lines folded at least [`RETIRED_FOR`] before `now`:

/// every reader that could need them has a list holding them by then.

fn expire_folded(m: &mut std::collections::BTreeMap<String, u64>, now: u64) {

    let hour = RETIRED_FOR.as_millis() as u64;

    m.retain(|_, t| *t == 0 || now.saturating_sub(*t) < hour);

}

    /// The live entry lists (spec §13.7) the pack listing showed: listed

    /// and not retired.

    entry_lists: RefCell<Vec<String>>,

    /// Whether the store is folded: a live entry list is complete.

    folded: std::cell::Cell<bool>,

    /// Whether the pack listing showed `pack-legacy-folded`.

    legacy_folded: std::cell::Cell<bool>,

    /// The entry keys of a folded store, read from its lists and journal

    /// on the first lookup after the packs were loaded.

    entries: RefCell<Option<std::collections::BTreeSet<String>>>,

    /// Entry keys written in an open batch, added to the journal when the

    /// batch is flushed.

    journal_queue: RefCell<Vec<String>>,

            entry_lists: RefCell::new(Vec::new()),

            folded: std::cell::Cell::new(false),

            legacy_folded: std::cell::Cell::new(false),

            entries: RefCell::new(None),

            journal_queue: RefCell::new(Vec::new()),

        *self.entries.borrow_mut() = None;

            let lists: Vec<String> = listed

                .iter()

                .map(|e| e.key.as_str())

                .filter(|k| is_entry_list(k) && !retired.contains_key(*k))

                .map(str::to_string)

                .collect();

            self.folded

                .set(lists.iter().any(|k| k.ends_with(LIST_SUFFIX)));

            self.legacy_folded

                .set(listed.iter().any(|e| e.key == LEGACY_FOLDED_KEY));

            *self.entry_lists.borrow_mut() = lists;

            if *self.packs_listed.borrow() != sig {

                *self.entries.borrow_mut() = None;

            }

        &self,

    ) -> Result<(

        std::collections::BTreeMap<String, u64>,

        Option<strata::Version>,

    )> {

        self.read_record(RETIRED_KEY)

    }

    /// A record of `<wall ms> <key>` lines (the retired record, the

    /// journal) and its version (`None`: never written).

    fn read_record(

        key: &str,

        match self.backend.read_ref(key).map_err(backend_err)? {

        &self,

        edit: impl FnMut(&mut std::collections::BTreeMap<String, u64>),

    ) -> Result<()> {

        self.update_record(RETIRED_KEY, "the record of retired packs", edit)

    }

    /// Change a record of `<wall ms> <key>` lines by compare-and-swap, as

    /// [`Store::update_retired`] does; `what` names it in the error when

    /// other writers kept winning.

    fn update_record(

        key: &str,

        what: &str,

            let (mut m, version) = self.read_record(key)?;

            // Unchanged; a record never written stays unwritten.

            if m == before && (version.is_some() || m.is_empty()) {

                .write_ref(key, &encode_retired(&m), version.as_ref())

        Err(StoreError::Io(io::Error::other(format!(

            "other writers kept changing {what}; run the command again"

        ))))

            } else if is_entry_list(key) {

                // An entry list a fold replaced (spec §13.7).

                delete(self.backend.as_ref(), key)?;

        let mut gone: Vec<String> = Vec::new();

            self.remove_sidecars(&r, &mut gone)?;

                self.remove_sidecars(&r, &mut gone)?;

        self.journal_forget(&gone)?;

    /// The entry keys deleted are added to `gone`, for the journal.

    fn remove_sidecars(&self, r: &ObjectRef, gone: &mut Vec<String>) -> Result<()> {

                gone.push(e.key);

    /// Take entry keys a collection deleted out of the journal, so the

    /// next complete fold leaves them out (spec §13.7).

    fn journal_forget(&self, gone: &[String]) -> Result<()> {

        if self.local || gone.is_empty() {

            return Ok(());

        }

        if let Some(set) = self.entries.borrow_mut().as_mut() {

            for k in gone {

                set.remove(k);

            }

        }

        self.update_record(JOURNAL_KEY, "the journal of entries", |m| {

            for k in gone {

                m.remove(k);

            }

        })

    }

            return self.flush_journal();

        self.write_pending(pending)?;

        self.flush_journal()

            return self.flush_journal();

        self.flush_journal()?;

        debug_assert!(ENTRY_INDEXES.iter().any(|(i, _)| *i == index));

        put_new_or_existing(self.backend.as_ref(), &entry, b"")?;

        self.journal_add(entry)

            .entry_keys(&fanout_prefix(index, key))?

            .filter_map(|k| k.rsplit('/').next().and_then(ObjectRef::from_hex))

        if !self.reads_legacy() {

            // Older lists folded into entries (spec §13.7).

        } else if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key(legacy, key))? {

        for key in self.entry_keys(&format!("{one}/"))? {

            let mut parts = key.splitn(4, '/');

        let legacy = if self.reads_legacy() {

            self.backend

        } else {

            Vec::new()

        };

        for e in legacy {

        for key in self.entry_keys(&fanout_prefix("sig", r))? {

            // Gone since it was listed or journalled: a collection deleted

            // it with its object, or a new signature by the same signer is

            // replacing it.

            let Some(bytes) = get_optional(self.backend.as_ref(), &key)? else {

                continue;

            };

        let legacy = if self.reads_legacy() {

            get_optional(self.backend.as_ref(), &legacy_key("sigs", r))?

        } else {

            None

        };

        if let Some(bytes) = legacy {

        if get_optional(self.backend.as_ref(), &key)?.as_deref() != Some(bytes.as_slice()) {

            delete(self.backend.as_ref(), &key)?;

            put_new_or_existing(self.backend.as_ref(), &key, &bytes)?;

        }

        // Journalled even when the key was there: an older writer's

        // entry is not in the journal or any list.

        self.journal_add(key)

    }

    // ---- the journal and entry lists (spec §13.7, #116)

    /// Whether lookups still read the one-list keys of older stores.

    fn reads_legacy(&self) -> bool {

        true

    }

    /// The entry keys under `prefix` (an index's `<index>/` or one

    /// object's `<index>/<hh>/<rest>/`), sorted: on a folded store from

    /// its lists and journal, with no request; otherwise by listing.

    fn entry_keys(&self, prefix: &str) -> Result<Vec<String>> {

        if let Some(keys) = self.with_entries(|set| {

            set.range(prefix.to_string()..)

                .take_while(|k| k.starts_with(prefix))

                .cloned()

                .collect::<Vec<String>>()

        })? {

            return Ok(keys);

        }

        Ok(self

            .backend

            .list(prefix)

            .map_err(backend_err)?

            .into_iter()

            .map(|e| e.key)

            .collect())

    }

    /// `f` over a folded store's entry keys, read on first use; `None`

    /// when the store is not folded (or is this disk's directory), and

    /// lookups list.

    fn with_entries<T>(

        &self,

        f: impl FnOnce(&std::collections::BTreeSet<String>) -> T,

    ) -> Result<Option<T>> {

        if self.local || !self.folded.get() {

            return Ok(None);

        }

        if self.entries.borrow().is_none() {

            let Some(set) = self.read_entries()? else {

                return Ok(None);

            };

            *self.entries.borrow_mut() = Some(set);

        }

        Ok(self.entries.borrow().as_ref().map(f))

    }

    /// Every live entry list's keys and every journal line's, folded or

    /// not, with what this handle has queued. A list deleted since the

    /// packs were listed (retired over an hour ago) has the packs listed

    /// again; `None` when one is still missing then, or the store is no

    /// longer folded: lookups list.

    fn read_entries(&self) -> Result<Option<std::collections::BTreeSet<String>>> {

        'attempt: for attempt in 0..2 {

            if !self.folded.get() {

                return Ok(None);

            }

            let mut set = std::collections::BTreeSet::new();

            let lists = self.entry_lists.borrow().clone();

            for key in &lists {

                match get_optional(self.backend.as_ref(), key)? {

                    Some(bytes) => set.extend(decode_entry_list(key, &bytes)?),

                    None if attempt == 0 => {

                        self.reload_packs()?;

                        continue 'attempt;

                    }

                    None => return Ok(None),

                }

            }

            set.extend(self.read_record(JOURNAL_KEY)?.0.into_keys());

            set.extend(self.journal_queue.borrow().iter().cloned());

            return Ok(Some(set));

        }

        Ok(None)

    }

    /// Add an entry key this handle wrote to the journal: at once, or

    /// when the open batch is flushed. One the handle already knows from

    /// the lists or the journal is not added again. Nothing on this

    /// disk's directory.

    fn journal_add(&self, key: String) -> Result<()> {

        if self.local {

            return Ok(());

        }

        if let Some(set) = self.entries.borrow_mut().as_mut() {

            if !set.insert(key.clone()) {

                return Ok(());

            }

        }

        if self.batch_depth.get() > 0 {

            self.journal_queue.borrow_mut().push(key);

            return Ok(());

        }

        self.journal_append(std::slice::from_ref(&key))

    }

    /// Add what the open batch queued to the journal.

    fn flush_journal(&self) -> Result<()> {

        let queued = std::mem::take(&mut *self.journal_queue.borrow_mut());

        self.journal_append(&queued)

    }

    /// Add `keys` to the journal, unfolded, in one compare-and-swap

    /// update, dropping lines folded over an hour ago; past

    /// [`JOURNAL_FOLD_AT`] unfolded lines, fold them into a partial list.

    fn journal_append(&self, keys: &[String]) -> Result<()> {

        if keys.is_empty() {

        }

        let now = now_ms();

        let mut unfolded = 0;

        self.update_record(JOURNAL_KEY, "the journal of entries", |m| {

            expire_folded(m, now);

            for k in keys {

                m.entry(k.clone()).or_insert(0);

            }

            unfolded = m.values().filter(|t| **t == 0).count();

        })?;

        if unfolded > JOURNAL_FOLD_AT {

            self.fold_journal()?;

        }

        Ok(())

    }

    /// Fold the journal's unfolded lines into a partial entry list, and

    /// mark them folded. Lines added meanwhile stay unfolded.

    fn fold_journal(&self) -> Result<()> {

        let (journal, _) = self.read_record(JOURNAL_KEY)?;

        let keys: Vec<String> = journal

            .into_iter()

            .filter(|(_, t)| *t == 0)

            .map(|(k, _)| k)

            .collect();

        if keys.is_empty() {

            return Ok(());

        }

        self.put_entry_list(PART_SUFFIX, keys.iter())?;

        self.mark_folded(&keys)

    }

    /// Mark these journal lines folded now, if they are not yet; drop

    /// those folded over an hour ago.

    fn mark_folded(&self, keys: &[String]) -> Result<()> {

        let now = now_ms();

        self.update_record(JOURNAL_KEY, "the journal of entries", |m| {

            expire_folded(m, now);

            for k in keys {

                if let Some(t) = m.get_mut(k) {

                    if *t == 0 {

                        *t = now;

                    }

                }

            }

        })

    }

    /// Write a new entry list of `keys` (sorted), complete or partial by

    /// `suffix`; its key.

    fn put_entry_list<'a>(

        &self,

        suffix: &str,

        keys: impl Iterator<Item = &'a String>,

    ) -> Result<String> {

        let key = format!("pack-{}{suffix}", fresh_id());

        let bytes = encode_entry_list(keys)?;

        self.backend.put_new(&key, &bytes).map_err(backend_err)?;

        Ok(key)

    }

    /// Fold every signature and index entry into one complete entry list

    /// (spec §13.7), what a collection does last: the journal's lines and

    /// a listing of every entry prefix, written as a new list, every list

    /// the packs' listing showed retired, and the journal lines read

    /// marked folded. Nothing when the store is this disk's directory, or

    /// when the one live list already holds exactly these entries and the

    /// journal has nothing unfolded. Whether a list was written.

    pub fn fold_entries(&self, progress: Progress) -> Result<bool> {

        if self.local {

            return Ok(false);

        self.reload_packs()?;

        let (journal, _) = self.read_record(JOURNAL_KEY)?;

        progress("folding signatures and index entries");

        let mut all: std::collections::BTreeSet<String> = journal.keys().cloned().collect();

        for (index, _) in ENTRY_INDEXES {

            for e in self

                .backend

                .list(&format!("{index}/"))

                .map_err(backend_err)?

            {

                all.insert(e.key);

            }

        }

        let lists = self.entry_lists.borrow().clone();

        let unfolded: Vec<String> = journal

            .iter()

            .filter(|(_, t)| **t == 0)

            .map(|(k, _)| k.clone())

            .collect();

        if unfolded.is_empty() && lists.len() == 1 && lists[0].ends_with(LIST_SUFFIX) {

            if let Some(bytes) = get_optional(self.backend.as_ref(), &lists[0])? {

                let held: std::collections::BTreeSet<String> =

                    decode_entry_list(&lists[0], &bytes)?.into_iter().collect();

                if held == all {

                    return Ok(false);

                }

            }

        }

        self.put_entry_list(LIST_SUFFIX, all.iter())?;

        self.retire(&lists)?;

        if !unfolded.is_empty() {

            self.mark_folded(&unfolded)?;

        }

        *self.entries.borrow_mut() = None;

        self.reload_packs()?;

        Ok(true)

        }

    }

}

/// Signatures and index entries on a store that is not this disk's

/// directory: the journal and the entry lists (spec §13.7, #116).

#[cfg(test)]

mod entry_tests {

    use super::*;

    use crate::identity::{Domain, SecretKey};

    use crate::strata::Store as _;

    use std::collections::HashMap;

    use std::sync::atomic::{AtomicBool, Ordering};

    use std::sync::Mutex;

    /// A store in memory that counts its requests by kind, as a bucket

    /// store's /health counters do; it can also slip a competing journal

    /// line in just before a write of the journal, as another process

    /// would.

    #[derive(Default)]

    struct Counted {

        inner: MemoryStore,

        counts: Mutex<HashMap<&'static str, usize>>,

        interfere: AtomicBool,

    }

    impl Counted {

        fn count(&self, kind: &'static str) {

            *self.counts.lock().unwrap().entry(kind).or_default() += 1;

        }

        fn n(&self, kind: &str) -> usize {

            self.counts.lock().unwrap().get(kind).copied().unwrap_or(0)

        fn reset(&self) {

            self.counts.lock().unwrap().clear();

        }

        fn total(&self) -> usize {

            self.counts.lock().unwrap().values().sum()

        }

        fn summary(&self) -> String {

            let c = self.counts.lock().unwrap();

            let mut v: Vec<String> = c.iter().map(|(k, n)| format!("{k} {n}")).collect();

            v.sort();

            v.join(", ")

        }

    }

    impl crate::strata::Store for Counted {

        fn describe(&self) -> String {

            "counted".into()

        }

        fn has(&self, k: &str) -> crate::strata::Result<bool> {

            self.count("has");

            self.inner.has(k)

        }

        fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {

            self.count(if is_legacy(k) { "get_legacy" } else { "get" });

            self.inner.get(k)

        }

        fn get_range(&self, k: &str, r: std::ops::Range<u64>) -> crate::strata::Result<Vec<u8>> {

            self.count("range");

            self.inner.get_range(k, r)

        }

        fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {

            self.count("put");

            self.inner.put_new(k, b)

        }

        fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {

            self.count(if p == "pack-" {

                "list_packs"

            } else if is_legacy(p) {

                "list_legacy"

            } else {

                "list"

            });

            self.inner.list(p)

        }

        fn delete(&self, k: &str) -> crate::strata::Result<()> {

            self.count("delete");

            self.inner.delete(k)

        }

        fn read_ref(

            &self,

            k: &str,

        ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {

            self.count(if k == JOURNAL_KEY {

                "read_journal"

            } else {

                "read_ref"

            });

            self.inner.read_ref(k)

        }

        fn write_ref(

            &self,

            k: &str,

            b: &[u8],

            e: Option<&crate::strata::Version>,

        ) -> crate::strata::Result<crate::strata::Version> {

            self.count("write_ref");

            if k == JOURNAL_KEY && self.interfere.swap(false, Ordering::SeqCst) {

                let (mut m, v) = match self.inner.read_ref(k)? {

                    Some((bytes, v)) => (parse_retired(&bytes), Some(v)),

                    None => (Default::default(), None),

                };

                m.insert(

                    format!("seal/{}", "c".repeat(64)) + "/" + &"d".repeat(64),

                    0,

                );

                self.inner.write_ref(k, &encode_retired(&m), v.as_ref())?;

            }

            self.inner.write_ref(k, b, e)

        }

    }

    /// Whether a key or prefix is under an older one-list layout.

    fn is_legacy(k: &str) -> bool {

        ENTRY_INDEXES

            .iter()

            .any(|(_, legacy)| k.starts_with(&format!("{legacy}/")))

    }

    struct Fixture {

        backend: Arc<Counted>,

        dir: tempfile::TempDir,

        handles: std::cell::Cell<usize>,

    }

    impl Fixture {

        fn new() -> Fixture {

            Fixture {

                backend: Arc::new(Counted::default()),

                dir: tempfile::tempdir().unwrap(),

                handles: Default::default(),

            }

        }

        /// Another handle on the store, as another process opens it.

        fn open(&self) -> Store {

            let n = self.handles.get() + 1;

            self.handles.set(n);

            Store::open_on(self.backend.clone(), self.dir.path().join(format!("s{n}"))).unwrap()

        }

    }

    fn signer(n: u8) -> SecretKey {

        SecretKey {

            ed_seed: [n; 32],

            ml_seed: [n + 100; 32],

        }

    }

    fn sign(s: &Store, r: &ObjectRef, n: u8) -> Signature {

        let sk = signer(n);

        let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap();

        let sig = sk.sign(Domain::Change, &key, r);

        s.put_signature(r, &sig).unwrap();

        sig

    }

    fn journal(s: &Store) -> std::collections::BTreeMap<String, u64> {

        s.read_record(JOURNAL_KEY).unwrap().0

    }

    fn fold(s: &Store) -> bool {

        s.fold_entries(&mut silent()).unwrap()

    }

    /// What 0.4.6 writes for a signature: its own key, and nothing else.

    fn old_writer_signs(f: &Fixture, r: &ObjectRef, n: u8) -> Signature {

        let sk = signer(n);

        let sig = sk.sign(Domain::Change, &ObjectRef([n; 32]), r);

        f.backend

            .put_new(

                &format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex()),

                &Signature::encode_list(std::slice::from_ref(&sig)),

            )

            .unwrap();

        sig

    }

    /// After a fold, a signature or seal another handle writes is found

    /// by a store opened later through the journal, with no listing; a

    /// handle that already read the entries sees it once refreshed.

    #[test]

    fn an_entry_written_after_a_fold_is_found_through_the_journal() {

        let f = Fixture::new();

        let w = f.open();

        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();

        let s1 = sign(&w, &r1, 1);

        assert!(fold(&w));

        assert!(!fold(&w), "nothing new: no second list");

        let early = f.open();

        assert_eq!(early.signatures(&r1).unwrap(), vec![s1.clone()]);

        let w2 = f.open();

        let r2 = w2.put(ObjectType::Chunk, b"two").unwrap();

        let s2 = sign(&w2, &r2, 2);

        w2.index_seal(&r2, &ObjectRef([9; 32])).unwrap();

        w2.index_receipt(&r2, &ObjectRef([8; 32])).unwrap();

        // Its own writes, at once.

        assert_eq!(w2.signatures(&r2).unwrap(), vec![s2.clone()]);

        let later = f.open();

        f.backend.reset();

        assert_eq!(later.signatures(&r1).unwrap(), vec![s1]);

        assert_eq!(later.signatures(&r2).unwrap(), vec![s2.clone()]);

        assert_eq!(later.seals_for(&r2).unwrap(), vec![ObjectRef([9; 32])]);

        assert_eq!(later.receipts_for(&r2).unwrap(), vec![ObjectRef([8; 32])]);

        assert_eq!(later.receipt_states().unwrap(), vec![r2]);

        assert!(later.seals_for(&r1).unwrap().is_empty());

        assert_eq!(f.backend.n("list"), 0, "{}", f.backend.summary());

        assert_eq!(f.backend.n("read_journal"), 1, "{}", f.backend.summary());

        // The handle that read the entries before keeps its view until

        // it is refreshed, as it keeps its operation head.

        assert!(early.signatures(&r2).unwrap().is_empty());

        early.refresh().unwrap();

        assert_eq!(early.signatures(&r2).unwrap(), vec![s2]);

    }

    /// A fold marks the journal's lines rather than dropping them, so a

    /// reader whose listing predates it still finds every entry; an hour

    /// on, the old list is deleted and the lines dropped, and readers

    /// find everything in the new list.

    #[test]

    fn a_fold_keeps_the_journal_for_the_hour_readers_may_need_it() {

        let f = Fixture::new();

        let w = f.open();

        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();

        let s1 = sign(&w, &r1, 1);

        assert!(fold(&w));

        let r2 = w.put(ObjectType::Chunk, b"two").unwrap();

        let s2 = sign(&w, &r2, 2);

        assert!(journal(&w).values().any(|t| *t == 0));

        // Opened before the second fold: lists the first list only.

        let stale = f.open();

        let stale_after_expiry = f.open();

        assert!(fold(&w));

        assert!(

            journal(&w).values().all(|t| *t != 0),

            "every line is marked folded"

        );

        assert!(journal(&w).len() >= 2, "and kept");

        assert_eq!(stale.signatures(&r1).unwrap(), vec![s1.clone()]);

        assert_eq!(stale.signatures(&r2).unwrap(), vec![s2.clone()]);

        // An hour later: the first list goes with the next collection,

        // and the next journal update drops the folded lines.

        reader_tests::age_retired(&w, RETIRED_FOR * 2);

        w.update_record(JOURNAL_KEY, "test", |m| {

            for t in m.values_mut() {

                *t = t.saturating_sub(2 * RETIRED_FOR.as_millis() as u64);

            }

        })

        .unwrap();

        w.expire_retired(RETIRED_FOR, &mut silent()).unwrap();

        let lists: Vec<String> = f

            .backend

            .inner

            .list("pack-")

            .unwrap()

            .into_iter()

            .map(|e| e.key)

            .filter(|k| is_entry_list(k))

            .collect();

        assert_eq!(lists.len(), 1, "the replaced list is deleted: {lists:?}");

        let r3 = w.put(ObjectType::Chunk, b"three").unwrap();

        let s3 = sign(&w, &r3, 3);

        let j = journal(&w);

        assert_eq!(j.len(), 1, "only the new line is left: {j:?}");

        assert!(j.keys().all(|k| k.starts_with(&fanout_prefix("sig", &r3))));

        // A handle whose listing named the deleted list lists again.

        assert_eq!(

            stale_after_expiry.signatures(&r1).unwrap(),

            vec![s1.clone()]

        );

        assert_eq!(

            stale_after_expiry.signatures(&r3).unwrap(),

            vec![s3.clone()]

        );

        let fresh = f.open();

        f.backend.reset();

        assert_eq!(fresh.signatures(&r1).unwrap(), vec![s1]);

        assert_eq!(fresh.signatures(&r2).unwrap(), vec![s2]);

        assert_eq!(fresh.signatures(&r3).unwrap(), vec![s3]);

        assert_eq!(f.backend.n("list"), 0);

    }

    /// A store no fold has touched (as 0.4.6 left it: per-object keys and

    /// older one-list keys, no journal, no lists) reads by listing, and

    /// reads the same once folded. An older writer's entry after the

    /// fold is missed until the next fold: the trade-off of §13.7.

    #[test]

    fn an_old_layout_store_reads_before_and_after_its_first_fold() {

        let f = Fixture::new();

        let w = f.open();

        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();

        let r2 = w.put(ObjectType::Chunk, b"two").unwrap();

        let s1 = old_writer_signs(&f, &r1, 1);

        let s2 = SecretKey {

            ed_seed: [2; 32],

            ml_seed: [3; 32],

        }

        .sign(Domain::Change, &ObjectRef([2; 32]), &r2);

        f.backend

            .put_new(

                &legacy_key("sigs", &r2),

                &Signature::encode_list(std::slice::from_ref(&s2)),

            )

            .unwrap();

        f.backend

            .put_new(

                &format!("{}{}", fanout_prefix("seal", &r1), "ab".repeat(32)),

                b"",

            )

            .unwrap();

        f.backend

            .put_new(

                &legacy_key("receipts", &r2),

                format!("{}\n", ObjectRef([7; 32]).to_hex()).as_bytes(),

            )

            .unwrap();

        let check = |s: &Store| {

            assert_eq!(s.signatures(&r1).unwrap(), vec![s1.clone()]);

            assert_eq!(s.signatures(&r2).unwrap(), vec![s2.clone()]);

            assert_eq!(s.seals_for(&r1).unwrap(), vec![ObjectRef([0xab; 32])]);

            assert_eq!(s.receipts_for(&r2).unwrap(), vec![ObjectRef([7; 32])]);

            assert_eq!(s.receipt_states().unwrap(), vec![r2]);

        };

        let reader = f.open();

        assert!(!reader.folded.get());

        check(&reader);

        assert!(f.backend.inner.read_ref(JOURNAL_KEY).unwrap().is_none());

        assert!(fold(&w));

        let reader = f.open();

        assert!(reader.folded.get());

        check(&reader);

        // An older writer after the fold: missed until the next one.

        let r3 = w.put(ObjectType::Chunk, b"three").unwrap();

        let s3 = old_writer_signs(&f, &r3, 3);

        assert!(f.open().signatures(&r3).unwrap().is_empty());

        assert!(fold(&w));

        assert_eq!(f.open().signatures(&r3).unwrap(), vec![s3]);

    }

    /// Writers racing on the journal never lose a line: a lost

    /// compare-and-swap reads the journal again and retries.

    #[test]

    fn concurrent_journal_appends_lose_nothing() {

        let f = Fixture::new();

        // One race, made to happen.

        let w = f.open();

        f.backend.interfere.store(true, Ordering::SeqCst);

        w.index_seal(&ObjectRef([1; 32]), &ObjectRef([2; 32]))

            .unwrap();

        let j = journal(&w);

        assert_eq!(j.len(), 2, "the competitor's line and ours: {j:?}");

        // Many, from threads, each its own handle.

        let backend: Arc<Backend> = f.backend.clone();

        std::thread::scope(|scope| {

            for t in 0..8u8 {

                let backend = backend.clone();

                let dir = f.dir.path().join(format!("t{t}"));

                scope.spawn(move || {

                    let s = Store::open_on(backend, dir).unwrap();

                    for i in 0..25u8 {

                        s.index_seal(&ObjectRef([t + 10; 32]), &ObjectRef([i; 32]))

                            .unwrap();

                    }

                });

            }

        });

        assert_eq!(journal(&w).len(), 2 + 8 * 25);

        assert!(fold(&w));

        let r = f.open();

        for t in 0..8u8 {

            assert_eq!(r.seals_for(&ObjectRef([t + 10; 32])).unwrap().len(), 25);

        }

    }

    /// A batch adds its entries to the journal in one update when it is

    /// flushed; past the limit the writer folds them into a partial list.

    #[test]

    fn a_long_journal_is_folded_into_a_partial_list() {

        let f = Fixture::new();

        let w = f.open();

        assert!(fold(&w), "folded, with nothing yet");

        let key = ObjectRef([5; 32]);

        f.backend.reset();

        let batch = w.begin_batch().unwrap();

        for i in 0..=JOURNAL_FOLD_AT as u32 {

            let mut v = [0u8; 32];

            v[..4].copy_from_slice(&i.to_le_bytes());

            w.index_seal(&key, &ObjectRef(v)).unwrap();

        }

        assert_eq!(f.backend.n("write_ref"), 0, "nothing until the flush");

        batch.finish().unwrap();

        let parts = f

            .backend

            .inner

            .list("pack-")

            .unwrap()

            .into_iter()

            .filter(|e| e.key.ends_with(PART_SUFFIX))

            .count();

        assert_eq!(parts, 1);

        assert!(journal(&w).values().all(|t| *t != 0), "all folded");

        let r = f.open();

        assert_eq!(r.seals_for(&key).unwrap().len(), JOURNAL_FOLD_AT + 1);

        // The next complete fold takes the partial list in.

        assert!(fold(&w));

        let lists: Vec<String> = f.open().entry_lists.borrow().clone();

        assert_eq!(lists.len(), 1);

        assert!(lists[0].ends_with(LIST_SUFFIX));

        assert_eq!(f.open().seals_for(&key).unwrap().len(), JOURNAL_FOLD_AT + 1);

    }

    /// A collection that deletes an object takes its entries out of the

    /// journal too, so the next fold leaves them out.

    #[test]

    fn pruned_objects_leave_the_journal() {

        let f = Fixture::new();

        let w = f.open();

        let gone = w.put(ObjectType::Chunk, b"gone").unwrap();

        let kept = w.put(ObjectType::Chunk, b"kept").unwrap();

        sign(&w, &gone, 1);

        let s = sign(&w, &kept, 2);

        assert!(fold(&w));

        let keep: std::collections::HashSet<ObjectRef> = w

            .list()

            .unwrap()

            .into_iter()

            .filter(|r| *r != gone)

            .collect();

        w.prune(&keep, std::time::Duration::ZERO).unwrap();

        assert!(journal(&w).keys().all(|k| !k.contains(&gone.to_hex()[2..])));

        fold(&w);

        let r = f.open();

        assert!(r.signatures(&gone).unwrap().is_empty());

        assert_eq!(r.signatures(&kept).unwrap(), vec![s]);

        let all = r.with_entries(|e| e.clone()).unwrap().unwrap();

        assert!(

            all.iter().all(|k| !k.contains(&gone.to_hex()[2..])),

            "{all:?}"

        );

    }

    /// Requests per lookup, before and after the fold (#116): twenty

    /// objects, five of them signed, each looked up for its signatures

    /// and its seals, as a fetch does.

    #[test]

    fn a_folded_store_looks_up_entries_without_listing() {

        let f = Fixture::new();

        let w = f.open();

        let refs: Vec<ObjectRef> = (0..20)

            .map(|i| {

                w.put(ObjectType::Chunk, format!("o {i}").as_bytes())

                    .unwrap()

            })

            .collect();

        for r in &refs[..5] {

            sign(&w, r, 1);

        }

        let lookups = |s: &Store| {

            for r in &refs {

                s.signatures(r).unwrap();

                s.seals_for(r).unwrap();

            }

        };

        let before = f.open();

        f.backend.reset();

        lookups(&before);

        let (before_total, before_list) = (f.backend.total(), f.backend.n("list"));

        eprintln!("40 lookups before the fold: {}", f.backend.summary());

        assert!(fold(&w));

        let after = f.open();

        f.backend.reset();

        lookups(&after);

        eprintln!("40 lookups after the fold: {}", f.backend.summary());

        assert_eq!(before_list, 40);

        assert_eq!(f.backend.n("list"), 0);

        assert_eq!(f.backend.n("read_journal"), 1);

        let legacy = |b: &Counted| b.n("get_legacy") + b.n("list_legacy");

        assert!(

            f.backend.total() - legacy(&f.backend) < (before_total - 40) / 4,

            "{}",

            f.backend.summary()

        );

    /// A read of a repository's journal of signatures and index entries

    /// (`pack-entries`, spec §13.7): one per open of a folded store, in

    /// place of a listing per lookup.

    ReadJournal,

    pub const ALL: [Op; 12] = [

        Op::ReadJournal,

            Op::ReadJournal => "read_journal",

        counted(if key == oxy_core::store::JOURNAL_KEY {

            Op::ReadJournal

        } else {

            Op::ReadRef

        });