Fold the older one-list keys on gc and stop reading them 8156967b882c

- **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. The first `oxy gc` also copies the one-list keys of much older stores (`sigs/`, `seals/`, `certs/`, …) into the current layout and marks them folded, so readers stop requesting them; until now nearly every lookup made one of these requests and found nothing. On the test store, 40 lookups went from 85 requests (40 LISTs) to 7 (one journal read and no LIST). 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.

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

    /// a complete fold has copied them into per-object entries and the

    /// pack listing shows its marker (spec §13.7).

        !self.legacy_folded.get()

    /// Copy every older one-list key's contents into per-object entries,

    /// keeping an entry already there, then write the marker that tells

    /// readers to stop reading those keys. The keys themselves are left

    /// for older readers. A list that cannot be read is passed over, and

    /// said.

    fn fold_legacy(&self, progress: Progress) -> Result<()> {

        let mut copied = 0usize;

        for (index, legacy) in ENTRY_INDEXES {

            for e in self

                .backend

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

                .map_err(backend_err)?

            {

                // <legacy>/<hh>/<rest>

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

                let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next())

                else {

                    continue;

                };

                let Some(r) = ObjectRef::from_hex(&format!("{hh}{rest}")) else {

                    continue;

                };

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

                    continue;

                };

                if index == "sig" {

                    let Ok(sigs) = Signature::decode_list(&bytes) else {

                        progress(&format!("{}: not a list of signatures, passed over", e.key));

                        continue;

                    };

                    for sig in sigs {

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

                        let one = Signature::encode_list(std::slice::from_ref(&sig));

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

                        copied += 1;

                    }

                } else {

                    for v in String::from_utf8_lossy(&bytes)

                        .lines()

                        .filter_map(ObjectRef::from_hex)

                    {

                        let key = format!("{}{}", fanout_prefix(index, &r), v.to_hex());

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

                        copied += 1;

                    }

                }

            }

        }

        if copied > 0 {

            progress(&format!("{copied} entries copied out of older lists"));

        }

        put_new_or_existing(

            self.backend.as_ref(),

            LEGACY_FOLDED_KEY,

            format!("{}\n", now_ms()).as_bytes(),

        )?;

        self.legacy_folded.set(true);

        Ok(())

    }

    /// (spec §13.7), what a collection does last: first the older

    /// one-list keys into per-object entries, once; then the journal's lines and

        if !self.legacy_folded.get() {

            progress("folding older lists of signatures and index entries");

            self.fold_legacy(progress)?;

        }

        // The older lists' entries are per-object entries now, and the

        // marker is there.

        assert!(f.backend.inner.has(LEGACY_FOLDED_KEY).unwrap());

        assert!(f

            .backend

            .inner

            .has(&format!("{}{}", fanout_prefix("sig", &r2), s2.key.to_hex()))

            .unwrap());

        assert!(f

            .backend

            .inner

            .has(&format!(

                "{}{}",

                fanout_prefix("receipt", &r2),

                ObjectRef([7; 32]).to_hex()

            ))

            .unwrap());

        assert_eq!(f.backend.n("get_legacy"), 0, "the older lists are folded");

            f.backend.total() < before_total / 10,

    }

    /// Once a fold has copied the older lists into per-object entries and

    /// written its marker, readers stop reading them, folded or not; the

    /// answers are the same.

    #[test]

    fn the_legacy_marker_stops_reads_of_the_older_lists() {

        let f = Fixture::new();

        let w = f.open();

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

        let sig = SecretKey {

            ed_seed: [4; 32],

            ml_seed: [5; 32],

        }

        .sign(Domain::Change, &ObjectRef([4; 32]), &r);

        f.backend

            .put_new(

                &legacy_key("sigs", &r),

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

            )

            .unwrap();

        for (legacy, v) in [("seals", 9u8), ("certs", 8), ("accepted-lists", 6)] {

            f.backend

                .put_new(

                    &legacy_key(legacy, &r),

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

                )

                .unwrap();

        }

        let check = |s: &Store| {

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

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

            assert_eq!(

                s.index_get("cert", "certs", &r).unwrap(),

                vec![ObjectRef([8; 32])]

            );

            assert_eq!(

                s.index_get_pub("accepted", &r).unwrap(),

                vec![ObjectRef([6; 32])]

            );

        };

        let before = f.open();

        f.backend.reset();

        check(&before);

        assert_eq!(f.backend.n("get_legacy"), 4, "{}", f.backend.summary());

        assert!(fold(&w));

        let folded = f.open();

        f.backend.reset();

        check(&folded);

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

        // Not folded (its lists gone), but marked: per-object listings,

        // still no older lists.

        for e in f.backend.inner.list("pack-").unwrap() {

            if is_entry_list(&e.key) {

                f.backend.inner.delete(&e.key).unwrap();

            }

        }

        let marked = f.open();

        assert!(!marked.folded.get() && marked.legacy_folded.get());

        f.backend.reset();

        check(&marked);

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

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