OXYGENHUB

Coverage / crates/oxy-core/src/repo/strata.rs 91.4%

930 of 1017 lines run, 146 of 188 functions · run #298 on main · run missed

1
//! A repository kept on a [`Store`] (Oxygen Strata, docs/stores.md): what
2
//! `oxy push dir://…` or `oxy clone s3://…` talk to.
3
//!
4
//! The store holds three kinds of key:
5
//!
6
//! - `oxygen.toml`: the layout's format and the repository's id, written
7
//!   once by the first push;
8
//! - `bundles/<seq>-<hash>.bundle`: each push's changes, signatures,
9
//!   seals, objects and policies, as the wire protocol encodes a bundle,
10
//!   written once and named by content;
11
//! - `refs`: every bookmark with its heads and attestation, every tag, the
12
//!   policy and its history, encoded as the protocol's `refs` response,
13
//!   moved by compare-and-swap.
14
//!
15
//! A store cannot run code, so the client does what a hub does on a push,
16
//! against a **mirror**: a bare repository of its own, kept beside the
17
//! remote's other state under `.oxy/remotes/<name>/strata/`, brought level
18
//! with the store before every use (the record read, the bundles it has
19
//! not seen imported, the view set from the record), and then used as a
20
//! [`LocalRemote`] would be. A push runs the hub's own checks
21
//! (`receive_push`) on the mirror, uploads the bundle, and moves `refs`
22
//! from the version it read; if the record moved meanwhile, the mirror is
23
//! put back and the push reports it, exactly as a hub reports a stale
24
//! push. Nothing in the store is trusted: every object is re-hashed as it
25
//! is imported, and the fetch that follows verifies each bookmark's
26
//! attestation under the policy, as it does for any remote.
27
28
use super::{fail, Repo, RepoError, Result};
29
use crate::hash::{ChangeHash, ObjectRef, ObjectType};
30
use crate::objects::StateObject;
31
use crate::protocol::{
32
    wire, Bundle, RefResult, RefUpdate, RefsResponse, Remote, RemoteOptions, SignedPolicy, TagRef,
33
    UNBOUND_ID,
34
};
35
use crate::strata::{Error as StoreError, Store, Version};
36
use std::cell::RefCell;
37
use std::collections::BTreeSet;
38
use std::fs;
39
use std::path::{Path, PathBuf};
40
41
const MARKER_KEY: &str = "oxygen.toml";
42
const REFS_KEY: &str = "refs";
43
const BUNDLES_PREFIX: &str = "bundles/";
44
const FORMAT: u32 = 1;
45
46
impl From<StoreError> for RepoError {
471×
    fn from(e: StoreError) -> RepoError {
481×
        RepoError::Msg(e.to_string())
491×
    }
50
}
51
52
/// The `oxygen.toml` marker: two lines, written once.
5311×
fn marker(id: [u8; 16]) -> Vec<u8> {
5411×
    format!("format = {FORMAT}\nrepo = \"{}\"\n", hex::encode(id)).into_bytes()
5511×
}
56
57115×
fn parse_marker(bytes: &[u8], what: &str) -> Result<[u8; 16]> {
58115×
    let text = String::from_utf8_lossy(bytes);
59115×
    let mut format = None;
60115×
    let mut repo = None;
61229×
    for line in text.lines() {
62229×
        let Some((k, v)) = line.split_once('=') else {
631×
            continue;
64
        };
65228×
        let v = v.trim().trim_matches('"');
66228×
        match k.trim() {
67228×
            "format" => format = v.parse::<u32>().ok(),
68114×
            "repo" => {
69114×
                repo = hex::decode(v)
70114×
                    .ok()
71114×
                    .and_then(|b| <[u8; 16]>::try_from(b).ok())
72
            }
730×
            _ => {}
74
        }
75
    }
76115×
    match (format, repo) {
77114×
        (Some(f), _) if f > FORMAT => fail(format!(
78
            "{what} holds a repository in Strata format {f}, newer than this oxy understands (format {FORMAT}): upgrade oxy"
79
        )),
80113×
        (Some(_), Some(id)) => Ok(id),
811×
        _ => fail(format!("{what}: its oxygen.toml is not a Strata marker")),
82
    }
83115×
}
84
85
/// A remote kept on a store.
86
pub struct BucketRemote {
87
    store: Box<dyn Store>,
88
    /// Where the mirror lives: the remote's directory under `.oxy/remotes`,
89
    /// or a temporary directory when there is no repository yet (a clone),
90
    /// removed on drop.
91
    dir: PathBuf,
92
    temporary: bool,
93
    claimed_id: Option<[u8; 16]>,
94
    mirror: RefCell<Option<Repo>>,
95
}
96
97
impl Drop for BucketRemote {
9859×
    fn drop(&mut self) {
9959×
        if self.temporary {
1009×
            let _ = fs::remove_dir_all(&self.dir);
10150×
        }
10259×
    }
103
}
104
105
/// What a sync found: the record's version (to move it from) and the
106
/// repository id the store holds, if any.
107
struct Synced {
108
    version: Option<Version>,
109
    id: Option<[u8; 16]>,
110
}
111
112
impl BucketRemote {
11359×
    pub fn open(store: Box<dyn Store>, options: &RemoteOptions) -> Result<BucketRemote> {
11459×
        let (dir, temporary) = match &options.cache_dir {
11550×
            Some(d) => (d.clone(), false),
116
            None => {
1179×
                let mut nonce = [0u8; 6];
1189×
                rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut nonce);
1199×
                (
1209×
                    std::env::temp_dir().join(format!(
1219×
                        "oxy-strata-{}-{}",
1229×
                        std::process::id(),
1239×
                        hex::encode(nonce)
1249×
                    )),
1259×
                    true,
1269×
                )
127
            }
128
        };
12959×
        Ok(BucketRemote {
13059×
            store,
13159×
            dir,
13259×
            temporary,
13359×
            claimed_id: options.repo_id,
13459×
            mirror: RefCell::new(None),
13559×
        })
13659×
    }
137
138113×
    fn what(&self) -> String {
139113×
        self.store.describe()
140113×
    }
141
142
    /// The repository id the store holds: `None` until a first push.
143137×
    fn stored_id(&self) -> Result<Option<[u8; 16]>> {
144137×
        match self.store.get(MARKER_KEY) {
145111×
            Ok(b) => Ok(Some(parse_marker(&b, &self.what())?)),
14626×
            Err(StoreError::NotFound(_)) => Ok(None),
1470×
            Err(e) => Err(e.into()),
148
        }
149137×
    }
150
15148×
    fn mirror_dir(&self) -> PathBuf {
15248×
        self.dir.join("mirror")
15348×
    }
154
155179×
    fn seen_path(&self) -> PathBuf {
156179×
        self.dir.join("seen")
157179×
    }
158
159
    /// The bundles already imported into the mirror.
160128×
    fn seen(&self) -> BTreeSet<String> {
161128×
        fs::read_to_string(self.seen_path())
162128×
            .map(|s| s.lines().map(str::to_string).collect())
163128×
            .unwrap_or_default()
164128×
    }
165
16651×
    fn mark_seen(&self, key: &str) -> Result<()> {
16751×
        let mut seen = self.seen();
16851×
        if seen.insert(key.to_string()) {
169114×
            let text: String = seen.iter().map(|k| format!("{k}\n")).collect();
17051×
            crate::durable::write_atomic(&self.seen_path(), text.as_bytes())?;
1710×
        }
17251×
        Ok(())
17351×
    }
174
175
    /// Open the mirror for repository `id`, creating it on first use.
17677×
    fn open_mirror(&self, id: [u8; 16]) -> Result<()> {
17777×
        if self.mirror.borrow().is_some() {
17829×
            return Ok(());
17948×
        }
18048×
        let dir = self.mirror_dir();
18148×
        let repo = if dir.join(super::OXY_DIR).is_dir() {
18227×
            let repo = Repo::open(&dir)?;
18327×
            if repo.id()? != id {
184
                // The store was emptied and reused for another repository:
185
                // start the mirror over.
1860×
                drop(repo);
1870×
                fs::remove_dir_all(&self.dir)?;
1880×
                fs::create_dir_all(&self.dir)?;
1890×
                Repo::init_bare_with_id(&dir, id)?
190
            } else {
19127×
                repo
192
            }
193
        } else {
19421×
            fs::create_dir_all(&self.dir)?;
19521×
            Repo::init_bare_with_id(&dir, id)?
196
        };
19748×
        *self.mirror.borrow_mut() = Some(repo);
19848×
        Ok(())
19977×
    }
200
201230×
    fn with_mirror<T>(&self, f: impl FnOnce(&Repo) -> Result<T>) -> Result<T> {
202230×
        let m = self.mirror.borrow();
203230×
        match m.as_ref() {
204230×
            Some(repo) => f(repo),
2050×
            None => fail(format!(
206
                "{} holds no repository yet: push one to it first",
2070×
                self.what()
208
            )),
209
        }
210230×
    }
211
212
    /// Bring the mirror level with the store.
21396×
    fn sync(&self) -> Result<Synced> {
21496×
        let Some(id) = self.stored_id()? else {
21519×
            return Ok(Synced {
21619×
                version: None,
21719×
                id: None,
21819×
            });
219
        };
22077×
        self.open_mirror(id)?;
22177×
        let record = self.store.read_ref(REFS_KEY)?;
22277×
        let (resp, version) = match record {
22367×
            Some((bytes, version)) => {
22467×
                let (rid, resp) = wire::decode_refs(&bytes)?;
22567×
                if rid != id {
2260×
                    return fail(format!(
227
                        "{}: its refs record names a different repository than its oxygen.toml; the store is damaged",
2280×
                        self.what()
229
                    ));
23067×
                }
23167×
                (resp, Some(version))
232
            }
23310×
            None => (RefsResponse::default(), None),
234
        };
23577×
        self.with_mirror(|mirror| {
236
            // The policy first: bundles are then imported under it.
23777×
            if let Some(sp) = &resp.policy {
2380×
                let have = mirror.current_policy()?.map(|(_, p)| p.policy.serial);
2390×
                if have.is_none_or(|s| s < sp.policy.serial) {
2400×
                    mirror
2410×
                        .adopt_policy_chain(&resp.history, sp, "strata sync")
2420×
                        .map_err(|e| RepoError::Msg(format!("policy from {}: {e}", self.what())))?;
2430×
                }
24477×
            }
24577×
            self.import_new_bundles(mirror)?;
24676×
            self.set_view(mirror, &resp)?;
24775×
            Ok(())
24877×
        })?;
24975×
        Ok(Synced {
25075×
            version,
25175×
            id: Some(id),
25275×
        })
25396×
    }
254
255
    /// Import every bundle in the store the mirror has not seen. Bundles
256
    /// are named in the order they were pushed, but two pushes from the
257
    /// same base may interleave, so a bundle whose dependencies are not
258
    /// there yet is tried again after the others.
25977×
    fn import_new_bundles(&self, mirror: &Repo) -> Result<()> {
26077×
        let seen = self.seen();
26177×
        let mut pending: Vec<String> = self
26277×
            .store
26377×
            .list(BUNDLES_PREFIX)?
26477×
            .into_iter()
26577×
            .map(|e| e.key)
266116×
            .filter(|k| !seen.contains(k))
26777×
            .collect();
26877×
        let strict = mirror.current_policy()?.is_some();
26996×
        while !pending.is_empty() {
27020×
            let mut deferred = Vec::new();
27120×
            let mut last_error = None;
27220×
            let before = pending.len();
27328×
            for key in pending.drain(..) {
27428×
                let bytes = match self.store.get(&key) {
27527×
                    Ok(b) => b,
276
                    // Listed, then gone: a gc merged it into a newer
277
                    // bundle, which this listing holds too.
2781×
                    Err(StoreError::NotFound(_)) => continue,
2790×
                    Err(e) => return Err(e.into()),
280
                };
28127×
                let bundle = wire::decode_bundle(&bytes)
28227×
                    .map_err(|e| RepoError::Msg(format!("{}: {key}: {e}", self.what())))?;
28327×
                match mirror.import_bundle(&bundle, strict) {
28424×
                    Ok(_) => self.mark_seen(&key)?,
2853×
                    Err(RepoError::Msg(m)) if m.contains("neither stored nor in the bundle") => {
2863×
                        last_error = Some(m);
2873×
                        deferred.push(key);
2883×
                    }
2890×
                    Err(e) => return Err(RepoError::Msg(format!("{}: {key}: {e}", self.what()))),
290
                }
291
            }
29220×
            if deferred.len() == before {
293
                // No progress: something those bundles need was never
294
                // pushed, or a push is still in progress.
2951×
                return fail(format!(
296
                    "{}: {} cannot be imported ({}); a push may be in progress, so try again, else the store is missing changes",
2971×
                    self.what(),
2981×
                    deferred.join(", "),
2991×
                    last_error.unwrap_or_default()
300
                ));
30119×
            }
30219×
            pending = deferred;
303
        }
30476×
        Ok(())
30577×
    }
306
307
    /// Make the mirror's view say what the record says.
30876×
    fn set_view(&self, mirror: &Repo, resp: &RefsResponse) -> Result<()> {
30976×
        let (op_ref, op) = mirror.head_op()?;
31076×
        let mut view = op.view.clone();
31176×
        view.bookmarks.clear();
31276×
        for r in &resp.refs {
31371×
            for h in &r.heads {
31471×
                if !mirror.store.has(h) {
3151×
                    return fail(format!(
316
                        "{}: bookmark {:?} names change {}, which no bundle there holds; a push may be in progress, so try again",
3171×
                        self.what(),
318
                        r.name,
3191×
                        h.short(12)
320
                    ));
32170×
                }
322
            }
32367×
            let state = mirror.store.put_state(&StateObject::new(r.heads.clone()))?;
32467×
            if state != r.state {
3250×
                return fail(format!(
326
                    "{}: bookmark {:?} names a state that does not match its heads; the store is damaged",
3270×
                    self.what(),
328
                    r.name
329
                ));
33067×
            }
33167×
            if let Some(sa) = &r.attestation {
3328×
                let current = mirror.latest_ref_attest(&r.name)?.map(|(_, a)| a);
3338×
                if current.as_ref() != Some(sa) {
3343×
                    let a = mirror.store.put_attest(&sa.attest)?;
3353×
                    for sig in &sa.sigs {
3363×
                        mirror.store.put_signature(&a, sig)?;
337
                    }
3383×
                    mirror.set_ref_attest(&r.name, &a)?;
3395×
                }
34059×
            }
34167×
            view.bookmarks.insert(r.name.clone(), state);
342
        }
34375×
        let policy = mirror.current_policy()?.map(|(_, sp)| sp.policy);
34475×
        for t in &resp.tags {
34521×
            mirror.store.put_state(&StateObject::new(t.heads.clone()))?;
34621×
            mirror
34721×
                .adopt_tag(&mut view, t, policy.as_ref())
34821×
                .map_err(|e| RepoError::Msg(format!("{}: {e}", self.what())))?;
349
        }
35075×
        if view != op.view {
35126×
            mirror.record_op(&[op_ref], "strata sync", view)?;
35249×
        }
35375×
        Ok(())
35476×
    }
355
356
    /// The record for the mirror's current references.
35726×
    fn record(&self, id: [u8; 16]) -> Result<Vec<u8>> {
35826×
        self.with_mirror(|mirror| {
35926×
            let mut resp = mirror.remote_refs()?;
360
            // A mirror has no checkout; the store's clone picks a bookmark
361
            // as any other remote's does.
36226×
            resp.default_bookmark = None;
36326×
            Ok(wire::encode_refs(id, &resp))
36426×
        })
36526×
    }
366
367
    /// Store a bundle under the next sequence number; returns its key.
36828×
    fn upload(&self, bundle: &Bundle) -> Result<String> {
36928×
        let bytes = wire::encode_bundle(bundle);
37028×
        let hash = blake3::hash(&bytes).to_hex();
37128×
        let seq = self.store.list(BUNDLES_PREFIX)?.len() + 1;
37228×
        let key = format!("{BUNDLES_PREFIX}{seq:08}-{}.bundle", &hash[..16]);
37328×
        match self.store.put_new(&key, &bytes) {
37427×
            Ok(()) | Err(StoreError::Exists(_)) => {}
3751×
            Err(e) => return Err(e.into()),
376
        }
37727×
        self.mark_seen(&key)?;
37827×
        Ok(key)
37928×
    }
380
}
381
382
impl Remote for BucketRemote {
38341×
    fn repo_id(&self) -> Result<[u8; 16]> {
38441×
        Ok(self.stored_id()?.unwrap_or(UNBOUND_ID))
38541×
    }
386
38755×
    fn refs(&self) -> Result<RefsResponse> {
38855×
        let synced = self.sync()?;
38953×
        if synced.id.is_none() {
3909×
            return Ok(RefsResponse::default());
39144×
        }
39244×
        self.with_mirror(|mirror| {
39344×
            let mut resp = mirror.remote_refs()?;
39444×
            resp.default_bookmark = None;
39544×
            Ok(resp)
39644×
        })
39755×
    }
398
39912×
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
40012×
        self.with_mirror(|m| m.negotiate(have, want))
40112×
    }
402
40312×
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
40412×
        self.with_mirror(|m| m.bundle_raw_for(changes, max_blob))
40512×
    }
406
4070×
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
4080×
        self.with_mirror(|m| m.objects_for(refs))
4090×
    }
410
4110×
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
4120×
        self.with_mirror(|m| m.checkpoint_bundle_for(heads))
4130×
    }
414
4150×
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
4160×
        self.with_mirror(|m| m.shallow_changes(heads, depth))
4170×
    }
418
41925×
    fn push(
42025×
        &self,
42125×
        bundle: Bundle,
42225×
        updates: Vec<RefUpdate>,
42325×
        policy: Option<SignedPolicy>,
42425×
        tags: Vec<TagRef>,
42525×
    ) -> Result<Vec<RefResult>> {
42625×
        let mut synced = self.sync()?;
42725×
        let id = match synced.id {
42815×
            Some(id) => id,
429
            None => {
430
                // The first push binds the store to this repository. Two
431
                // first pushes race on the marker; the loser reads the
432
                // winner's and goes on only if it is the same repository.
43310×
                let Some(id) = self.claimed_id else {
4340×
                    return fail(format!(
435
                        "{} is empty, and this client did not say which repository it is",
4360×
                        self.what()
437
                    ));
438
                };
43910×
                match self.store.put_new(MARKER_KEY, &marker(id)) {
44010×
                    Ok(()) => {}
4410×
                    Err(StoreError::Exists(_)) => {}
4420×
                    Err(e) => return Err(e.into()),
443
                }
44410×
                synced = self.sync()?;
44510×
                match synced.id {
44610×
                    Some(stored) if stored == id => id,
447
                    _ => {
4480×
                        return fail(format!(
449
                            "{} was just bound to another repository by a concurrent push",
4500×
                            self.what()
451
                        ))
452
                    }
453
                }
454
            }
455
        };
45625×
        if let Some(claimed) = self.claimed_id {
45725×
            if claimed != id {
4580×
                return fail(format!(
459
                    "{} holds a different repository (id {}) than this one ({}): a store holds one repository; use another prefix or bucket",
4600×
                    self.what(),
4610×
                    hex::encode(id),
4620×
                    hex::encode(claimed)
463
                ));
46425×
            }
4650×
        }
466
        // The bundle is checked against the mirror and uploaded once; the
467
        // record is then moved from the version read. If it moved
468
        // meanwhile (another push landed), the mirror is put back, brought
469
        // level again, and the updates applied over what landed: two
470
        // pushes to different bookmarks both get through, and one to the
471
        // same bookmark is told it is stale, as a hub would say.
47225×
        let mut uploaded = false;
47327×
        for attempt in 0..PUSH_ATTEMPTS {
47427×
            let (before_ref, before) = self.with_mirror(|m| m.head_op())?;
47527×
            let results = self.with_mirror(|m| {
47627×
                m.receive_push(
47727×
                    bundle.clone(),
47827×
                    updates.clone(),
47927×
                    policy.clone(),
48027×
                    tags.clone(),
481
                )
48227×
            })?;
48327×
            if !results.iter().any(|r| r.ok) {
4840×
                return Ok(results);
48527×
            }
48627×
            if !bundle.is_empty() && !uploaded {
48725×
                self.upload(&bundle)?;
48824×
                uploaded = true;
4892×
            }
49026×
            let record = self.record(id)?;
49126×
            match self
49226×
                .store
49326×
                .write_ref(REFS_KEY, &record, synced.version.as_ref())
494
            {
49524×
                Ok(_) => return Ok(results),
496
                Err(StoreError::Conflict(_)) => {
4972×
                    self.with_mirror(|m| {
4982×
                        let (now, _) = m.head_op()?;
4992×
                        if now != before_ref {
5002×
                            m.record_op(
5012×
                                &[now],
5022×
                                "strata push undone: the remote moved",
5032×
                                before.view.clone(),
5040×
                            )?;
5050×
                        }
5062×
                        Ok(())
5072×
                    })?;
5082×
                    synced = self.sync()?;
5092×
                    if attempt + 1 == PUSH_ATTEMPTS {
5100×
                        return Ok(results
5110×
                            .into_iter()
5120×
                            .map(|r| RefResult {
5130×
                                name: r.name,
514
                                ok: false,
5150×
                                reason: "the remote kept moving while pushing (other pushes landed first); pull, then push again".into(),
5160×
                            })
5170×
                            .collect());
5182×
                    }
519
                }
5200×
                Err(e) => return Err(e.into()),
521
            }
522
        }
5230×
        unreachable!("the loop returns")
52425×
    }
525
}
526
527
/// How often a push is applied over what landed meanwhile before giving up.
528
const PUSH_ATTEMPTS: usize = 4;
529
530
/// The grace period before `gc` removes a bundle it merged: a reader that
531
/// listed it before the merge can still fetch it.
532
pub const REMOTE_GC_GRACE: std::time::Duration = std::time::Duration::from_secs(7 * 86_400);
533
534
/// Key of the record naming the bundles a gc merged, with when.
535
const TRASH_KEY: &str = "gc";
536
537
/// The superseded bundles and when (milliseconds) each was superseded.
538
type Trash = Vec<(String, u64)>;
539
540
/// What a gc against a store did.
541
#[derive(Clone, Debug, Default, PartialEq, Eq)]
542
pub struct RemoteGc {
543
    /// Bundles folded into one new bundle.
544
    pub merged: usize,
545
    /// The merged bundle's size, bytes.
546
    pub bundle_bytes: usize,
547
    /// Changes in the merged bundle.
548
    pub changes: usize,
549
    /// Superseded bundles removed, past the grace period.
550
    pub deleted: usize,
551
    /// Superseded bundles kept for now by the grace period.
552
    pub deferred: usize,
553
}
554
5557×
fn now_ms() -> u64 {
5567×
    std::time::SystemTime::now()
5577×
        .duration_since(std::time::UNIX_EPOCH)
5587×
        .map(|d| d.as_millis() as u64)
5597×
        .unwrap_or(0)
5607×
}
561
562
impl BucketRemote {
563
    /// The bundles a gc superseded, with when: one `KEY\tMILLIS` per line.
5644×
    fn trash(&self) -> Result<(Trash, Option<Version>)> {
5654×
        match self.store.read_ref(TRASH_KEY)? {
5662×
            Some((bytes, v)) => {
5672×
                let text = String::from_utf8_lossy(&bytes);
5682×
                let list = text
5692×
                    .lines()
5704×
                    .filter_map(|l| {
5714×
                        let (k, t) = l.split_once('\t')?;
5724×
                        Some((k.to_string(), t.parse().ok()?))
5734×
                    })
5742×
                    .collect();
5752×
                Ok((list, Some(v)))
576
            }
5772×
            None => Ok((Vec::new(), None)),
578
        }
5794×
    }
580
5816×
    fn write_trash(&self, list: &[(String, u64)], expected: Option<&Version>) -> Result<Version> {
5828×
        let text: String = list.iter().map(|(k, t)| format!("{k}\t{t}\n")).collect();
5836×
        Ok(self.store.write_ref(TRASH_KEY, text.as_bytes(), expected)?)
5846×
    }
585
586
    /// Fold every bundle into one holding what the record reaches, and
587
    /// remove the bundles that made a gc redundant once `grace` has
588
    /// passed since they were superseded (stores.md §6).
5894×
    pub fn gc(&self, grace: std::time::Duration) -> Result<RemoteGc> {
5904×
        let synced = self.sync()?;
5914×
        let Some(id) = synced.id else {
5920×
            return fail(format!("{} holds no repository yet", self.what()));
593
        };
5944×
        let mut report = RemoteGc::default();
5954×
        let existing: Vec<String> = self
5964×
            .store
5974×
            .list(BUNDLES_PREFIX)?
5984×
            .into_iter()
5994×
            .map(|e| e.key)
6004×
            .collect();
6014×
        let (mut trash, mut trash_version) = self.trash()?;
6024×
        let live: Vec<&String> = existing
6034×
            .iter()
60414×
            .filter(|k| !trash.iter().any(|(t, _)| t == *k))
6054×
            .collect();
6064×
        if live.len() > 1 {
607
            // Everything the record reaches, in dependency order, with
608
            // the state objects the bookmarks and tags name.
6093×
            let (bundle, changes) = self.with_mirror(|m| {
6103×
                let resp = m.remote_refs()?;
6113×
                let mut heads: Vec<ChangeHash> = Vec::new();
6123×
                for r in &resp.refs {
6133×
                    heads.extend(r.heads.iter().copied());
6143×
                }
6153×
                for t in &resp.tags {
6162×
                    heads.extend(t.heads.iter().copied());
6172×
                }
6183×
                heads.sort();
6193×
                heads.dedup();
6203×
                let changes = m.negotiate(&[], &heads)?;
6213×
                let mut bundle = m.bundle_for(&changes)?;
6223×
                for r in &resp.refs {
6233×
                    bundle.objects.push((
6243×
                        ObjectType::State,
6253×
                        StateObject::new(r.heads.clone()).encode(),
6263×
                    ));
6273×
                }
6283×
                for t in &resp.tags {
6292×
                    bundle.objects.push((
6302×
                        ObjectType::State,
6312×
                        StateObject::new(t.heads.clone()).encode(),
6322×
                    ));
6332×
                }
6343×
                Ok((bundle, changes.len()))
6353×
            })?;
6363×
            report.changes = changes;
6373×
            report.bundle_bytes = wire::encode_bundle(&bundle).len();
6383×
            let merged_key = self.upload(&bundle)?;
6393×
            let now = now_ms();
6408×
            for k in live {
6418×
                if *k != merged_key {
6428×
                    trash.push((k.clone(), now));
6438×
                    report.merged += 1;
6448×
                }
645
            }
6463×
            trash_version = Some(self.write_trash(&trash, trash_version.as_ref())?);
6471×
        }
648
        // What the grace period has released.
6494×
        let now = now_ms();
6504×
        let mut kept = Vec::new();
65112×
        for (k, when) in trash {
65212×
            if now.saturating_sub(when) >= grace.as_millis() as u64 {
6538×
                self.store.delete(&k)?;
6548×
                report.deleted += 1;
6554×
            } else {
6564×
                kept.push((k, when));
6574×
                report.deferred += 1;
6584×
            }
659
        }
6604×
        if report.deleted > 0 {
6613×
            self.write_trash(&kept, trash_version.as_ref())?;
6621×
        }
6634×
        let _ = id;
6644×
        Ok(report)
6654×
    }
666
}
667
668
impl Repo {
669
    /// `oxy gc --remote NAME`: a gc against a remote kept on a store.
6706×
    pub fn gc_remote(&self, name: &str, now: bool) -> Result<RemoteGc> {
6716×
        let Some(url) = self.config.remotes.get(name).cloned() else {
6722×
            return fail(format!("no remote {name:?}"));
673
        };
6744×
        let options = self.remote_options(name);
6754×
        let Some(store) = crate::protocol::open_store(&url, &options) else {
6760×
            return fail(format!(
677
                "gc --remote works on a remote kept on a store (s3://, gs://, az://, dir://); {name} is {url}"
678
            ));
679
        };
6804×
        let remote = BucketRemote::open(store?, &options)?;
6814×
        let grace = if now {
6823×
            std::time::Duration::ZERO
683
        } else {
6841×
            REMOTE_GC_GRACE
685
        };
6864×
        remote.gc(grace)
6876×
    }
688
}
689
690
/// Open a remote on `store` for the local repository described by `options`.
69148×
pub fn open_store_remote(
69248×
    store: Box<dyn Store>,
69348×
    options: &RemoteOptions,
69448×
) -> Result<Box<dyn Remote>> {
69548×
    Ok(Box::new(BucketRemote::open(store, options)?))
69648×
}
697
698
/// Whether `path` holds a Strata store rather than a repository.
6990×
pub fn is_store_dir(path: &Path) -> bool {
7000×
    path.join(MARKER_KEY).is_file() || path.join(REFS_KEY).is_file()
7010×
}
702
703
#[cfg(test)]
704
mod tests {
705
    use super::*;
706
    use crate::repo::IdentityScope;
707
    use crate::strata::{Entry, LocalStore, MemoryStore};
708
    use std::ops::Range;
709
    use std::sync::Mutex;
710
7119×
    fn hermetic() {
712
        static ONCE: std::sync::Once = std::sync::Once::new();
713
        static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();
7149×
        ONCE.call_once(|| {
7151×
            let d = tempfile::tempdir().unwrap();
7161×
            std::env::set_var("OXY_HOME", d.path());
7171×
            DIR.set(d).ok();
7181×
        });
7199×
    }
720
72122×
    fn write(root: &Path, rel: &str, content: &[u8]) {
72222×
        let p = root.join(rel);
72322×
        fs::create_dir_all(p.parent().unwrap()).unwrap();
72422×
        fs::write(p, content).unwrap();
72522×
    }
726
7276×
    fn add_remote(repo: &mut Repo, name: &str, url: &str) {
7286×
        repo.config.set(&format!("remote.{name}.url"), url).unwrap();
7296×
        repo.save_config().unwrap();
7306×
    }
731
73211×
    fn keys(store: &dyn Store, prefix: &str) -> Vec<String> {
73311×
        store
73411×
            .list(prefix)
73511×
            .unwrap()
73611×
            .into_iter()
73711×
            .map(|e| e.key)
73811×
            .collect()
73911×
    }
740
741
    #[test]
7421×
    fn push_clone_and_pull_through_a_directory_store() {
7431×
        hermetic();
7441×
        let dir = tempfile::tempdir().unwrap();
7451×
        let root = dir.path();
7461×
        let url = format!("dir://{}", root.join("bucket").display());
7471×
        let store = LocalStore::open(root.join("bucket")).unwrap();
748
749
        // A signs, commits, tags and pushes: the store gets the marker, one
750
        // bundle, and a record.
7511×
        let mut a = Repo::init(&root.join("a"))
7521×
            .unwrap()
7531×
            .with_home(root.join("home-a"));
7541×
        a.create_identity("Ada", "[email protected]", IdentityScope::Repo)
7551×
            .unwrap();
7561×
        add_remote(&mut a, "bucket", &url);
7571×
        write(&root.join("a"), "f.txt", b"a\n");
7581×
        let h1 = a.commit("first", &[]).unwrap();
7591×
        a.tag_create("v1", &[h1], "one").unwrap();
7601×
        let results = a.push("bucket", false).unwrap();
7611×
        assert!(results.iter().all(|r| r.ok), "{results:?}");
7621×
        assert_eq!(
7631×
            keys(&store, ""),
7641×
            [
7651×
                "bundles/00000001-".to_string() + &keys(&store, "bundles/")[0][17..],
7661×
                "oxygen.toml".into(),
7671×
                "refs".into()
7681×
            ]
769
        );
7701×
        assert_eq!(
7711×
            parse_marker(&store.get("oxygen.toml").unwrap(), "t").unwrap(),
7721×
            a.id().unwrap()
773
        );
7741×
        let again = a.push("bucket", false).unwrap();
7751×
        assert!(
7761×
            again[0].ok && again[0].reason.contains("up to date"),
777
            "{again:?}"
778
        );
7791×
        assert_eq!(
7801×
            keys(&store, "bundles/").len(),
781
            1,
782
            "a no-op push uploads nothing"
783
        );
784
785
        // B clones from the store: same id, same file, the tag, and the
786
        // seal on the change.
7871×
        let b = Repo::clone(&url, &root.join("b")).unwrap();
7881×
        assert_eq!(b.id().unwrap(), a.id().unwrap());
7891×
        assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"a\n");
7901×
        assert_eq!(b.bookmark_heads().unwrap(), vec![h1]);
7911×
        assert_eq!(
7921×
            b.tags()
7931×
                .unwrap()
7941×
                .iter()
7951×
                .map(|t| t.name.clone())
7961×
                .collect::<Vec<_>>(),
797
            ["v1"]
798
        );
7991×
        assert!(matches!(
8001×
            b.authenticate_change(&h1).unwrap(),
801
            crate::repo::SignatureStatus::Valid { sealed: true, .. }
802
        ));
8031×
        assert!(
8041×
            root.join("b/.oxy/remotes/origin/strata/mirror/.oxy")
8051×
                .is_dir(),
806
            "the mirror lives with the remote's state"
807
        );
8081×
        assert!(
8091×
            !std::env::temp_dir()
8101×
                .read_dir()
8111×
                .unwrap()
8121×
                .flatten()
8137×
                .any(|e| e.file_name().to_string_lossy().starts_with("oxy-strata-")),
814
            "a clone's temporary mirror is removed"
815
        );
816
817
        // B commits and pushes; A pulls it. A then commits and pushes.
8181×
        write(&root.join("b"), "f.txt", b"a\nb\n");
8191×
        let h2 = b.commit("second", &[]).unwrap();
8201×
        assert!(b.push("origin", false).unwrap()[0].ok);
8211×
        assert_eq!(keys(&store, "bundles/").len(), 2);
8221×
        let pr = a.pull("bucket").unwrap();
8231×
        assert!(pr.changed && pr.checked_out);
8241×
        assert_eq!(fs::read(root.join("a/f.txt")).unwrap(), b"a\nb\n");
8251×
        write(&root.join("a"), "g.txt", b"g\n");
8261×
        let h3 = a.commit("third", &[]).unwrap();
8271×
        assert!(a.push("bucket", false).unwrap()[0].ok);
8281×
        let pr = b.pull("origin").unwrap();
8291×
        assert!(pr.changed);
830
        // g.txt is new, so its change commutes with B's: two heads.
8311×
        let mut want = vec![h2, h3];
8321×
        want.sort();
8331×
        assert_eq!(b.bookmark_heads().unwrap(), want);
8341×
        assert_eq!(fs::read(root.join("b/g.txt")).unwrap(), b"g\n");
835
836
        // A third clone gets everything from the three bundles.
8371×
        let c = Repo::clone(&url, &root.join("c")).unwrap();
8381×
        assert_eq!(c.bookmark_heads().unwrap(), want);
8391×
        assert_eq!(fs::read(root.join("c/f.txt")).unwrap(), b"a\nb\n");
8401×
    }
841
842
    #[test]
8431×
    fn a_push_behind_the_remote_is_refused_before_anything_is_uploaded() {
8441×
        hermetic();
8451×
        let dir = tempfile::tempdir().unwrap();
8461×
        let root = dir.path();
8471×
        let url = format!("dir://{}", root.join("bucket").display());
8481×
        let store = LocalStore::open(root.join("bucket")).unwrap();
8491×
        let mut a = Repo::init(&root.join("a")).unwrap();
8501×
        add_remote(&mut a, "bucket", &url);
8511×
        write(&root.join("a"), "f.txt", b"a\n");
8521×
        a.commit("first", &[]).unwrap();
8531×
        assert!(a.push("bucket", false).unwrap()[0].ok);
8541×
        let b = Repo::clone(&url, &root.join("b")).unwrap();
8551×
        write(&root.join("b"), "f.txt", b"b\n");
8561×
        b.commit("from b", &[]).unwrap();
8571×
        assert!(b.push("origin", false).unwrap()[0].ok);
8581×
        write(&root.join("a"), "g.txt", b"g\n");
8591×
        a.commit("from a", &[]).unwrap();
8601×
        let e = a.push("bucket", false).unwrap_err().to_string();
8611×
        assert!(e.contains("pull first"), "{e}");
8621×
        assert_eq!(keys(&store, "bundles/").len(), 2);
8631×
        assert!(a.pull("bucket").unwrap().changed);
8641×
        assert!(a.push("bucket", false).unwrap()[0].ok);
8651×
        assert_eq!(keys(&store, "bundles/").len(), 3);
8661×
    }
867
868
    /// A store whose record moves under a writer the first time it is
869
    /// written to: what a concurrent push looks like between the read and
870
    /// the compare-and-swap.
871
    struct Racing {
872
        inner: MemoryStore,
873
        raced: Mutex<bool>,
874
    }
875
876
    impl Store for Racing {
8774×
        fn describe(&self) -> String {
8784×
            "racing".into()
8794×
        }
8800×
        fn has(&self, key: &str) -> crate::strata::Result<bool> {
8810×
            self.inner.has(key)
8820×
        }
8835×
        fn get(&self, key: &str) -> crate::strata::Result<Vec<u8>> {
8845×
            self.inner.get(key)
8855×
        }
8860×
        fn get_range(&self, key: &str, range: Range<u64>) -> crate::strata::Result<Vec<u8>> {
8870×
            self.inner.get_range(key, range)
8880×
        }
8893×
        fn put_new(&self, key: &str, bytes: &[u8]) -> crate::strata::Result<()> {
8903×
            self.inner.put_new(key, bytes)
8913×
        }
8926×
        fn list(&self, prefix: &str) -> crate::strata::Result<Vec<Entry>> {
8936×
            self.inner.list(prefix)
8946×
        }
8950×
        fn delete(&self, key: &str) -> crate::strata::Result<()> {
8960×
            self.inner.delete(key)
8970×
        }
8984×
        fn read_ref(&self, key: &str) -> crate::strata::Result<Option<(Vec<u8>, Version)>> {
8994×
            self.inner.read_ref(key)
9004×
        }
9013×
        fn write_ref(
9023×
            &self,
9033×
            key: &str,
9043×
            bytes: &[u8],
9053×
            expected: Option<&Version>,
9063×
        ) -> crate::strata::Result<Version> {
9073×
            let mut raced = self.raced.lock().unwrap();
9083×
            if !*raced && expected.is_some() {
9091×
                *raced = true;
910
                // Someone else lands a record: the same references with a
911
                // different detail, so the version moves.
9121×
                let (current, v) = self.inner.read_ref(key)?.unwrap();
9131×
                let (id, mut resp) =
9141×
                    wire::decode_refs(&current).map_err(|e| StoreError::Backend(e.to_string()))?;
9151×
                resp.default_bookmark = Some("raced".into());
9161×
                self.inner
9171×
                    .write_ref(key, &wire::encode_refs(id, &resp), Some(&v))?;
9182×
            }
9193×
            self.inner.write_ref(key, bytes, expected)
9203×
        }
921
    }
922
923
    #[test]
9241×
    fn a_record_moved_during_a_push_is_applied_over_what_landed() {
9251×
        hermetic();
9261×
        let dir = tempfile::tempdir().unwrap();
9271×
        let root = dir.path();
9281×
        let a = Repo::init(&root.join("a")).unwrap();
9291×
        write(&root.join("a"), "f.txt", b"a\n");
9301×
        let h1 = a.commit("first", &[]).unwrap();
9311×
        let store = Box::new(Racing {
9321×
            inner: MemoryStore::new(),
9331×
            raced: Mutex::new(false),
9341×
        });
9351×
        let options = RemoteOptions {
9361×
            repo_id: Some(a.id().unwrap()),
9371×
            cache_dir: Some(root.join("cache")),
9381×
            ..Default::default()
9391×
        };
9401×
        let remote = BucketRemote::open(store, &options).unwrap();
941
        // The first push creates the record (expected None: no race yet).
9421×
        let state = a.store.put_state(&StateObject::new(vec![h1])).unwrap();
9431×
        let empty = a.store.put_state(&StateObject::new(vec![])).unwrap();
9441×
        let mut bundle = a.bundle_for(&[h1]).unwrap();
9451×
        bundle
9461×
            .objects
9471×
            .push((ObjectType::State, StateObject::new(vec![h1]).encode()));
9481×
        let update = |old, new| RefUpdate {
9492×
            name: "main".into(),
9502×
            old_state: old,
9512×
            new_state: new,
952
            force: false,
9532×
            attestation: None,
9542×
        };
9551×
        let r = remote
9561×
            .push(bundle, vec![update(empty, state)], None, vec![])
9571×
            .unwrap();
9581×
        assert!(r[0].ok, "{r:?}");
959
        // The second push meets the race: refused, nothing moved.
9601×
        write(&root.join("a"), "f.txt", b"ab\n");
9611×
        let h2 = a.commit("second", &[]).unwrap();
9621×
        let state2 = a.store.put_state(&StateObject::new(vec![h2])).unwrap();
9631×
        let mut bundle = a.bundle_for(&[h2]).unwrap();
9641×
        bundle
9651×
            .objects
9661×
            .push((ObjectType::State, StateObject::new(vec![h2]).encode()));
9671×
        let r = remote
9681×
            .push(bundle, vec![update(state, state2)], None, vec![])
9691×
            .unwrap();
9701×
        assert!(r[0].ok, "applied over what landed meanwhile: {r:?}");
9711×
        let refs = remote.refs().unwrap();
9721×
        assert_eq!(
9731×
            refs.refs[0].heads,
9741×
            vec![h2],
975
            "the store and the mirror both moved"
976
        );
9771×
    }
978
979
    #[test]
9801×
    fn a_store_holds_one_repository() {
9811×
        hermetic();
9821×
        let dir = tempfile::tempdir().unwrap();
9831×
        let root = dir.path();
9841×
        let url = format!("dir://{}", root.join("bucket").display());
9851×
        let mut a = Repo::init(&root.join("a")).unwrap();
9861×
        add_remote(&mut a, "bucket", &url);
9871×
        write(&root.join("a"), "f.txt", b"a\n");
9881×
        a.commit("first", &[]).unwrap();
9891×
        assert!(a.push("bucket", false).unwrap()[0].ok);
9901×
        let mut other = Repo::init(&root.join("other")).unwrap();
9911×
        add_remote(&mut other, "bucket", &url);
9921×
        write(&root.join("other"), "f.txt", b"o\n");
9931×
        other.commit("mine", &[]).unwrap();
9941×
        let e = other.push("bucket", false).unwrap_err().to_string();
9951×
        assert!(
9961×
            e.contains("different repository") || e.contains("another repository"),
997
            "{e}"
998
        );
9991×
        assert_eq!(
10001×
            keys(&LocalStore::open(root.join("bucket")).unwrap(), "bundles/").len(),
1001
            1
1002
        );
10031×
    }
1004
1005
    #[test]
10061×
    fn a_damaged_bundle_is_caught_and_an_empty_store_cannot_be_cloned() {
10071×
        hermetic();
10081×
        let dir = tempfile::tempdir().unwrap();
10091×
        let root = dir.path();
10101×
        let url = format!("dir://{}", root.join("bucket").display());
10111×
        let e = Repo::clone(&url, &root.join("nothing"))
10121×
            .err()
10131×
            .expect("refused")
10141×
            .to_string();
10151×
        assert!(e.contains("empty and unbound"), "{e}");
10161×
        let mut a = Repo::init(&root.join("a")).unwrap();
10171×
        add_remote(&mut a, "bucket", &url);
10181×
        write(&root.join("a"), "f.txt", b"a\n");
10191×
        a.commit("first", &[]).unwrap();
10201×
        assert!(a.push("bucket", false).unwrap()[0].ok);
10211×
        let store = LocalStore::open(root.join("bucket")).unwrap();
10221×
        let key = keys(&store, "bundles/").remove(0);
10231×
        let mut bytes = store.get(&key).unwrap();
10241×
        let mid = bytes.len() / 2;
10251×
        bytes[mid] ^= 0xff;
10261×
        store.delete(&key).unwrap();
10271×
        store.put_new(&key, &bytes).unwrap();
10281×
        let e = Repo::clone(&url, &root.join("b"))
10291×
            .err()
10301×
            .expect("refused")
10311×
            .to_string();
10321×
        assert!(
10331×
            e.contains(&key) || e.contains("bundle") || e.contains("hash"),
1034
            "{e}"
1035
        );
10361×
        assert!(!root.join("b/f.txt").exists());
10371×
    }
1038
1039
    /// A store shared by two remotes in one test, with a hook run once at
1040
    /// the first conditional write: what another push landing between a
1041
    /// read of the record and its compare-and-swap looks like.
1042
    struct Shared {
1043
        inner: std::sync::Arc<MemoryStore>,
1044
        fail_uploads: std::sync::atomic::AtomicBool,
1045
        vanish: Mutex<Option<String>>,
1046
    }
1047
1048
    thread_local! {
1049
        static HOOK: std::cell::RefCell<Option<Box<dyn FnOnce()>>> = const { std::cell::RefCell::new(None) };
1050
    }
1051
1052
    impl Shared {
10536×
        fn new(inner: std::sync::Arc<MemoryStore>) -> Box<Shared> {
10546×
            Box::new(Shared {
10556×
                inner,
10566×
                fail_uploads: Default::default(),
10576×
                vanish: Mutex::new(None),
10586×
            })
10596×
        }
1060
    }
1061
1062
    impl Store for Shared {
106324×
        fn describe(&self) -> String {
106424×
            "shared".into()
106524×
        }
10660×
        fn has(&self, key: &str) -> crate::strata::Result<bool> {
10670×
            self.inner.has(key)
10680×
        }
106938×
        fn get(&self, key: &str) -> crate::strata::Result<Vec<u8>> {
107038×
            let mut vanish = self.vanish.lock().unwrap();
107138×
            if vanish.as_deref() == Some(key) {
10721×
                *vanish = None;
10731×
                return Err(StoreError::NotFound(key.to_string()));
107437×
            }
107537×
            self.inner.get(key)
107638×
        }
10770×
        fn get_range(&self, key: &str, range: Range<u64>) -> crate::strata::Result<Vec<u8>> {
10780×
            self.inner.get_range(key, range)
10790×
        }
108011×
        fn put_new(&self, key: &str, bytes: &[u8]) -> crate::strata::Result<()> {
108111×
            if key.starts_with("bundles/")
10828×
                && self.fail_uploads.load(std::sync::atomic::Ordering::SeqCst)
1083
            {
10841×
                return Err(StoreError::Backend("shared: the disk is full".into()));
108510×
            }
108610×
            self.inner.put_new(key, bytes)
108711×
        }
108831×
        fn list(&self, prefix: &str) -> crate::strata::Result<Vec<Entry>> {
108931×
            self.inner.list(prefix)
109031×
        }
10910×
        fn delete(&self, key: &str) -> crate::strata::Result<()> {
10920×
            self.inner.delete(key)
10930×
        }
109423×
        fn read_ref(&self, key: &str) -> crate::strata::Result<Option<(Vec<u8>, Version)>> {
109523×
            self.inner.read_ref(key)
109623×
        }
10978×
        fn write_ref(
10988×
            &self,
10998×
            key: &str,
11008×
            bytes: &[u8],
11018×
            expected: Option<&Version>,
11028×
        ) -> crate::strata::Result<Version> {
11038×
            if key == "refs" && expected.is_some() {
11045×
                if let Some(hook) = HOOK.with(|h| h.borrow_mut().take()) {
11051×
                    hook();
11064×
                }
11073×
            }
11088×
            self.inner.write_ref(key, bytes, expected)
11098×
        }
1110
    }
1111
11123×
    fn repo_with_commit(root: &Path, name: &str, file: &str, text: &[u8], msg: &str) -> Repo {
11133×
        let repo = Repo::init(&root.join(name)).unwrap();
11143×
        write(&root.join(name), file, text);
11153×
        repo.commit(msg, &[]).unwrap();
11163×
        repo
11173×
    }
1118
11196×
    fn remote_for(repo: &Repo, store: Box<dyn Store>, root: &Path, name: &str) -> BucketRemote {
11206×
        let options = RemoteOptions {
11216×
            repo_id: Some(repo.id().unwrap()),
11226×
            cache_dir: Some(root.join(format!("cache-{name}"))),
11236×
            ..Default::default()
11246×
        };
11256×
        BucketRemote::open(store, &options).unwrap()
11266×
    }
1127
11288×
    fn push_bookmark(
11298×
        repo: &Repo,
11308×
        remote: &BucketRemote,
11318×
        bookmark: &str,
11328×
        old: Option<ObjectRef>,
11338×
    ) -> Vec<RefResult> {
11348×
        let heads = repo.heads_of(&repo.view().unwrap(), bookmark).unwrap();
11358×
        let state = repo
11368×
            .store
11378×
            .put_state(&StateObject::new(heads.clone()))
11388×
            .unwrap();
11398×
        let empty = repo.store.put_state(&StateObject::new(vec![])).unwrap();
1140
        // Only what the remote lacks, as push_bookmark sends.
11418×
        let have: Vec<ChangeHash> = remote
11428×
            .refs()
11438×
            .unwrap()
11448×
            .refs
11458×
            .iter()
11468×
            .flat_map(|r| r.heads.clone())
11478×
            .collect();
11488×
        let changes = repo.negotiate(&have, &heads).unwrap();
11498×
        let mut bundle = repo.bundle_for(&changes).unwrap();
11508×
        bundle
11518×
            .objects
11528×
            .push((ObjectType::State, StateObject::new(heads).encode()));
11538×
        let update = RefUpdate {
11548×
            name: bookmark.into(),
11558×
            old_state: old.unwrap_or(empty),
11568×
            new_state: state,
11578×
            force: false,
11588×
            attestation: None,
11598×
        };
11608×
        remote.push(bundle, vec![update], None, vec![]).unwrap()
11618×
    }
1162
1163
    #[test]
11641×
    fn two_pushes_to_different_bookmarks_both_land() {
11651×
        hermetic();
11661×
        let dir = tempfile::tempdir().unwrap();
11671×
        let root = dir.path();
11681×
        let inner = std::sync::Arc::new(MemoryStore::new());
1169
        // A pushes main; B (a clone of the same repository) pushes a
1170
        // second bookmark between A's read and A's compare-and-swap.
11711×
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
11721×
        let ra = remote_for(&a, Shared::new(inner.clone()), root, "a");
11731×
        let r = push_bookmark(&a, &ra, "main", None);
11741×
        assert!(r[0].ok, "{r:?}");
11751×
        let b = Repo::init_with_id(&root.join("b"), a.id().unwrap()).unwrap();
11761×
        let rb = remote_for(&b, Shared::new(inner.clone()), root, "b");
11771×
        rb.refs().unwrap();
11781×
        write(&root.join("b"), "g.txt", b"g\n");
11791×
        b.commit("on b", &[]).unwrap();
11801×
        let (_, op) = b.head_op().unwrap();
11811×
        let mut view = op.view.clone();
11821×
        let feature_heads = b.heads_of(&view, "main").unwrap();
11831×
        view.bookmarks.insert(
11841×
            "feature".into(),
11851×
            b.store.put_state(&StateObject::new(feature_heads)).unwrap(),
1186
        );
11871×
        b.record_op(&[b.head_op().unwrap().0], "bookmark create feature", view)
11881×
            .unwrap();
1189
        // A's second push to main; B's push of feature lands in between.
11901×
        write(&root.join("a"), "f.txt", b"aa\n");
11911×
        a.commit("second", &[]).unwrap();
11921×
        let a_old = ra
11931×
            .refs()
11941×
            .unwrap()
11951×
            .refs
11961×
            .iter()
11971×
            .find(|r| r.name == "main")
11981×
            .unwrap()
1199
            .state;
12001×
        let b_remote = std::sync::Arc::new(std::sync::Mutex::new(Some(rb)));
12011×
        let b_handle = b_remote.clone();
12021×
        HOOK.with(|h| {
12031×
            *h.borrow_mut() = Some(Box::new(move || {
12041×
                let rb = b_handle.lock().unwrap().take().unwrap();
12051×
                let r = push_bookmark(&b, &rb, "feature", None);
12061×
                assert!(r[0].ok, "{r:?}");
12071×
            }))
12081×
        });
12091×
        let r = push_bookmark(&a, &ra, "main", Some(a_old));
12101×
        assert!(r[0].ok, "a's push is applied over b's: {r:?}");
12111×
        let refs = ra.refs().unwrap();
12122×
        let names: Vec<&str> = refs.refs.iter().map(|r| r.name.as_str()).collect();
12131×
        assert_eq!(names, ["feature", "main"], "both landed");
12141×
        assert_eq!(inner.list("bundles/").unwrap().len(), 3);
12151×
        let _ = b_remote;
12161×
    }
1217
1218
    #[test]
12191×
    fn an_interrupted_push_leaves_the_store_usable() {
12201×
        hermetic();
12211×
        let dir = tempfile::tempdir().unwrap();
12221×
        let root = dir.path();
12231×
        let inner = std::sync::Arc::new(MemoryStore::new());
12241×
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
12251×
        let shared = Shared::new(inner.clone());
12261×
        let fail = &shared.fail_uploads as *const std::sync::atomic::AtomicBool;
12271×
        let ra = remote_for(&a, shared, root, "a");
12281×
        assert!(push_bookmark(&a, &ra, "main", None)[0].ok);
12291×
        let old = ra.refs().unwrap().refs[0].state;
1230
        // The upload fails: nothing in the store changes, the mirror is
1231
        // honest, and the next push gets through.
12321×
        write(&root.join("a"), "f.txt", b"aa\n");
12331×
        let h2 = a.commit("second", &[]).unwrap();
12341×
        unsafe { (*fail).store(true, std::sync::atomic::Ordering::SeqCst) };
12351×
        let e = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
12361×
            push_bookmark(&a, &ra, "main", Some(old))
12371×
        }));
12381×
        assert!(e.is_err(), "the push reports the failed upload");
12391×
        assert_eq!(inner.list("bundles/").unwrap().len(), 1);
12401×
        let refs = ra.refs().unwrap();
12411×
        assert_eq!(refs.refs[0].state, old, "the record did not move");
12421×
        unsafe { (*fail).store(false, std::sync::atomic::Ordering::SeqCst) };
12431×
        let r = push_bookmark(&a, &ra, "main", Some(old));
12441×
        assert!(r[0].ok, "{r:?}");
12451×
        assert_eq!(ra.refs().unwrap().refs[0].heads, vec![h2]);
12461×
    }
1247
1248
    #[test]
12491×
    fn bundles_out_of_order_or_gone_are_handled() {
12501×
        hermetic();
12511×
        let dir = tempfile::tempdir().unwrap();
12521×
        let root = dir.path();
12531×
        let inner = std::sync::Arc::new(MemoryStore::new());
12541×
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
12551×
        let ra = remote_for(&a, Shared::new(inner.clone()), root, "a");
12561×
        assert!(push_bookmark(&a, &ra, "main", None)[0].ok);
12571×
        let old = ra.refs().unwrap().refs[0].state;
12581×
        write(&root.join("a"), "f.txt", b"aa\n");
12591×
        let h2 = a.commit("second", &[]).unwrap();
12601×
        assert!(push_bookmark(&a, &ra, "main", Some(old))[0].ok);
1261
        // The second bundle (which needs the first) renamed to sort first.
12621×
        let keys: Vec<String> = inner
12631×
            .list("bundles/")
12641×
            .unwrap()
12651×
            .into_iter()
12661×
            .map(|e| e.key)
12671×
            .collect();
12681×
        assert_eq!(keys.len(), 2);
12691×
        let bytes = inner.get(&keys[1]).unwrap();
12701×
        inner.delete(&keys[1]).unwrap();
12711×
        inner
12721×
            .put_new("bundles/00000000-first.bundle", &bytes)
12731×
            .unwrap();
12741×
        let shared = Shared::new(inner.clone());
12751×
        let c = Repo::init_with_id(&root.join("c"), a.id().unwrap()).unwrap();
12761×
        let rc = remote_for(&c, shared, root, "c");
12771×
        assert_eq!(
12781×
            rc.refs().unwrap().refs[0].heads,
12791×
            vec![h2],
1280
            "the dependent bundle waited for its dependency"
1281
        );
1282
        // A bundle gone between the listing and the read is skipped.
12831×
        let shared = Shared::new(inner.clone());
12841×
        *shared.vanish.lock().unwrap() = Some(keys[0].clone());
12851×
        let d = Repo::init_with_id(&root.join("d"), a.id().unwrap()).unwrap();
12861×
        let rd = remote_for(&d, shared, root, "d");
1287
        // Without the first bundle the second cannot be imported: the sync
1288
        // says so rather than failing silently.
12891×
        let e = rd.refs().expect_err("missing changes").to_string();
12901×
        assert!(e.contains("cannot be imported"), "{e}");
12911×
    }
1292
1293
    #[test]
12941×
    fn gc_merges_bundles_and_prunes_after_the_grace_period() {
12951×
        hermetic();
12961×
        let dir = tempfile::tempdir().unwrap();
12971×
        let root = dir.path();
12981×
        let url = format!("dir://{}", root.join("bucket").display());
12991×
        let store = LocalStore::open(root.join("bucket")).unwrap();
13001×
        let mut a = Repo::init(&root.join("a")).unwrap();
13011×
        add_remote(&mut a, "bucket", &url);
13023×
        for i in 1..=3 {
13033×
            write(&root.join("a"), "f.txt", format!("{i}\n").as_bytes());
13043×
            a.commit(&format!("change {i}"), &[]).unwrap();
13053×
            assert!(a.push("bucket", false).unwrap()[0].ok);
1306
        }
13071×
        a.tag_create("v3", &a.bookmark_heads().unwrap(), "three")
13081×
            .unwrap();
13091×
        assert!(a.push("bucket", false).unwrap().iter().all(|r| r.ok));
13101×
        assert_eq!(
13111×
            keys(&store, "bundles/").len(),
1312
            4,
1313
            "three pushes and the tag's"
1314
        );
1315
        // A gc folds the four into one; the old ones wait out the grace.
13161×
        let r = a.gc_remote("bucket", false).unwrap();
13171×
        assert_eq!(
13181×
            (r.merged, r.changes, r.deleted, r.deferred),
1319
            (4, 3, 0, 4),
1320
            "{r:?}"
1321
        );
13221×
        assert!(r.bundle_bytes > 0);
13231×
        assert_eq!(keys(&store, "bundles/").len(), 5);
13241×
        let b = Repo::clone(&url, &root.join("b")).unwrap();
13251×
        assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"3\n");
13261×
        assert_eq!(b.tags().unwrap().len(), 1);
1327
        // Past the grace period they go, and a clone needs only the one.
13281×
        let r = a.gc_remote("bucket", true).unwrap();
13291×
        assert_eq!((r.merged, r.deleted, r.deferred), (0, 4, 0), "{r:?}");
13301×
        assert_eq!(keys(&store, "bundles/").len(), 1);
13311×
        let c = Repo::clone(&url, &root.join("c")).unwrap();
13321×
        assert_eq!(c.bookmark_heads().unwrap(), a.bookmark_heads().unwrap());
1333
        // The earlier clone, which saw the old bundles, carries on.
13341×
        write(&root.join("b"), "f.txt", b"4\n");
13351×
        b.commit("four", &[]).unwrap();
13361×
        assert!(b.push("origin", false).unwrap()[0].ok);
13371×
        assert!(a.pull("bucket").unwrap().changed);
13381×
        assert_eq!(fs::read(root.join("a/f.txt")).unwrap(), b"4\n");
1339
        // Nothing more to merge: a second gc is a no-op.
13401×
        let r = a.gc_remote("bucket", true).unwrap();
13411×
        assert_eq!((r.merged, r.deleted), (2, 2));
13421×
        assert!(a.gc_remote("nope", true).is_err());
13431×
    }
1344
1345
    #[test]
13461×
    fn markers_parse_and_refuse_newer_formats() {
13471×
        let id = [7u8; 16];
13481×
        assert_eq!(parse_marker(&marker(id), "s").unwrap(), id);
13491×
        let e = parse_marker(
13501×
            b"format = 9\nrepo = \"07070707070707070707070707070707\"\n",
13511×
            "s",
13521×
        )
13531×
        .unwrap_err()
13541×
        .to_string();
13551×
        assert!(e.contains("format 9"), "{e}");
13561×
        assert!(parse_marker(b"hello", "s").is_err());
13571×
    }
1358
}