//! A repository kept on a [`Store`] (Oxygen Strata, docs/stores.md): what
//! `oxy push dir://…` or `oxy clone s3://…` talk to.
//!
//! The store holds three kinds of key:
//!
//! - `oxygen.toml`: the layout's format and the repository's id, written
//!   once by the first push;
//! - `bundles/<seq>-<hash>.bundle`: each push's changes, signatures,
//!   seals, objects and policies, as the wire protocol encodes a bundle,
//!   written once and named by content;
//! - `refs`: every bookmark with its heads and attestation, every tag, the
//!   policy and its history, encoded as the protocol's `refs` response,
//!   moved by compare-and-swap.
//!
//! A store cannot run code, so the client does what a hub does on a push,
//! against a **mirror**: a bare repository of its own, kept beside the
//! remote's other state under `.oxy/remotes/<name>/strata/`, brought level
//! with the store before every use (the record read, the bundles it has
//! not seen imported, the view set from the record), and then used as a
//! [`LocalRemote`] would be. A push runs the hub's own checks
//! (`receive_push`) on the mirror, uploads the bundle, and moves `refs`
//! from the version it read; if the record moved meanwhile, the mirror is
//! put back and the push reports it, exactly as a hub reports a stale
//! push. Nothing in the store is trusted: every object is re-hashed as it
//! is imported, and the fetch that follows verifies each bookmark's
//! attestation under the policy, as it does for any remote.

use super::{fail, Repo, RepoError, Result};
use crate::hash::{ChangeHash, ObjectRef, ObjectType};
use crate::objects::StateObject;
use crate::protocol::{
    wire, Bundle, RefResult, RefUpdate, RefsResponse, Remote, RemoteOptions, SignedPolicy, TagRef,
    UNBOUND_ID,
};
use crate::strata::{Error as StoreError, Store, Version};
use std::cell::RefCell;
use std::collections::BTreeSet;
use std::fs;
use std::path::{Path, PathBuf};

const MARKER_KEY: &str = "oxygen.toml";
const REFS_KEY: &str = "refs";
const BUNDLES_PREFIX: &str = "bundles/";
const FORMAT: u32 = 1;

impl From<StoreError> for RepoError {
    fn from(e: StoreError) -> RepoError {
        RepoError::Msg(e.to_string())
    }
}

/// The `oxygen.toml` marker: two lines, written once.
fn marker(id: [u8; 16]) -> Vec<u8> {
    format!("format = {FORMAT}\nrepo = \"{}\"\n", hex::encode(id)).into_bytes()
}

fn parse_marker(bytes: &[u8], what: &str) -> Result<[u8; 16]> {
    let text = String::from_utf8_lossy(bytes);
    let mut format = None;
    let mut repo = None;
    for line in text.lines() {
        let Some((k, v)) = line.split_once('=') else {
            continue;
        };
        let v = v.trim().trim_matches('"');
        match k.trim() {
            "format" => format = v.parse::<u32>().ok(),
            "repo" => {
                repo = hex::decode(v)
                    .ok()
                    .and_then(|b| <[u8; 16]>::try_from(b).ok())
            }
            _ => {}
        }
    }
    match (format, repo) {
        (Some(f), _) if f > FORMAT => fail(format!(
            "{what} holds a repository in Strata format {f}, newer than this oxy understands (format {FORMAT}): upgrade oxy"
        )),
        (Some(_), Some(id)) => Ok(id),
        _ => fail(format!("{what}: its oxygen.toml is not a Strata marker")),
    }
}

/// A remote kept on a store.
pub struct BucketRemote {
    store: Box<dyn Store>,
    /// Where the mirror lives: the remote's directory under `.oxy/remotes`,
    /// or a temporary directory when there is no repository yet (a clone),
    /// removed on drop.
    dir: PathBuf,
    temporary: bool,
    claimed_id: Option<[u8; 16]>,
    mirror: RefCell<Option<Repo>>,
}

impl Drop for BucketRemote {
    fn drop(&mut self) {
        if self.temporary {
            let _ = fs::remove_dir_all(&self.dir);
        }
    }
}

/// What a sync found: the record's version (to move it from) and the
/// repository id the store holds, if any.
struct Synced {
    version: Option<Version>,
    id: Option<[u8; 16]>,
}

impl BucketRemote {
    pub fn open(store: Box<dyn Store>, options: &RemoteOptions) -> Result<BucketRemote> {
        let (dir, temporary) = match &options.cache_dir {
            Some(d) => (d.clone(), false),
            None => {
                let mut nonce = [0u8; 6];
                rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut nonce);
                (
                    std::env::temp_dir().join(format!(
                        "oxy-strata-{}-{}",
                        std::process::id(),
                        hex::encode(nonce)
                    )),
                    true,
                )
            }
        };
        Ok(BucketRemote {
            store,
            dir,
            temporary,
            claimed_id: options.repo_id,
            mirror: RefCell::new(None),
        })
    }

    fn what(&self) -> String {
        self.store.describe()
    }

    /// The repository id the store holds: `None` until a first push.
    fn stored_id(&self) -> Result<Option<[u8; 16]>> {
        match self.store.get(MARKER_KEY) {
            Ok(b) => Ok(Some(parse_marker(&b, &self.what())?)),
            Err(StoreError::NotFound(_)) => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    fn mirror_dir(&self) -> PathBuf {
        self.dir.join("mirror")
    }

    fn seen_path(&self) -> PathBuf {
        self.dir.join("seen")
    }

    /// The bundles already imported into the mirror.
    fn seen(&self) -> BTreeSet<String> {
        fs::read_to_string(self.seen_path())
            .map(|s| s.lines().map(str::to_string).collect())
            .unwrap_or_default()
    }

    fn mark_seen(&self, key: &str) -> Result<()> {
        let mut seen = self.seen();
        if seen.insert(key.to_string()) {
            let text: String = seen.iter().map(|k| format!("{k}\n")).collect();
            crate::durable::write_atomic(&self.seen_path(), text.as_bytes())?;
        }
        Ok(())
    }

    /// Open the mirror for repository `id`, creating it on first use.
    fn open_mirror(&self, id: [u8; 16]) -> Result<()> {
        if self.mirror.borrow().is_some() {
            return Ok(());
        }
        let dir = self.mirror_dir();
        let repo = if dir.join(super::OXY_DIR).is_dir() {
            let repo = Repo::open(&dir)?;
            if repo.id()? != id {
                // The store was emptied and reused for another repository:
                // start the mirror over.
                drop(repo);
                fs::remove_dir_all(&self.dir)?;
                fs::create_dir_all(&self.dir)?;
                Repo::init_bare_with_id(&dir, id)?
            } else {
                repo
            }
        } else {
            fs::create_dir_all(&self.dir)?;
            Repo::init_bare_with_id(&dir, id)?
        };
        *self.mirror.borrow_mut() = Some(repo);
        Ok(())
    }

    fn with_mirror<T>(&self, f: impl FnOnce(&Repo) -> Result<T>) -> Result<T> {
        let m = self.mirror.borrow();
        match m.as_ref() {
            Some(repo) => f(repo),
            None => fail(format!(
                "{} holds no repository yet: push one to it first",
                self.what()
            )),
        }
    }

    /// Bring the mirror level with the store.
    fn sync(&self) -> Result<Synced> {
        let Some(id) = self.stored_id()? else {
            return Ok(Synced {
                version: None,
                id: None,
            });
        };
        self.open_mirror(id)?;
        let record = self.store.read_ref(REFS_KEY)?;
        let (resp, version) = match record {
            Some((bytes, version)) => {
                let (rid, resp) = wire::decode_refs(&bytes)?;
                if rid != id {
                    return fail(format!(
                        "{}: its refs record names a different repository than its oxygen.toml; the store is damaged",
                        self.what()
                    ));
                }
                (resp, Some(version))
            }
            None => (RefsResponse::default(), None),
        };
        self.with_mirror(|mirror| {
            // Another process may have brought the mirror level since this
            // handle read its head.
            mirror.refresh()?;
            // The policy first: bundles are then imported under it.
            if let Some(sp) = &resp.policy {
                let have = mirror.current_policy()?.map(|(_, p)| p.policy.serial);
                if have.is_none_or(|s| s < sp.policy.serial) {
                    mirror
                        .adopt_policy_chain(&resp.history, sp, "strata sync")
                        .map_err(|e| RepoError::Msg(format!("policy from {}: {e}", self.what())))?;
                }
            }
            self.import_new_bundles(mirror)?;
            self.set_view(mirror, &resp)?;
            Ok(())
        })?;
        Ok(Synced {
            version,
            id: Some(id),
        })
    }

    /// Import every bundle in the store the mirror has not seen. Bundles
    /// are named in the order they were pushed, but two pushes from the
    /// same base may interleave, so a bundle whose dependencies are not
    /// there yet is tried again after the others.
    fn import_new_bundles(&self, mirror: &Repo) -> Result<()> {
        let seen = self.seen();
        let mut pending: Vec<String> = self
            .store
            .list(BUNDLES_PREFIX)?
            .into_iter()
            .map(|e| e.key)
            .filter(|k| !seen.contains(k))
            .collect();
        let strict = mirror.current_policy()?.is_some();
        while !pending.is_empty() {
            let mut deferred = Vec::new();
            let mut last_error = None;
            let before = pending.len();
            for key in pending.drain(..) {
                let bytes = match self.store.get(&key) {
                    Ok(b) => b,
                    // Listed, then gone: a gc merged it into a newer
                    // bundle, which this listing holds too.
                    Err(StoreError::NotFound(_)) => continue,
                    Err(e) => return Err(e.into()),
                };
                let bundle = wire::decode_bundle(&bytes)
                    .map_err(|e| RepoError::Msg(format!("{}: {key}: {e}", self.what())))?;
                match mirror.import_bundle(&bundle, strict) {
                    Ok(_) => self.mark_seen(&key)?,
                    Err(RepoError::Msg(m)) if m.contains("neither stored nor in the bundle") => {
                        last_error = Some(m);
                        deferred.push(key);
                    }
                    Err(e) => return Err(RepoError::Msg(format!("{}: {key}: {e}", self.what()))),
                }
            }
            if deferred.len() == before {
                // No progress: something those bundles need was never
                // pushed, or a push is still in progress.
                return fail(format!(
                    "{}: {} cannot be imported ({}); a push may be in progress, so try again, else the store is missing changes",
                    self.what(),
                    deferred.join(", "),
                    last_error.unwrap_or_default()
                ));
            }
            pending = deferred;
        }
        Ok(())
    }

    /// Make the mirror's view say what the record says.
    fn set_view(&self, mirror: &Repo, resp: &RefsResponse) -> Result<()> {
        let (op_ref, op) = mirror.head_op()?;
        let mut view = op.view.clone();
        view.bookmarks.clear();
        for r in &resp.refs {
            for h in &r.heads {
                if !mirror.store.has(h) {
                    return fail(format!(
                        "{}: bookmark {:?} names change {}, which no bundle there holds; a push may be in progress, so try again",
                        self.what(),
                        r.name,
                        h.short(12)
                    ));
                }
            }
            let state = mirror.store.put_state(&StateObject::new(r.heads.clone()))?;
            if state != r.state {
                return fail(format!(
                    "{}: bookmark {:?} names a state that does not match its heads; the store is damaged",
                    self.what(),
                    r.name
                ));
            }
            if let Some(sa) = &r.attestation {
                let current = mirror.latest_ref_attest(&r.name)?.map(|(_, a)| a);
                if current.as_ref() != Some(sa) {
                    let a = mirror.store.put_attest(&sa.attest)?;
                    for sig in &sa.sigs {
                        mirror.store.put_signature(&a, sig)?;
                    }
                    mirror.set_ref_attest(&r.name, &a)?;
                }
            }
            view.bookmarks.insert(r.name.clone(), state);
        }
        let policy = mirror.current_policy()?.map(|(_, sp)| sp.policy);
        for t in &resp.tags {
            mirror.store.put_state(&StateObject::new(t.heads.clone()))?;
            mirror
                .adopt_tag(&mut view, t, policy.as_ref())
                .map_err(|e| RepoError::Msg(format!("{}: {e}", self.what())))?;
        }
        if view != op.view {
            mirror.record_op(&[op_ref], "strata sync", view)?;
        }
        Ok(())
    }

    /// The record for the mirror's current references.
    fn record(&self, id: [u8; 16]) -> Result<Vec<u8>> {
        self.with_mirror(|mirror| {
            let mut resp = mirror.remote_refs()?;
            // A mirror has no checkout; the store's clone picks a bookmark
            // as any other remote's does.
            resp.default_bookmark = None;
            Ok(wire::encode_refs(id, &resp))
        })
    }

    /// Store a bundle under the next sequence number; returns its key.
    fn upload(&self, bundle: &Bundle) -> Result<String> {
        let bytes = wire::encode_bundle(bundle);
        let hash = blake3::hash(&bytes).to_hex();
        let seq = self.store.list(BUNDLES_PREFIX)?.len() + 1;
        let key = format!("{BUNDLES_PREFIX}{seq:08}-{}.bundle", &hash[..16]);
        match self.store.put_new(&key, &bytes) {
            Ok(()) | Err(StoreError::Exists(_)) => {}
            Err(e) => return Err(e.into()),
        }
        self.mark_seen(&key)?;
        Ok(key)
    }
}

impl Remote for BucketRemote {
    fn repo_id(&self) -> Result<[u8; 16]> {
        Ok(self.stored_id()?.unwrap_or(UNBOUND_ID))
    }

    fn refs(&self) -> Result<RefsResponse> {
        let synced = self.sync()?;
        if synced.id.is_none() {
            return Ok(RefsResponse::default());
        }
        self.with_mirror(|mirror| {
            let mut resp = mirror.remote_refs()?;
            resp.default_bookmark = None;
            Ok(resp)
        })
    }

    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
        self.with_mirror(|m| m.negotiate(have, want))
    }

    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
        self.with_mirror(|m| m.bundle_raw_for(changes, max_blob))
    }

    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
        self.with_mirror(|m| m.objects_for(refs))
    }

    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
        self.with_mirror(|m| m.checkpoint_bundle_for(heads))
    }

    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
        self.with_mirror(|m| m.shallow_changes(heads, depth))
    }

    fn push(
        &self,
        bundle: Bundle,
        updates: Vec<RefUpdate>,
        policy: Option<SignedPolicy>,
        tags: Vec<TagRef>,
    ) -> Result<Vec<RefResult>> {
        let mut synced = self.sync()?;
        let id = match synced.id {
            Some(id) => id,
            None => {
                // The first push binds the store to this repository. Two
                // first pushes race on the marker; the loser reads the
                // winner's and goes on only if it is the same repository.
                let Some(id) = self.claimed_id else {
                    return fail(format!(
                        "{} is empty, and this client did not say which repository it is",
                        self.what()
                    ));
                };
                match self.store.put_new(MARKER_KEY, &marker(id)) {
                    Ok(()) => {}
                    Err(StoreError::Exists(_)) => {}
                    Err(e) => return Err(e.into()),
                }
                synced = self.sync()?;
                match synced.id {
                    Some(stored) if stored == id => id,
                    _ => {
                        return fail(format!(
                            "{} was just bound to another repository by a concurrent push",
                            self.what()
                        ))
                    }
                }
            }
        };
        if let Some(claimed) = self.claimed_id {
            if claimed != id {
                return fail(format!(
                    "{} holds a different repository (id {}) than this one ({}): a store holds one repository; use another prefix or bucket",
                    self.what(),
                    hex::encode(id),
                    hex::encode(claimed)
                ));
            }
        }
        // The bundle is checked against the mirror and uploaded once; the
        // record is then moved from the version read. If it moved
        // meanwhile (another push landed), the mirror is put back, brought
        // level again, and the updates applied over what landed: two
        // pushes to different bookmarks both get through, and one to the
        // same bookmark is told it is stale, as a hub would say.
        let mut uploaded = false;
        for attempt in 0..PUSH_ATTEMPTS {
            let (before_ref, before) = self.with_mirror(|m| m.head_op())?;
            let results = self.with_mirror(|m| {
                m.receive_push(
                    bundle.clone(),
                    updates.clone(),
                    policy.clone(),
                    tags.clone(),
                )
            })?;
            if !results.iter().any(|r| r.ok) {
                return Ok(results);
            }
            if !bundle.is_empty() && !uploaded {
                self.upload(&bundle)?;
                uploaded = true;
            }
            let record = self.record(id)?;
            match self
                .store
                .write_ref(REFS_KEY, &record, synced.version.as_ref())
            {
                Ok(_) => return Ok(results),
                Err(StoreError::Conflict(_)) => {
                    self.with_mirror(|m| {
                        let (now, _) = m.head_op()?;
                        if now != before_ref {
                            m.record_op(
                                &[now],
                                "strata push undone: the remote moved",
                                before.view.clone(),
                            )?;
                        }
                        Ok(())
                    })?;
                    synced = self.sync()?;
                    if attempt + 1 == PUSH_ATTEMPTS {
                        return Ok(results
                            .into_iter()
                            .map(|r| RefResult {
                                name: r.name,
                                ok: false,
                                reason: "the remote kept moving while pushing (other pushes landed first); pull, then push again".into(),
                            })
                            .collect());
                    }
                }
                Err(e) => return Err(e.into()),
            }
        }
        unreachable!("the loop returns")
    }
}

/// How often a push is applied over what landed meanwhile before giving up.
const PUSH_ATTEMPTS: usize = 4;

/// The grace period before `gc` removes a bundle it merged: a reader that
/// listed it before the merge can still fetch it.
pub const REMOTE_GC_GRACE: std::time::Duration = std::time::Duration::from_secs(7 * 86_400);

/// Key of the record naming the bundles a gc merged, with when.
const TRASH_KEY: &str = "gc";

/// The superseded bundles and when (milliseconds) each was superseded.
type Trash = Vec<(String, u64)>;

/// What a gc against a store did.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct RemoteGc {
    /// Bundles folded into one new bundle.
    pub merged: usize,
    /// The merged bundle's size, bytes.
    pub bundle_bytes: usize,
    /// Changes in the merged bundle.
    pub changes: usize,
    /// Superseded bundles removed, past the grace period.
    pub deleted: usize,
    /// Superseded bundles kept for now by the grace period.
    pub deferred: usize,
}

fn now_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

impl BucketRemote {
    /// The bundles a gc superseded, with when: one `KEY\tMILLIS` per line.
    fn trash(&self) -> Result<(Trash, Option<Version>)> {
        match self.store.read_ref(TRASH_KEY)? {
            Some((bytes, v)) => {
                let text = String::from_utf8_lossy(&bytes);
                let list = text
                    .lines()
                    .filter_map(|l| {
                        let (k, t) = l.split_once('\t')?;
                        Some((k.to_string(), t.parse().ok()?))
                    })
                    .collect();
                Ok((list, Some(v)))
            }
            None => Ok((Vec::new(), None)),
        }
    }

    fn write_trash(&self, list: &[(String, u64)], expected: Option<&Version>) -> Result<Version> {
        let text: String = list.iter().map(|(k, t)| format!("{k}\t{t}\n")).collect();
        Ok(self.store.write_ref(TRASH_KEY, text.as_bytes(), expected)?)
    }

    /// Fold every bundle into one holding what the record reaches, and
    /// remove the bundles that made a gc redundant once `grace` has
    /// passed since they were superseded (stores.md §6).
    pub fn gc(&self, grace: std::time::Duration) -> Result<RemoteGc> {
        let synced = self.sync()?;
        let Some(id) = synced.id else {
            return fail(format!("{} holds no repository yet", self.what()));
        };
        let mut report = RemoteGc::default();
        let existing: Vec<String> = self
            .store
            .list(BUNDLES_PREFIX)?
            .into_iter()
            .map(|e| e.key)
            .collect();
        let (mut trash, mut trash_version) = self.trash()?;
        let live: Vec<&String> = existing
            .iter()
            .filter(|k| !trash.iter().any(|(t, _)| t == *k))
            .collect();
        if live.len() > 1 {
            // Everything the record reaches, in dependency order, with
            // the state objects the bookmarks and tags name.
            let (bundle, changes) = self.with_mirror(|m| {
                let resp = m.remote_refs()?;
                let mut heads: Vec<ChangeHash> = Vec::new();
                for r in &resp.refs {
                    heads.extend(r.heads.iter().copied());
                }
                for t in &resp.tags {
                    heads.extend(t.heads.iter().copied());
                }
                heads.sort();
                heads.dedup();
                let changes = m.negotiate(&[], &heads)?;
                let mut bundle = m.bundle_for(&changes)?;
                for r in &resp.refs {
                    bundle.objects.push((
                        ObjectType::State,
                        StateObject::new(r.heads.clone()).encode(),
                    ));
                }
                for t in &resp.tags {
                    bundle.objects.push((
                        ObjectType::State,
                        StateObject::new(t.heads.clone()).encode(),
                    ));
                }
                Ok((bundle, changes.len()))
            })?;
            report.changes = changes;
            report.bundle_bytes = wire::encode_bundle(&bundle).len();
            let merged_key = self.upload(&bundle)?;
            let now = now_ms();
            for k in live {
                if *k != merged_key {
                    trash.push((k.clone(), now));
                    report.merged += 1;
                }
            }
            trash_version = Some(self.write_trash(&trash, trash_version.as_ref())?);
        }
        // What the grace period has released.
        let now = now_ms();
        let mut kept = Vec::new();
        for (k, when) in trash {
            if now.saturating_sub(when) >= grace.as_millis() as u64 {
                self.store.delete(&k)?;
                report.deleted += 1;
            } else {
                kept.push((k, when));
                report.deferred += 1;
            }
        }
        if report.deleted > 0 {
            self.write_trash(&kept, trash_version.as_ref())?;
        }
        let _ = id;
        Ok(report)
    }
}

impl Repo {
    /// `oxy gc --remote NAME`: a gc against a remote kept on a store.
    pub fn gc_remote(&self, name: &str, now: bool) -> Result<RemoteGc> {
        let Some(url) = self.config.remotes.get(name).cloned() else {
            return fail(format!("no remote {name:?}"));
        };
        let options = self.remote_options(name);
        let Some(store) = crate::protocol::open_store(&url, &options) else {
            return fail(format!(
                "gc --remote works on a remote kept on a store (s3://, gs://, az://, dir://); {name} is {url}"
            ));
        };
        let remote = BucketRemote::open(store?, &options)?;
        let grace = if now {
            std::time::Duration::ZERO
        } else {
            REMOTE_GC_GRACE
        };
        remote.gc(grace)
    }
}

/// Open a remote on `store` for the local repository described by `options`.
pub fn open_store_remote(
    store: Box<dyn Store>,
    options: &RemoteOptions,
) -> Result<Box<dyn Remote>> {
    Ok(Box::new(BucketRemote::open(store, options)?))
}

/// Whether `path` holds a Strata store rather than a repository.
pub fn is_store_dir(path: &Path) -> bool {
    path.join(MARKER_KEY).is_file() || path.join(REFS_KEY).is_file()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::repo::IdentityScope;
    use crate::strata::{Entry, LocalStore, MemoryStore};
    use std::ops::Range;
    use std::sync::Mutex;

    fn hermetic() {
        static ONCE: std::sync::Once = std::sync::Once::new();
        static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();
        ONCE.call_once(|| {
            let d = tempfile::tempdir().unwrap();
            std::env::set_var("OXY_HOME", d.path());
            DIR.set(d).ok();
        });
    }

    fn write(root: &Path, rel: &str, content: &[u8]) {
        let p = root.join(rel);
        fs::create_dir_all(p.parent().unwrap()).unwrap();
        fs::write(p, content).unwrap();
    }

    fn add_remote(repo: &mut Repo, name: &str, url: &str) {
        repo.config.set(&format!("remote.{name}.url"), url).unwrap();
        repo.save_config().unwrap();
    }

    fn keys(store: &dyn Store, prefix: &str) -> Vec<String> {
        store
            .list(prefix)
            .unwrap()
            .into_iter()
            .map(|e| e.key)
            .collect()
    }

    #[test]
    fn push_clone_and_pull_through_a_directory_store() {
        hermetic();
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let url = format!("dir://{}", root.join("bucket").display());
        let store = LocalStore::open(root.join("bucket")).unwrap();

        // A signs, commits, tags and pushes: the store gets the marker, one
        // bundle, and a record.
        let mut a = Repo::init(&root.join("a"))
            .unwrap()
            .with_home(root.join("home-a"));
        a.create_identity("Ada", "ada@example.org", IdentityScope::Repo)
            .unwrap();
        add_remote(&mut a, "bucket", &url);
        write(&root.join("a"), "f.txt", b"a\n");
        let h1 = a.commit("first", &[]).unwrap();
        a.tag_create("v1", &[h1], "one").unwrap();
        let results = a.push("bucket", false).unwrap();
        assert!(results.iter().all(|r| r.ok), "{results:?}");
        assert_eq!(
            keys(&store, ""),
            [
                "bundles/00000001-".to_string() + &keys(&store, "bundles/")[0][17..],
                "oxygen.toml".into(),
                "refs".into()
            ]
        );
        assert_eq!(
            parse_marker(&store.get("oxygen.toml").unwrap(), "t").unwrap(),
            a.id().unwrap()
        );
        let again = a.push("bucket", false).unwrap();
        assert!(
            again[0].ok && again[0].reason.contains("up to date"),
            "{again:?}"
        );
        assert_eq!(
            keys(&store, "bundles/").len(),
            1,
            "a no-op push uploads nothing"
        );

        // B clones from the store: same id, same file, the tag, and the
        // seal on the change.
        let b = Repo::clone(&url, &root.join("b")).unwrap();
        assert_eq!(b.id().unwrap(), a.id().unwrap());
        assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"a\n");
        assert_eq!(b.bookmark_heads().unwrap(), vec![h1]);
        assert_eq!(
            b.tags()
                .unwrap()
                .iter()
                .map(|t| t.name.clone())
                .collect::<Vec<_>>(),
            ["v1"]
        );
        assert!(matches!(
            b.authenticate_change(&h1).unwrap(),
            crate::repo::SignatureStatus::Valid { sealed: true, .. }
        ));
        assert!(
            root.join("b/.oxy/remotes/origin/strata/mirror/.oxy")
                .is_dir(),
            "the mirror lives with the remote's state"
        );
        assert!(
            !std::env::temp_dir()
                .read_dir()
                .unwrap()
                .flatten()
                .any(|e| e.file_name().to_string_lossy().starts_with("oxy-strata-")),
            "a clone's temporary mirror is removed"
        );

        // B commits and pushes; A pulls it. A then commits and pushes.
        write(&root.join("b"), "f.txt", b"a\nb\n");
        let h2 = b.commit("second", &[]).unwrap();
        assert!(b.push("origin", false).unwrap()[0].ok);
        assert_eq!(keys(&store, "bundles/").len(), 2);
        let pr = a.pull("bucket").unwrap();
        assert!(pr.changed && pr.checked_out);
        assert_eq!(fs::read(root.join("a/f.txt")).unwrap(), b"a\nb\n");
        write(&root.join("a"), "g.txt", b"g\n");
        let h3 = a.commit("third", &[]).unwrap();
        assert!(a.push("bucket", false).unwrap()[0].ok);
        let pr = b.pull("origin").unwrap();
        assert!(pr.changed);
        // g.txt is new, so its change commutes with B's: two heads.
        let mut want = vec![h2, h3];
        want.sort();
        assert_eq!(b.bookmark_heads().unwrap(), want);
        assert_eq!(fs::read(root.join("b/g.txt")).unwrap(), b"g\n");

        // A third clone gets everything from the three bundles.
        let c = Repo::clone(&url, &root.join("c")).unwrap();
        assert_eq!(c.bookmark_heads().unwrap(), want);
        assert_eq!(fs::read(root.join("c/f.txt")).unwrap(), b"a\nb\n");
    }

    #[test]
    fn a_push_behind_the_remote_is_refused_before_anything_is_uploaded() {
        hermetic();
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let url = format!("dir://{}", root.join("bucket").display());
        let store = LocalStore::open(root.join("bucket")).unwrap();
        let mut a = Repo::init(&root.join("a")).unwrap();
        add_remote(&mut a, "bucket", &url);
        write(&root.join("a"), "f.txt", b"a\n");
        a.commit("first", &[]).unwrap();
        assert!(a.push("bucket", false).unwrap()[0].ok);
        let b = Repo::clone(&url, &root.join("b")).unwrap();
        write(&root.join("b"), "f.txt", b"b\n");
        b.commit("from b", &[]).unwrap();
        assert!(b.push("origin", false).unwrap()[0].ok);
        write(&root.join("a"), "g.txt", b"g\n");
        a.commit("from a", &[]).unwrap();
        let e = a.push("bucket", false).unwrap_err().to_string();
        assert!(e.contains("pull first"), "{e}");
        assert_eq!(keys(&store, "bundles/").len(), 2);
        assert!(a.pull("bucket").unwrap().changed);
        assert!(a.push("bucket", false).unwrap()[0].ok);
        assert_eq!(keys(&store, "bundles/").len(), 3);
    }

    /// A store whose record moves under a writer the first time it is
    /// written to: what a concurrent push looks like between the read and
    /// the compare-and-swap.
    struct Racing {
        inner: MemoryStore,
        raced: Mutex<bool>,
    }

    impl Store for Racing {
        fn describe(&self) -> String {
            "racing".into()
        }
        fn has(&self, key: &str) -> crate::strata::Result<bool> {
            self.inner.has(key)
        }
        fn get(&self, key: &str) -> crate::strata::Result<Vec<u8>> {
            self.inner.get(key)
        }
        fn get_range(&self, key: &str, range: Range<u64>) -> crate::strata::Result<Vec<u8>> {
            self.inner.get_range(key, range)
        }
        fn put_new(&self, key: &str, bytes: &[u8]) -> crate::strata::Result<()> {
            self.inner.put_new(key, bytes)
        }
        fn list(&self, prefix: &str) -> crate::strata::Result<Vec<Entry>> {
            self.inner.list(prefix)
        }
        fn delete(&self, key: &str) -> crate::strata::Result<()> {
            self.inner.delete(key)
        }
        fn read_ref(&self, key: &str) -> crate::strata::Result<Option<(Vec<u8>, Version)>> {
            self.inner.read_ref(key)
        }
        fn write_ref(
            &self,
            key: &str,
            bytes: &[u8],
            expected: Option<&Version>,
        ) -> crate::strata::Result<Version> {
            let mut raced = self.raced.lock().unwrap();
            if !*raced && expected.is_some() {
                *raced = true;
                // Someone else lands a record: the same references with a
                // different detail, so the version moves.
                let (current, v) = self.inner.read_ref(key)?.unwrap();
                let (id, mut resp) =
                    wire::decode_refs(&current).map_err(|e| StoreError::Backend(e.to_string()))?;
                resp.default_bookmark = Some("raced".into());
                self.inner
                    .write_ref(key, &wire::encode_refs(id, &resp), Some(&v))?;
            }
            self.inner.write_ref(key, bytes, expected)
        }
    }

    #[test]
    fn a_record_moved_during_a_push_is_applied_over_what_landed() {
        hermetic();
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let a = Repo::init(&root.join("a")).unwrap();
        write(&root.join("a"), "f.txt", b"a\n");
        let h1 = a.commit("first", &[]).unwrap();
        let store = Box::new(Racing {
            inner: MemoryStore::new(),
            raced: Mutex::new(false),
        });
        let options = RemoteOptions {
            repo_id: Some(a.id().unwrap()),
            cache_dir: Some(root.join("cache")),
            ..Default::default()
        };
        let remote = BucketRemote::open(store, &options).unwrap();
        // The first push creates the record (expected None: no race yet).
        let state = a.store.put_state(&StateObject::new(vec![h1])).unwrap();
        let empty = a.store.put_state(&StateObject::new(vec![])).unwrap();
        let mut bundle = a.bundle_for(&[h1]).unwrap();
        bundle
            .objects
            .push((ObjectType::State, StateObject::new(vec![h1]).encode()));
        let update = |old, new| RefUpdate {
            name: "main".into(),
            old_state: old,
            new_state: new,
            force: false,
            attestation: None,
        };
        let r = remote
            .push(bundle, vec![update(empty, state)], None, vec![])
            .unwrap();
        assert!(r[0].ok, "{r:?}");
        // The second push meets the race: refused, nothing moved.
        write(&root.join("a"), "f.txt", b"ab\n");
        let h2 = a.commit("second", &[]).unwrap();
        let state2 = a.store.put_state(&StateObject::new(vec![h2])).unwrap();
        let mut bundle = a.bundle_for(&[h2]).unwrap();
        bundle
            .objects
            .push((ObjectType::State, StateObject::new(vec![h2]).encode()));
        let r = remote
            .push(bundle, vec![update(state, state2)], None, vec![])
            .unwrap();
        assert!(r[0].ok, "applied over what landed meanwhile: {r:?}");
        let refs = remote.refs().unwrap();
        assert_eq!(
            refs.refs[0].heads,
            vec![h2],
            "the store and the mirror both moved"
        );
    }

    #[test]
    fn a_store_holds_one_repository() {
        hermetic();
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let url = format!("dir://{}", root.join("bucket").display());
        let mut a = Repo::init(&root.join("a")).unwrap();
        add_remote(&mut a, "bucket", &url);
        write(&root.join("a"), "f.txt", b"a\n");
        a.commit("first", &[]).unwrap();
        assert!(a.push("bucket", false).unwrap()[0].ok);
        let mut other = Repo::init(&root.join("other")).unwrap();
        add_remote(&mut other, "bucket", &url);
        write(&root.join("other"), "f.txt", b"o\n");
        other.commit("mine", &[]).unwrap();
        let e = other.push("bucket", false).unwrap_err().to_string();
        assert!(
            e.contains("different repository") || e.contains("another repository"),
            "{e}"
        );
        assert_eq!(
            keys(&LocalStore::open(root.join("bucket")).unwrap(), "bundles/").len(),
            1
        );
    }

    #[test]
    fn a_damaged_bundle_is_caught_and_an_empty_store_cannot_be_cloned() {
        hermetic();
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let url = format!("dir://{}", root.join("bucket").display());
        let e = Repo::clone(&url, &root.join("nothing"))
            .err()
            .expect("refused")
            .to_string();
        assert!(e.contains("empty and unbound"), "{e}");
        let mut a = Repo::init(&root.join("a")).unwrap();
        add_remote(&mut a, "bucket", &url);
        write(&root.join("a"), "f.txt", b"a\n");
        a.commit("first", &[]).unwrap();
        assert!(a.push("bucket", false).unwrap()[0].ok);
        let store = LocalStore::open(root.join("bucket")).unwrap();
        let key = keys(&store, "bundles/").remove(0);
        let mut bytes = store.get(&key).unwrap();
        let mid = bytes.len() / 2;
        bytes[mid] ^= 0xff;
        store.delete(&key).unwrap();
        store.put_new(&key, &bytes).unwrap();
        let e = Repo::clone(&url, &root.join("b"))
            .err()
            .expect("refused")
            .to_string();
        assert!(
            e.contains(&key) || e.contains("bundle") || e.contains("hash"),
            "{e}"
        );
        assert!(!root.join("b/f.txt").exists());
    }

    /// A store shared by two remotes in one test, with a hook run once at
    /// the first conditional write: what another push landing between a
    /// read of the record and its compare-and-swap looks like.
    struct Shared {
        inner: std::sync::Arc<MemoryStore>,
        fail_uploads: std::sync::atomic::AtomicBool,
        vanish: Mutex<Option<String>>,
    }

    thread_local! {
        static HOOK: std::cell::RefCell<Option<Box<dyn FnOnce()>>> = const { std::cell::RefCell::new(None) };
    }

    impl Shared {
        fn new(inner: std::sync::Arc<MemoryStore>) -> Box<Shared> {
            Box::new(Shared {
                inner,
                fail_uploads: Default::default(),
                vanish: Mutex::new(None),
            })
        }
    }

    impl Store for Shared {
        fn describe(&self) -> String {
            "shared".into()
        }
        fn has(&self, key: &str) -> crate::strata::Result<bool> {
            self.inner.has(key)
        }
        fn get(&self, key: &str) -> crate::strata::Result<Vec<u8>> {
            let mut vanish = self.vanish.lock().unwrap();
            if vanish.as_deref() == Some(key) {
                *vanish = None;
                return Err(StoreError::NotFound(key.to_string()));
            }
            self.inner.get(key)
        }
        fn get_range(&self, key: &str, range: Range<u64>) -> crate::strata::Result<Vec<u8>> {
            self.inner.get_range(key, range)
        }
        fn put_new(&self, key: &str, bytes: &[u8]) -> crate::strata::Result<()> {
            if key.starts_with("bundles/")
                && self.fail_uploads.load(std::sync::atomic::Ordering::SeqCst)
            {
                return Err(StoreError::Backend("shared: the disk is full".into()));
            }
            self.inner.put_new(key, bytes)
        }
        fn list(&self, prefix: &str) -> crate::strata::Result<Vec<Entry>> {
            self.inner.list(prefix)
        }
        fn delete(&self, key: &str) -> crate::strata::Result<()> {
            self.inner.delete(key)
        }
        fn read_ref(&self, key: &str) -> crate::strata::Result<Option<(Vec<u8>, Version)>> {
            self.inner.read_ref(key)
        }
        fn write_ref(
            &self,
            key: &str,
            bytes: &[u8],
            expected: Option<&Version>,
        ) -> crate::strata::Result<Version> {
            if key == "refs" && expected.is_some() {
                if let Some(hook) = HOOK.with(|h| h.borrow_mut().take()) {
                    hook();
                }
            }
            self.inner.write_ref(key, bytes, expected)
        }
    }

    fn repo_with_commit(root: &Path, name: &str, file: &str, text: &[u8], msg: &str) -> Repo {
        let repo = Repo::init(&root.join(name)).unwrap();
        write(&root.join(name), file, text);
        repo.commit(msg, &[]).unwrap();
        repo
    }

    fn remote_for(repo: &Repo, store: Box<dyn Store>, root: &Path, name: &str) -> BucketRemote {
        let options = RemoteOptions {
            repo_id: Some(repo.id().unwrap()),
            cache_dir: Some(root.join(format!("cache-{name}"))),
            ..Default::default()
        };
        BucketRemote::open(store, &options).unwrap()
    }

    fn push_bookmark(
        repo: &Repo,
        remote: &BucketRemote,
        bookmark: &str,
        old: Option<ObjectRef>,
    ) -> Vec<RefResult> {
        let heads = repo.heads_of(&repo.view().unwrap(), bookmark).unwrap();
        let state = repo
            .store
            .put_state(&StateObject::new(heads.clone()))
            .unwrap();
        let empty = repo.store.put_state(&StateObject::new(vec![])).unwrap();
        // Only what the remote lacks, as push_bookmark sends.
        let have: Vec<ChangeHash> = remote
            .refs()
            .unwrap()
            .refs
            .iter()
            .flat_map(|r| r.heads.clone())
            .collect();
        let changes = repo.negotiate(&have, &heads).unwrap();
        let mut bundle = repo.bundle_for(&changes).unwrap();
        bundle
            .objects
            .push((ObjectType::State, StateObject::new(heads).encode()));
        let update = RefUpdate {
            name: bookmark.into(),
            old_state: old.unwrap_or(empty),
            new_state: state,
            force: false,
            attestation: None,
        };
        remote.push(bundle, vec![update], None, vec![]).unwrap()
    }

    #[test]
    fn two_pushes_to_different_bookmarks_both_land() {
        hermetic();
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let inner = std::sync::Arc::new(MemoryStore::new());
        // A pushes main; B (a clone of the same repository) pushes a
        // second bookmark between A's read and A's compare-and-swap.
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
        let ra = remote_for(&a, Shared::new(inner.clone()), root, "a");
        let r = push_bookmark(&a, &ra, "main", None);
        assert!(r[0].ok, "{r:?}");
        let b = Repo::init_with_id(&root.join("b"), a.id().unwrap()).unwrap();
        let rb = remote_for(&b, Shared::new(inner.clone()), root, "b");
        rb.refs().unwrap();
        write(&root.join("b"), "g.txt", b"g\n");
        b.commit("on b", &[]).unwrap();
        let (_, op) = b.head_op().unwrap();
        let mut view = op.view.clone();
        let feature_heads = b.heads_of(&view, "main").unwrap();
        view.bookmarks.insert(
            "feature".into(),
            b.store.put_state(&StateObject::new(feature_heads)).unwrap(),
        );
        b.record_op(&[b.head_op().unwrap().0], "bookmark create feature", view)
            .unwrap();
        // A's second push to main; B's push of feature lands in between.
        write(&root.join("a"), "f.txt", b"aa\n");
        a.commit("second", &[]).unwrap();
        let a_old = ra
            .refs()
            .unwrap()
            .refs
            .iter()
            .find(|r| r.name == "main")
            .unwrap()
            .state;
        let b_remote = std::sync::Arc::new(std::sync::Mutex::new(Some(rb)));
        let b_handle = b_remote.clone();
        HOOK.with(|h| {
            *h.borrow_mut() = Some(Box::new(move || {
                let rb = b_handle.lock().unwrap().take().unwrap();
                let r = push_bookmark(&b, &rb, "feature", None);
                assert!(r[0].ok, "{r:?}");
            }))
        });
        let r = push_bookmark(&a, &ra, "main", Some(a_old));
        assert!(r[0].ok, "a's push is applied over b's: {r:?}");
        let refs = ra.refs().unwrap();
        let names: Vec<&str> = refs.refs.iter().map(|r| r.name.as_str()).collect();
        assert_eq!(names, ["feature", "main"], "both landed");
        assert_eq!(inner.list("bundles/").unwrap().len(), 3);
        let _ = b_remote;
    }

    #[test]
    fn an_interrupted_push_leaves_the_store_usable() {
        hermetic();
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let inner = std::sync::Arc::new(MemoryStore::new());
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
        let shared = Shared::new(inner.clone());
        let fail = &shared.fail_uploads as *const std::sync::atomic::AtomicBool;
        let ra = remote_for(&a, shared, root, "a");
        assert!(push_bookmark(&a, &ra, "main", None)[0].ok);
        let old = ra.refs().unwrap().refs[0].state;
        // The upload fails: nothing in the store changes, the mirror is
        // honest, and the next push gets through.
        write(&root.join("a"), "f.txt", b"aa\n");
        let h2 = a.commit("second", &[]).unwrap();
        unsafe { (*fail).store(true, std::sync::atomic::Ordering::SeqCst) };
        let e = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            push_bookmark(&a, &ra, "main", Some(old))
        }));
        assert!(e.is_err(), "the push reports the failed upload");
        assert_eq!(inner.list("bundles/").unwrap().len(), 1);
        let refs = ra.refs().unwrap();
        assert_eq!(refs.refs[0].state, old, "the record did not move");
        unsafe { (*fail).store(false, std::sync::atomic::Ordering::SeqCst) };
        let r = push_bookmark(&a, &ra, "main", Some(old));
        assert!(r[0].ok, "{r:?}");
        assert_eq!(ra.refs().unwrap().refs[0].heads, vec![h2]);
    }

    #[test]
    fn bundles_out_of_order_or_gone_are_handled() {
        hermetic();
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let inner = std::sync::Arc::new(MemoryStore::new());
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
        let ra = remote_for(&a, Shared::new(inner.clone()), root, "a");
        assert!(push_bookmark(&a, &ra, "main", None)[0].ok);
        let old = ra.refs().unwrap().refs[0].state;
        write(&root.join("a"), "f.txt", b"aa\n");
        let h2 = a.commit("second", &[]).unwrap();
        assert!(push_bookmark(&a, &ra, "main", Some(old))[0].ok);
        // The second bundle (which needs the first) renamed to sort first.
        let keys: Vec<String> = inner
            .list("bundles/")
            .unwrap()
            .into_iter()
            .map(|e| e.key)
            .collect();
        assert_eq!(keys.len(), 2);
        let bytes = inner.get(&keys[1]).unwrap();
        inner.delete(&keys[1]).unwrap();
        inner
            .put_new("bundles/00000000-first.bundle", &bytes)
            .unwrap();
        let shared = Shared::new(inner.clone());
        let c = Repo::init_with_id(&root.join("c"), a.id().unwrap()).unwrap();
        let rc = remote_for(&c, shared, root, "c");
        assert_eq!(
            rc.refs().unwrap().refs[0].heads,
            vec![h2],
            "the dependent bundle waited for its dependency"
        );
        // A bundle gone between the listing and the read is skipped.
        let shared = Shared::new(inner.clone());
        *shared.vanish.lock().unwrap() = Some(keys[0].clone());
        let d = Repo::init_with_id(&root.join("d"), a.id().unwrap()).unwrap();
        let rd = remote_for(&d, shared, root, "d");
        // Without the first bundle the second cannot be imported: the sync
        // says so rather than failing silently.
        let e = rd.refs().expect_err("missing changes").to_string();
        assert!(e.contains("cannot be imported"), "{e}");
    }

    #[test]
    fn gc_merges_bundles_and_prunes_after_the_grace_period() {
        hermetic();
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let url = format!("dir://{}", root.join("bucket").display());
        let store = LocalStore::open(root.join("bucket")).unwrap();
        let mut a = Repo::init(&root.join("a")).unwrap();
        add_remote(&mut a, "bucket", &url);
        for i in 1..=3 {
            write(&root.join("a"), "f.txt", format!("{i}\n").as_bytes());
            a.commit(&format!("change {i}"), &[]).unwrap();
            assert!(a.push("bucket", false).unwrap()[0].ok);
        }
        a.tag_create("v3", &a.bookmark_heads().unwrap(), "three")
            .unwrap();
        assert!(a.push("bucket", false).unwrap().iter().all(|r| r.ok));
        assert_eq!(
            keys(&store, "bundles/").len(),
            4,
            "three pushes and the tag's"
        );
        // A gc folds the four into one; the old ones wait out the grace.
        let r = a.gc_remote("bucket", false).unwrap();
        assert_eq!(
            (r.merged, r.changes, r.deleted, r.deferred),
            (4, 3, 0, 4),
            "{r:?}"
        );
        assert!(r.bundle_bytes > 0);
        assert_eq!(keys(&store, "bundles/").len(), 5);
        let b = Repo::clone(&url, &root.join("b")).unwrap();
        assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"3\n");
        assert_eq!(b.tags().unwrap().len(), 1);
        // Past the grace period they go, and a clone needs only the one.
        let r = a.gc_remote("bucket", true).unwrap();
        assert_eq!((r.merged, r.deleted, r.deferred), (0, 4, 0), "{r:?}");
        assert_eq!(keys(&store, "bundles/").len(), 1);
        let c = Repo::clone(&url, &root.join("c")).unwrap();
        assert_eq!(c.bookmark_heads().unwrap(), a.bookmark_heads().unwrap());
        // The earlier clone, which saw the old bundles, carries on.
        write(&root.join("b"), "f.txt", b"4\n");
        b.commit("four", &[]).unwrap();
        assert!(b.push("origin", false).unwrap()[0].ok);
        assert!(a.pull("bucket").unwrap().changed);
        assert_eq!(fs::read(root.join("a/f.txt")).unwrap(), b"4\n");
        // Nothing more to merge: a second gc is a no-op.
        let r = a.gc_remote("bucket", true).unwrap();
        assert_eq!((r.merged, r.deleted), (2, 2));
        assert!(a.gc_remote("nope", true).is_err());
    }

    #[test]
    fn markers_parse_and_refuse_newer_formats() {
        let id = [7u8; 16];
        assert_eq!(parse_marker(&marker(id), "s").unwrap(), id);
        let e = parse_marker(
            b"format = 9\nrepo = \"07070707070707070707070707070707\"\n",
            "s",
        )
        .unwrap_err()
        .to_string();
        assert!(e.contains("format 9"), "{e}");
        assert!(parse_marker(b"hello", "s").is_err());
    }
}