OXYGENHUB

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

935 of 1022 lines run, 146 of 188 functions · run #367 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
            // Another process may have brought the mirror level since this
237
            // handle read its head.
23877×
            mirror.refresh()?;
239
            // The policy first: bundles are then imported under it.
24077×
            if let Some(sp) = &resp.policy {
2410×
                let have = mirror.current_policy()?.map(|(_, p)| p.policy.serial);
2420×
                if have.is_none_or(|s| s < sp.policy.serial) {
2430×
                    mirror
2440×
                        .adopt_policy_chain(&resp.history, sp, "strata sync")
2450×
                        .map_err(|e| RepoError::Msg(format!("policy from {}: {e}", self.what())))?;
2460×
                }
24777×
            }
24877×
            self.import_new_bundles(mirror)?;
24976×
            self.set_view(mirror, &resp)?;
25075×
            Ok(())
25177×
        })?;
25275×
        Ok(Synced {
25375×
            version,
25475×
            id: Some(id),
25575×
        })
25696×
    }
257
258
    /// Import every bundle in the store the mirror has not seen. Bundles
259
    /// are named in the order they were pushed, but two pushes from the
260
    /// same base may interleave, so a bundle whose dependencies are not
261
    /// there yet is tried again after the others.
26277×
    fn import_new_bundles(&self, mirror: &Repo) -> Result<()> {
26377×
        let seen = self.seen();
26477×
        let mut pending: Vec<String> = self
26577×
            .store
26677×
            .list(BUNDLES_PREFIX)?
26777×
            .into_iter()
26877×
            .map(|e| e.key)
269116×
            .filter(|k| !seen.contains(k))
27077×
            .collect();
27177×
        let strict = mirror.current_policy()?.is_some();
27296×
        while !pending.is_empty() {
27320×
            let mut deferred = Vec::new();
27420×
            let mut last_error = None;
27520×
            let before = pending.len();
27628×
            for key in pending.drain(..) {
27728×
                let bytes = match self.store.get(&key) {
27827×
                    Ok(b) => b,
279
                    // Listed, then gone: a gc merged it into a newer
280
                    // bundle, which this listing holds too.
2811×
                    Err(StoreError::NotFound(_)) => continue,
2820×
                    Err(e) => return Err(e.into()),
283
                };
28427×
                let bundle = wire::decode_bundle(&bytes)
28527×
                    .map_err(|e| RepoError::Msg(format!("{}: {key}: {e}", self.what())))?;
28627×
                match mirror.import_bundle(&bundle, strict) {
28724×
                    Ok(_) => self.mark_seen(&key)?,
2883×
                    Err(RepoError::Msg(m)) if m.contains("neither stored nor in the bundle") => {
2893×
                        last_error = Some(m);
2903×
                        deferred.push(key);
2913×
                    }
2920×
                    Err(e) => return Err(RepoError::Msg(format!("{}: {key}: {e}", self.what()))),
293
                }
294
            }
29520×
            if deferred.len() == before {
296
                // No progress: something those bundles need was never
297
                // pushed, or a push is still in progress.
2981×
                return fail(format!(
299
                    "{}: {} cannot be imported ({}); a push may be in progress, so try again, else the store is missing changes",
3001×
                    self.what(),
3011×
                    deferred.join(", "),
3021×
                    last_error.unwrap_or_default()
303
                ));
30419×
            }
30519×
            pending = deferred;
306
        }
30776×
        Ok(())
30877×
    }
309
310
    /// Make the mirror's view say what the record says.
31176×
    fn set_view(&self, mirror: &Repo, resp: &RefsResponse) -> Result<()> {
31276×
        let (op_ref, op) = mirror.head_op()?;
31376×
        let mut view = op.view.clone();
31476×
        view.bookmarks.clear();
31576×
        for r in &resp.refs {
31671×
            for h in &r.heads {
31771×
                if !mirror.store.has(h) {
3181×
                    return fail(format!(
319
                        "{}: bookmark {:?} names change {}, which no bundle there holds; a push may be in progress, so try again",
3201×
                        self.what(),
321
                        r.name,
3221×
                        h.short(12)
323
                    ));
32470×
                }
325
            }
32667×
            let state = mirror.store.put_state(&StateObject::new(r.heads.clone()))?;
32767×
            if state != r.state {
3280×
                return fail(format!(
329
                    "{}: bookmark {:?} names a state that does not match its heads; the store is damaged",
3300×
                    self.what(),
331
                    r.name
332
                ));
33367×
            }
33467×
            if let Some(sa) = &r.attestation {
3358×
                let current = mirror.latest_ref_attest(&r.name)?.map(|(_, a)| a);
3368×
                if current.as_ref() != Some(sa) {
3373×
                    let a = mirror.store.put_attest(&sa.attest)?;
3383×
                    for sig in &sa.sigs {
3393×
                        mirror.store.put_signature(&a, sig)?;
340
                    }
3413×
                    mirror.set_ref_attest(&r.name, &a)?;
3425×
                }
34359×
            }
34467×
            view.bookmarks.insert(r.name.clone(), state);
345
        }
34675×
        let policy = mirror.current_policy()?.map(|(_, sp)| sp.policy);
34775×
        for t in &resp.tags {
34821×
            mirror.store.put_state(&StateObject::new(t.heads.clone()))?;
34921×
            mirror
35021×
                .adopt_tag(
35121×
                    &mut view,
35221×
                    t,
35321×
                    policy.as_ref(),
35421×
                    &mut super::tags::TagCheck::Mirrored,
355
                )
35621×
                .map_err(|e| RepoError::Msg(format!("{}: {}", self.what(), e.reason)))?;
357
        }
35875×
        if view != op.view {
35926×
            mirror.record_op(&[op_ref], "strata sync", view)?;
36049×
        }
36175×
        Ok(())
36276×
    }
363
364
    /// The record for the mirror's current references.
36526×
    fn record(&self, id: [u8; 16]) -> Result<Vec<u8>> {
36626×
        self.with_mirror(|mirror| {
36726×
            let mut resp = mirror.remote_refs()?;
368
            // A mirror has no checkout; the store's clone picks a bookmark
369
            // as any other remote's does.
37026×
            resp.default_bookmark = None;
37126×
            Ok(wire::encode_refs(id, &resp))
37226×
        })
37326×
    }
374
375
    /// Store a bundle under the next sequence number; returns its key.
37628×
    fn upload(&self, bundle: &Bundle) -> Result<String> {
37728×
        let bytes = wire::encode_bundle(bundle);
37828×
        let hash = blake3::hash(&bytes).to_hex();
37928×
        let seq = self.store.list(BUNDLES_PREFIX)?.len() + 1;
38028×
        let key = format!("{BUNDLES_PREFIX}{seq:08}-{}.bundle", &hash[..16]);
38128×
        match self.store.put_new(&key, &bytes) {
38227×
            Ok(()) | Err(StoreError::Exists(_)) => {}
3831×
            Err(e) => return Err(e.into()),
384
        }
38527×
        self.mark_seen(&key)?;
38627×
        Ok(key)
38728×
    }
388
}
389
390
impl Remote for BucketRemote {
39141×
    fn repo_id(&self) -> Result<[u8; 16]> {
39241×
        Ok(self.stored_id()?.unwrap_or(UNBOUND_ID))
39341×
    }
394
39555×
    fn refs(&self) -> Result<RefsResponse> {
39655×
        let synced = self.sync()?;
39753×
        if synced.id.is_none() {
3989×
            return Ok(RefsResponse::default());
39944×
        }
40044×
        self.with_mirror(|mirror| {
40144×
            let mut resp = mirror.remote_refs()?;
40244×
            resp.default_bookmark = None;
40344×
            Ok(resp)
40444×
        })
40555×
    }
406
40712×
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
40812×
        self.with_mirror(|m| m.negotiate(have, want))
40912×
    }
410
41112×
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
41212×
        self.with_mirror(|m| m.bundle_raw_for(changes, max_blob))
41312×
    }
414
4150×
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
4160×
        self.with_mirror(|m| m.objects_for(refs))
4170×
    }
418
4190×
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
4200×
        self.with_mirror(|m| m.checkpoint_bundle_for(heads))
4210×
    }
422
4230×
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
4240×
        self.with_mirror(|m| m.shallow_changes(heads, depth))
4250×
    }
426
42725×
    fn push(
42825×
        &self,
42925×
        bundle: Bundle,
43025×
        updates: Vec<RefUpdate>,
43125×
        policy: Option<SignedPolicy>,
43225×
        tags: Vec<TagRef>,
43325×
    ) -> Result<Vec<RefResult>> {
43425×
        let mut synced = self.sync()?;
43525×
        let id = match synced.id {
43615×
            Some(id) => id,
437
            None => {
438
                // The first push binds the store to this repository. Two
439
                // first pushes race on the marker; the loser reads the
440
                // winner's and goes on only if it is the same repository.
44110×
                let Some(id) = self.claimed_id else {
4420×
                    return fail(format!(
443
                        "{} is empty, and this client did not say which repository it is",
4440×
                        self.what()
445
                    ));
446
                };
44710×
                match self.store.put_new(MARKER_KEY, &marker(id)) {
44810×
                    Ok(()) => {}
4490×
                    Err(StoreError::Exists(_)) => {}
4500×
                    Err(e) => return Err(e.into()),
451
                }
45210×
                synced = self.sync()?;
45310×
                match synced.id {
45410×
                    Some(stored) if stored == id => id,
455
                    _ => {
4560×
                        return fail(format!(
457
                            "{} was just bound to another repository by a concurrent push",
4580×
                            self.what()
459
                        ))
460
                    }
461
                }
462
            }
463
        };
46425×
        if let Some(claimed) = self.claimed_id {
46525×
            if claimed != id {
4660×
                return fail(format!(
467
                    "{} holds a different repository (id {}) than this one ({}): a store holds one repository; use another prefix or bucket",
4680×
                    self.what(),
4690×
                    hex::encode(id),
4700×
                    hex::encode(claimed)
471
                ));
47225×
            }
4730×
        }
474
        // The bundle is checked against the mirror and uploaded once; the
475
        // record is then moved from the version read. If it moved
476
        // meanwhile (another push landed), the mirror is put back, brought
477
        // level again, and the updates applied over what landed: two
478
        // pushes to different bookmarks both get through, and one to the
479
        // same bookmark is told it is stale, as a hub would say.
48025×
        let mut uploaded = false;
48127×
        for attempt in 0..PUSH_ATTEMPTS {
48227×
            let (before_ref, before) = self.with_mirror(|m| m.head_op())?;
48327×
            let results = self.with_mirror(|m| {
48427×
                m.receive_push(
48527×
                    bundle.clone(),
48627×
                    updates.clone(),
48727×
                    policy.clone(),
48827×
                    tags.clone(),
489
                )
49027×
            })?;
49127×
            if !results.iter().any(|r| r.ok) {
4920×
                return Ok(results);
49327×
            }
49427×
            if !bundle.is_empty() && !uploaded {
49525×
                self.upload(&bundle)?;
49624×
                uploaded = true;
4972×
            }
49826×
            let record = self.record(id)?;
49926×
            match self
50026×
                .store
50126×
                .write_ref(REFS_KEY, &record, synced.version.as_ref())
502
            {
50324×
                Ok(_) => return Ok(results),
504
                Err(StoreError::Conflict(_)) => {
5052×
                    self.with_mirror(|m| {
5062×
                        let (now, _) = m.head_op()?;
5072×
                        if now != before_ref {
5082×
                            m.record_op(
5092×
                                &[now],
5102×
                                "strata push undone: the remote moved",
5112×
                                before.view.clone(),
5120×
                            )?;
5130×
                        }
5142×
                        Ok(())
5152×
                    })?;
5162×
                    synced = self.sync()?;
5172×
                    if attempt + 1 == PUSH_ATTEMPTS {
5180×
                        return Ok(results
5190×
                            .into_iter()
5200×
                            .map(|r| RefResult {
5210×
                                name: r.name,
522
                                ok: false,
5230×
                                reason: "the remote kept moving while pushing (other pushes landed first); pull, then push again".into(),
5240×
                            })
5250×
                            .collect());
5262×
                    }
527
                }
5280×
                Err(e) => return Err(e.into()),
529
            }
530
        }
5310×
        unreachable!("the loop returns")
53225×
    }
533
}
534
535
/// How often a push is applied over what landed meanwhile before giving up.
536
const PUSH_ATTEMPTS: usize = 4;
537
538
/// The grace period before `gc` removes a bundle it merged: a reader that
539
/// listed it before the merge can still fetch it.
540
pub const REMOTE_GC_GRACE: std::time::Duration = std::time::Duration::from_secs(7 * 86_400);
541
542
/// Key of the record naming the bundles a gc merged, with when.
543
const TRASH_KEY: &str = "gc";
544
545
/// The superseded bundles and when (milliseconds) each was superseded.
546
type Trash = Vec<(String, u64)>;
547
548
/// What a gc against a store did.
549
#[derive(Clone, Debug, Default, PartialEq, Eq)]
550
pub struct RemoteGc {
551
    /// Bundles folded into one new bundle.
552
    pub merged: usize,
553
    /// The merged bundle's size, bytes.
554
    pub bundle_bytes: usize,
555
    /// Changes in the merged bundle.
556
    pub changes: usize,
557
    /// Superseded bundles removed, past the grace period.
558
    pub deleted: usize,
559
    /// Superseded bundles kept for now by the grace period.
560
    pub deferred: usize,
561
}
562
5637×
fn now_ms() -> u64 {
5647×
    std::time::SystemTime::now()
5657×
        .duration_since(std::time::UNIX_EPOCH)
5667×
        .map(|d| d.as_millis() as u64)
5677×
        .unwrap_or(0)
5687×
}
569
570
impl BucketRemote {
571
    /// The bundles a gc superseded, with when: one `KEY\tMILLIS` per line.
5724×
    fn trash(&self) -> Result<(Trash, Option<Version>)> {
5734×
        match self.store.read_ref(TRASH_KEY)? {
5742×
            Some((bytes, v)) => {
5752×
                let text = String::from_utf8_lossy(&bytes);
5762×
                let list = text
5772×
                    .lines()
5784×
                    .filter_map(|l| {
5794×
                        let (k, t) = l.split_once('\t')?;
5804×
                        Some((k.to_string(), t.parse().ok()?))
5814×
                    })
5822×
                    .collect();
5832×
                Ok((list, Some(v)))
584
            }
5852×
            None => Ok((Vec::new(), None)),
586
        }
5874×
    }
588
5896×
    fn write_trash(&self, list: &[(String, u64)], expected: Option<&Version>) -> Result<Version> {
5908×
        let text: String = list.iter().map(|(k, t)| format!("{k}\t{t}\n")).collect();
5916×
        Ok(self.store.write_ref(TRASH_KEY, text.as_bytes(), expected)?)
5926×
    }
593
594
    /// Fold every bundle into one holding what the record reaches, and
595
    /// remove the bundles that made a gc redundant once `grace` has
596
    /// passed since they were superseded (stores.md §6).
5974×
    pub fn gc(&self, grace: std::time::Duration) -> Result<RemoteGc> {
5984×
        let synced = self.sync()?;
5994×
        let Some(id) = synced.id else {
6000×
            return fail(format!("{} holds no repository yet", self.what()));
601
        };
6024×
        let mut report = RemoteGc::default();
6034×
        let existing: Vec<String> = self
6044×
            .store
6054×
            .list(BUNDLES_PREFIX)?
6064×
            .into_iter()
6074×
            .map(|e| e.key)
6084×
            .collect();
6094×
        let (mut trash, mut trash_version) = self.trash()?;
6104×
        let live: Vec<&String> = existing
6114×
            .iter()
61214×
            .filter(|k| !trash.iter().any(|(t, _)| t == *k))
6134×
            .collect();
6144×
        if live.len() > 1 {
615
            // Everything the record reaches, in dependency order, with
616
            // the state objects the bookmarks and tags name.
6173×
            let (bundle, changes) = self.with_mirror(|m| {
6183×
                let resp = m.remote_refs()?;
6193×
                let mut heads: Vec<ChangeHash> = Vec::new();
6203×
                for r in &resp.refs {
6213×
                    heads.extend(r.heads.iter().copied());
6223×
                }
6233×
                for t in &resp.tags {
6242×
                    heads.extend(t.heads.iter().copied());
6252×
                }
6263×
                heads.sort();
6273×
                heads.dedup();
6283×
                let changes = m.negotiate(&[], &heads)?;
6293×
                let mut bundle = m.bundle_for(&changes)?;
6303×
                for r in &resp.refs {
6313×
                    bundle.objects.push((
6323×
                        ObjectType::State,
6333×
                        StateObject::new(r.heads.clone()).encode(),
6343×
                    ));
6353×
                }
6363×
                for t in &resp.tags {
6372×
                    bundle.objects.push((
6382×
                        ObjectType::State,
6392×
                        StateObject::new(t.heads.clone()).encode(),
6402×
                    ));
6412×
                }
6423×
                Ok((bundle, changes.len()))
6433×
            })?;
6443×
            report.changes = changes;
6453×
            report.bundle_bytes = wire::encode_bundle(&bundle).len();
6463×
            let merged_key = self.upload(&bundle)?;
6473×
            let now = now_ms();
6488×
            for k in live {
6498×
                if *k != merged_key {
6508×
                    trash.push((k.clone(), now));
6518×
                    report.merged += 1;
6528×
                }
653
            }
6543×
            trash_version = Some(self.write_trash(&trash, trash_version.as_ref())?);
6551×
        }
656
        // What the grace period has released.
6574×
        let now = now_ms();
6584×
        let mut kept = Vec::new();
65912×
        for (k, when) in trash {
66012×
            if now.saturating_sub(when) >= grace.as_millis() as u64 {
6618×
                self.store.delete(&k)?;
6628×
                report.deleted += 1;
6634×
            } else {
6644×
                kept.push((k, when));
6654×
                report.deferred += 1;
6664×
            }
667
        }
6684×
        if report.deleted > 0 {
6693×
            self.write_trash(&kept, trash_version.as_ref())?;
6701×
        }
6714×
        let _ = id;
6724×
        Ok(report)
6734×
    }
674
}
675
676
impl Repo {
677
    /// `oxy gc --remote NAME`: a gc against a remote kept on a store.
6786×
    pub fn gc_remote(&self, name: &str, now: bool) -> Result<RemoteGc> {
6796×
        let Some(url) = self.config.remotes.get(name).cloned() else {
6802×
            return fail(format!("no remote {name:?}"));
681
        };
6824×
        let options = self.remote_options(name);
6834×
        let Some(store) = crate::protocol::open_store(&url, &options) else {
6840×
            return fail(format!(
685
                "gc --remote works on a remote kept on a store (s3://, gs://, az://, dir://); {name} is {url}"
686
            ));
687
        };
6884×
        let remote = BucketRemote::open(store?, &options)?;
6894×
        let grace = if now {
6903×
            std::time::Duration::ZERO
691
        } else {
6921×
            REMOTE_GC_GRACE
693
        };
6944×
        remote.gc(grace)
6956×
    }
696
}
697
698
/// Open a remote on `store` for the local repository described by `options`.
69948×
pub fn open_store_remote(
70048×
    store: Box<dyn Store>,
70148×
    options: &RemoteOptions,
70248×
) -> Result<Box<dyn Remote>> {
70348×
    Ok(Box::new(BucketRemote::open(store, options)?))
70448×
}
705
706
/// Whether `path` holds a Strata store rather than a repository.
7070×
pub fn is_store_dir(path: &Path) -> bool {
7080×
    path.join(MARKER_KEY).is_file() || path.join(REFS_KEY).is_file()
7090×
}
710
711
#[cfg(test)]
712
mod tests {
713
    use super::*;
714
    use crate::repo::IdentityScope;
715
    use crate::strata::{Entry, LocalStore, MemoryStore};
716
    use std::ops::Range;
717
    use std::sync::Mutex;
718
7199×
    fn hermetic() {
720
        static ONCE: std::sync::Once = std::sync::Once::new();
721
        static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();
7229×
        ONCE.call_once(|| {
7231×
            let d = tempfile::tempdir().unwrap();
7241×
            std::env::set_var("OXY_HOME", d.path());
7251×
            DIR.set(d).ok();
7261×
        });
7279×
    }
728
72922×
    fn write(root: &Path, rel: &str, content: &[u8]) {
73022×
        let p = root.join(rel);
73122×
        fs::create_dir_all(p.parent().unwrap()).unwrap();
73222×
        fs::write(p, content).unwrap();
73322×
    }
734
7356×
    fn add_remote(repo: &mut Repo, name: &str, url: &str) {
7366×
        repo.config.set(&format!("remote.{name}.url"), url).unwrap();
7376×
        repo.save_config().unwrap();
7386×
    }
739
74011×
    fn keys(store: &dyn Store, prefix: &str) -> Vec<String> {
74111×
        store
74211×
            .list(prefix)
74311×
            .unwrap()
74411×
            .into_iter()
74511×
            .map(|e| e.key)
74611×
            .collect()
74711×
    }
748
749
    #[test]
7501×
    fn push_clone_and_pull_through_a_directory_store() {
7511×
        hermetic();
7521×
        let dir = tempfile::tempdir().unwrap();
7531×
        let root = dir.path();
7541×
        let url = format!("dir://{}", root.join("bucket").display());
7551×
        let store = LocalStore::open(root.join("bucket")).unwrap();
756
757
        // A signs, commits, tags and pushes: the store gets the marker, one
758
        // bundle, and a record.
7591×
        let mut a = Repo::init(&root.join("a"))
7601×
            .unwrap()
7611×
            .with_home(root.join("home-a"));
7621×
        a.create_identity("Ada", "[email protected]", IdentityScope::Repo)
7631×
            .unwrap();
7641×
        add_remote(&mut a, "bucket", &url);
7651×
        write(&root.join("a"), "f.txt", b"a\n");
7661×
        let h1 = a.commit("first", &[]).unwrap();
7671×
        a.tag_create("v1", &[h1], "one").unwrap();
7681×
        let results = a.push("bucket", false).unwrap();
7691×
        assert!(results.iter().all(|r| r.ok), "{results:?}");
7701×
        assert_eq!(
7711×
            keys(&store, ""),
7721×
            [
7731×
                "bundles/00000001-".to_string() + &keys(&store, "bundles/")[0][17..],
7741×
                "oxygen.toml".into(),
7751×
                "refs".into()
7761×
            ]
777
        );
7781×
        assert_eq!(
7791×
            parse_marker(&store.get("oxygen.toml").unwrap(), "t").unwrap(),
7801×
            a.id().unwrap()
781
        );
7821×
        let again = a.push("bucket", false).unwrap();
7831×
        assert!(
7841×
            again[0].ok && again[0].reason.contains("up to date"),
785
            "{again:?}"
786
        );
7871×
        assert_eq!(
7881×
            keys(&store, "bundles/").len(),
789
            1,
790
            "a no-op push uploads nothing"
791
        );
792
793
        // B clones from the store: same id, same file, the tag, and the
794
        // seal on the change.
7951×
        let b = Repo::clone(&url, &root.join("b")).unwrap();
7961×
        assert_eq!(b.id().unwrap(), a.id().unwrap());
7971×
        assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"a\n");
7981×
        assert_eq!(b.bookmark_heads().unwrap(), vec![h1]);
7991×
        assert_eq!(
8001×
            b.tags()
8011×
                .unwrap()
8021×
                .iter()
8031×
                .map(|t| t.name.clone())
8041×
                .collect::<Vec<_>>(),
805
            ["v1"]
806
        );
8071×
        assert!(matches!(
8081×
            b.authenticate_change(&h1).unwrap(),
809
            crate::repo::SignatureStatus::Valid { sealed: true, .. }
810
        ));
8111×
        assert!(
8121×
            root.join("b/.oxy/remotes/origin/strata/mirror/.oxy")
8131×
                .is_dir(),
814
            "the mirror lives with the remote's state"
815
        );
8161×
        assert!(
8171×
            !std::env::temp_dir()
8181×
                .read_dir()
8191×
                .unwrap()
8201×
                .flatten()
8218×
                .any(|e| e.file_name().to_string_lossy().starts_with("oxy-strata-")),
822
            "a clone's temporary mirror is removed"
823
        );
824
825
        // B commits and pushes; A pulls it. A then commits and pushes.
8261×
        write(&root.join("b"), "f.txt", b"a\nb\n");
8271×
        let h2 = b.commit("second", &[]).unwrap();
8281×
        assert!(b.push("origin", false).unwrap()[0].ok);
8291×
        assert_eq!(keys(&store, "bundles/").len(), 2);
8301×
        let pr = a.pull("bucket").unwrap();
8311×
        assert!(pr.changed && pr.checked_out);
8321×
        assert_eq!(fs::read(root.join("a/f.txt")).unwrap(), b"a\nb\n");
8331×
        write(&root.join("a"), "g.txt", b"g\n");
8341×
        let h3 = a.commit("third", &[]).unwrap();
8351×
        assert!(a.push("bucket", false).unwrap()[0].ok);
8361×
        let pr = b.pull("origin").unwrap();
8371×
        assert!(pr.changed);
838
        // g.txt is new, so its change commutes with B's: two heads.
8391×
        let mut want = vec![h2, h3];
8401×
        want.sort();
8411×
        assert_eq!(b.bookmark_heads().unwrap(), want);
8421×
        assert_eq!(fs::read(root.join("b/g.txt")).unwrap(), b"g\n");
843
844
        // A third clone gets everything from the three bundles.
8451×
        let c = Repo::clone(&url, &root.join("c")).unwrap();
8461×
        assert_eq!(c.bookmark_heads().unwrap(), want);
8471×
        assert_eq!(fs::read(root.join("c/f.txt")).unwrap(), b"a\nb\n");
8481×
    }
849
850
    #[test]
8511×
    fn a_push_behind_the_remote_is_refused_before_anything_is_uploaded() {
8521×
        hermetic();
8531×
        let dir = tempfile::tempdir().unwrap();
8541×
        let root = dir.path();
8551×
        let url = format!("dir://{}", root.join("bucket").display());
8561×
        let store = LocalStore::open(root.join("bucket")).unwrap();
8571×
        let mut a = Repo::init(&root.join("a")).unwrap();
8581×
        add_remote(&mut a, "bucket", &url);
8591×
        write(&root.join("a"), "f.txt", b"a\n");
8601×
        a.commit("first", &[]).unwrap();
8611×
        assert!(a.push("bucket", false).unwrap()[0].ok);
8621×
        let b = Repo::clone(&url, &root.join("b")).unwrap();
8631×
        write(&root.join("b"), "f.txt", b"b\n");
8641×
        b.commit("from b", &[]).unwrap();
8651×
        assert!(b.push("origin", false).unwrap()[0].ok);
8661×
        write(&root.join("a"), "g.txt", b"g\n");
8671×
        a.commit("from a", &[]).unwrap();
8681×
        let e = a.push("bucket", false).unwrap_err().to_string();
8691×
        assert!(e.contains("pull first"), "{e}");
8701×
        assert_eq!(keys(&store, "bundles/").len(), 2);
8711×
        assert!(a.pull("bucket").unwrap().changed);
8721×
        assert!(a.push("bucket", false).unwrap()[0].ok);
8731×
        assert_eq!(keys(&store, "bundles/").len(), 3);
8741×
    }
875
876
    /// A store whose record moves under a writer the first time it is
877
    /// written to: what a concurrent push looks like between the read and
878
    /// the compare-and-swap.
879
    struct Racing {
880
        inner: MemoryStore,
881
        raced: Mutex<bool>,
882
    }
883
884
    impl Store for Racing {
8854×
        fn describe(&self) -> String {
8864×
            "racing".into()
8874×
        }
8880×
        fn has(&self, key: &str) -> crate::strata::Result<bool> {
8890×
            self.inner.has(key)
8900×
        }
8915×
        fn get(&self, key: &str) -> crate::strata::Result<Vec<u8>> {
8925×
            self.inner.get(key)
8935×
        }
8940×
        fn get_range(&self, key: &str, range: Range<u64>) -> crate::strata::Result<Vec<u8>> {
8950×
            self.inner.get_range(key, range)
8960×
        }
8973×
        fn put_new(&self, key: &str, bytes: &[u8]) -> crate::strata::Result<()> {
8983×
            self.inner.put_new(key, bytes)
8993×
        }
9006×
        fn list(&self, prefix: &str) -> crate::strata::Result<Vec<Entry>> {
9016×
            self.inner.list(prefix)
9026×
        }
9030×
        fn delete(&self, key: &str) -> crate::strata::Result<()> {
9040×
            self.inner.delete(key)
9050×
        }
9064×
        fn read_ref(&self, key: &str) -> crate::strata::Result<Option<(Vec<u8>, Version)>> {
9074×
            self.inner.read_ref(key)
9084×
        }
9093×
        fn write_ref(
9103×
            &self,
9113×
            key: &str,
9123×
            bytes: &[u8],
9133×
            expected: Option<&Version>,
9143×
        ) -> crate::strata::Result<Version> {
9153×
            let mut raced = self.raced.lock().unwrap();
9163×
            if !*raced && expected.is_some() {
9171×
                *raced = true;
918
                // Someone else lands a record: the same references with a
919
                // different detail, so the version moves.
9201×
                let (current, v) = self.inner.read_ref(key)?.unwrap();
9211×
                let (id, mut resp) =
9221×
                    wire::decode_refs(&current).map_err(|e| StoreError::Backend(e.to_string()))?;
9231×
                resp.default_bookmark = Some("raced".into());
9241×
                self.inner
9251×
                    .write_ref(key, &wire::encode_refs(id, &resp), Some(&v))?;
9262×
            }
9273×
            self.inner.write_ref(key, bytes, expected)
9283×
        }
929
    }
930
931
    #[test]
9321×
    fn a_record_moved_during_a_push_is_applied_over_what_landed() {
9331×
        hermetic();
9341×
        let dir = tempfile::tempdir().unwrap();
9351×
        let root = dir.path();
9361×
        let a = Repo::init(&root.join("a")).unwrap();
9371×
        write(&root.join("a"), "f.txt", b"a\n");
9381×
        let h1 = a.commit("first", &[]).unwrap();
9391×
        let store = Box::new(Racing {
9401×
            inner: MemoryStore::new(),
9411×
            raced: Mutex::new(false),
9421×
        });
9431×
        let options = RemoteOptions {
9441×
            repo_id: Some(a.id().unwrap()),
9451×
            cache_dir: Some(root.join("cache")),
9461×
            ..Default::default()
9471×
        };
9481×
        let remote = BucketRemote::open(store, &options).unwrap();
949
        // The first push creates the record (expected None: no race yet).
9501×
        let state = a.store.put_state(&StateObject::new(vec![h1])).unwrap();
9511×
        let empty = a.store.put_state(&StateObject::new(vec![])).unwrap();
9521×
        let mut bundle = a.bundle_for(&[h1]).unwrap();
9531×
        bundle
9541×
            .objects
9551×
            .push((ObjectType::State, StateObject::new(vec![h1]).encode()));
9561×
        let update = |old, new| RefUpdate {
9572×
            name: "main".into(),
9582×
            old_state: old,
9592×
            new_state: new,
960
            force: false,
9612×
            attestation: None,
9622×
        };
9631×
        let r = remote
9641×
            .push(bundle, vec![update(empty, state)], None, vec![])
9651×
            .unwrap();
9661×
        assert!(r[0].ok, "{r:?}");
967
        // The second push meets the race: refused, nothing moved.
9681×
        write(&root.join("a"), "f.txt", b"ab\n");
9691×
        let h2 = a.commit("second", &[]).unwrap();
9701×
        let state2 = a.store.put_state(&StateObject::new(vec![h2])).unwrap();
9711×
        let mut bundle = a.bundle_for(&[h2]).unwrap();
9721×
        bundle
9731×
            .objects
9741×
            .push((ObjectType::State, StateObject::new(vec![h2]).encode()));
9751×
        let r = remote
9761×
            .push(bundle, vec![update(state, state2)], None, vec![])
9771×
            .unwrap();
9781×
        assert!(r[0].ok, "applied over what landed meanwhile: {r:?}");
9791×
        let refs = remote.refs().unwrap();
9801×
        assert_eq!(
9811×
            refs.refs[0].heads,
9821×
            vec![h2],
983
            "the store and the mirror both moved"
984
        );
9851×
    }
986
987
    #[test]
9881×
    fn a_store_holds_one_repository() {
9891×
        hermetic();
9901×
        let dir = tempfile::tempdir().unwrap();
9911×
        let root = dir.path();
9921×
        let url = format!("dir://{}", root.join("bucket").display());
9931×
        let mut a = Repo::init(&root.join("a")).unwrap();
9941×
        add_remote(&mut a, "bucket", &url);
9951×
        write(&root.join("a"), "f.txt", b"a\n");
9961×
        a.commit("first", &[]).unwrap();
9971×
        assert!(a.push("bucket", false).unwrap()[0].ok);
9981×
        let mut other = Repo::init(&root.join("other")).unwrap();
9991×
        add_remote(&mut other, "bucket", &url);
10001×
        write(&root.join("other"), "f.txt", b"o\n");
10011×
        other.commit("mine", &[]).unwrap();
10021×
        let e = other.push("bucket", false).unwrap_err().to_string();
10031×
        assert!(
10041×
            e.contains("different repository") || e.contains("another repository"),
1005
            "{e}"
1006
        );
10071×
        assert_eq!(
10081×
            keys(&LocalStore::open(root.join("bucket")).unwrap(), "bundles/").len(),
1009
            1
1010
        );
10111×
    }
1012
1013
    #[test]
10141×
    fn a_damaged_bundle_is_caught_and_an_empty_store_cannot_be_cloned() {
10151×
        hermetic();
10161×
        let dir = tempfile::tempdir().unwrap();
10171×
        let root = dir.path();
10181×
        let url = format!("dir://{}", root.join("bucket").display());
10191×
        let e = Repo::clone(&url, &root.join("nothing"))
10201×
            .err()
10211×
            .expect("refused")
10221×
            .to_string();
10231×
        assert!(e.contains("empty and unbound"), "{e}");
10241×
        let mut a = Repo::init(&root.join("a")).unwrap();
10251×
        add_remote(&mut a, "bucket", &url);
10261×
        write(&root.join("a"), "f.txt", b"a\n");
10271×
        a.commit("first", &[]).unwrap();
10281×
        assert!(a.push("bucket", false).unwrap()[0].ok);
10291×
        let store = LocalStore::open(root.join("bucket")).unwrap();
10301×
        let key = keys(&store, "bundles/").remove(0);
10311×
        let mut bytes = store.get(&key).unwrap();
10321×
        let mid = bytes.len() / 2;
10331×
        bytes[mid] ^= 0xff;
10341×
        store.delete(&key).unwrap();
10351×
        store.put_new(&key, &bytes).unwrap();
10361×
        let e = Repo::clone(&url, &root.join("b"))
10371×
            .err()
10381×
            .expect("refused")
10391×
            .to_string();
10401×
        assert!(
10411×
            e.contains(&key) || e.contains("bundle") || e.contains("hash"),
1042
            "{e}"
1043
        );
10441×
        assert!(!root.join("b/f.txt").exists());
10451×
    }
1046
1047
    /// A store shared by two remotes in one test, with a hook run once at
1048
    /// the first conditional write: what another push landing between a
1049
    /// read of the record and its compare-and-swap looks like.
1050
    struct Shared {
1051
        inner: std::sync::Arc<MemoryStore>,
1052
        fail_uploads: std::sync::atomic::AtomicBool,
1053
        vanish: Mutex<Option<String>>,
1054
    }
1055
1056
    thread_local! {
1057
        static HOOK: std::cell::RefCell<Option<Box<dyn FnOnce()>>> = const { std::cell::RefCell::new(None) };
1058
    }
1059
1060
    impl Shared {
10616×
        fn new(inner: std::sync::Arc<MemoryStore>) -> Box<Shared> {
10626×
            Box::new(Shared {
10636×
                inner,
10646×
                fail_uploads: Default::default(),
10656×
                vanish: Mutex::new(None),
10666×
            })
10676×
        }
1068
    }
1069
1070
    impl Store for Shared {
107124×
        fn describe(&self) -> String {
107224×
            "shared".into()
107324×
        }
10740×
        fn has(&self, key: &str) -> crate::strata::Result<bool> {
10750×
            self.inner.has(key)
10760×
        }
107738×
        fn get(&self, key: &str) -> crate::strata::Result<Vec<u8>> {
107838×
            let mut vanish = self.vanish.lock().unwrap();
107938×
            if vanish.as_deref() == Some(key) {
10801×
                *vanish = None;
10811×
                return Err(StoreError::NotFound(key.to_string()));
108237×
            }
108337×
            self.inner.get(key)
108438×
        }
10850×
        fn get_range(&self, key: &str, range: Range<u64>) -> crate::strata::Result<Vec<u8>> {
10860×
            self.inner.get_range(key, range)
10870×
        }
108811×
        fn put_new(&self, key: &str, bytes: &[u8]) -> crate::strata::Result<()> {
108911×
            if key.starts_with("bundles/")
10908×
                && self.fail_uploads.load(std::sync::atomic::Ordering::SeqCst)
1091
            {
10921×
                return Err(StoreError::Backend("shared: the disk is full".into()));
109310×
            }
109410×
            self.inner.put_new(key, bytes)
109511×
        }
109631×
        fn list(&self, prefix: &str) -> crate::strata::Result<Vec<Entry>> {
109731×
            self.inner.list(prefix)
109831×
        }
10990×
        fn delete(&self, key: &str) -> crate::strata::Result<()> {
11000×
            self.inner.delete(key)
11010×
        }
110223×
        fn read_ref(&self, key: &str) -> crate::strata::Result<Option<(Vec<u8>, Version)>> {
110323×
            self.inner.read_ref(key)
110423×
        }
11058×
        fn write_ref(
11068×
            &self,
11078×
            key: &str,
11088×
            bytes: &[u8],
11098×
            expected: Option<&Version>,
11108×
        ) -> crate::strata::Result<Version> {
11118×
            if key == "refs" && expected.is_some() {
11125×
                if let Some(hook) = HOOK.with(|h| h.borrow_mut().take()) {
11131×
                    hook();
11144×
                }
11153×
            }
11168×
            self.inner.write_ref(key, bytes, expected)
11178×
        }
1118
    }
1119
11203×
    fn repo_with_commit(root: &Path, name: &str, file: &str, text: &[u8], msg: &str) -> Repo {
11213×
        let repo = Repo::init(&root.join(name)).unwrap();
11223×
        write(&root.join(name), file, text);
11233×
        repo.commit(msg, &[]).unwrap();
11243×
        repo
11253×
    }
1126
11276×
    fn remote_for(repo: &Repo, store: Box<dyn Store>, root: &Path, name: &str) -> BucketRemote {
11286×
        let options = RemoteOptions {
11296×
            repo_id: Some(repo.id().unwrap()),
11306×
            cache_dir: Some(root.join(format!("cache-{name}"))),
11316×
            ..Default::default()
11326×
        };
11336×
        BucketRemote::open(store, &options).unwrap()
11346×
    }
1135
11368×
    fn push_bookmark(
11378×
        repo: &Repo,
11388×
        remote: &BucketRemote,
11398×
        bookmark: &str,
11408×
        old: Option<ObjectRef>,
11418×
    ) -> Vec<RefResult> {
11428×
        let heads = repo.heads_of(&repo.view().unwrap(), bookmark).unwrap();
11438×
        let state = repo
11448×
            .store
11458×
            .put_state(&StateObject::new(heads.clone()))
11468×
            .unwrap();
11478×
        let empty = repo.store.put_state(&StateObject::new(vec![])).unwrap();
1148
        // Only what the remote lacks, as push_bookmark sends.
11498×
        let have: Vec<ChangeHash> = remote
11508×
            .refs()
11518×
            .unwrap()
11528×
            .refs
11538×
            .iter()
11548×
            .flat_map(|r| r.heads.clone())
11558×
            .collect();
11568×
        let changes = repo.negotiate(&have, &heads).unwrap();
11578×
        let mut bundle = repo.bundle_for(&changes).unwrap();
11588×
        bundle
11598×
            .objects
11608×
            .push((ObjectType::State, StateObject::new(heads).encode()));
11618×
        let update = RefUpdate {
11628×
            name: bookmark.into(),
11638×
            old_state: old.unwrap_or(empty),
11648×
            new_state: state,
11658×
            force: false,
11668×
            attestation: None,
11678×
        };
11688×
        remote.push(bundle, vec![update], None, vec![]).unwrap()
11698×
    }
1170
1171
    #[test]
11721×
    fn two_pushes_to_different_bookmarks_both_land() {
11731×
        hermetic();
11741×
        let dir = tempfile::tempdir().unwrap();
11751×
        let root = dir.path();
11761×
        let inner = std::sync::Arc::new(MemoryStore::new());
1177
        // A pushes main; B (a clone of the same repository) pushes a
1178
        // second bookmark between A's read and A's compare-and-swap.
11791×
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
11801×
        let ra = remote_for(&a, Shared::new(inner.clone()), root, "a");
11811×
        let r = push_bookmark(&a, &ra, "main", None);
11821×
        assert!(r[0].ok, "{r:?}");
11831×
        let b = Repo::init_with_id(&root.join("b"), a.id().unwrap()).unwrap();
11841×
        let rb = remote_for(&b, Shared::new(inner.clone()), root, "b");
11851×
        rb.refs().unwrap();
11861×
        write(&root.join("b"), "g.txt", b"g\n");
11871×
        b.commit("on b", &[]).unwrap();
11881×
        let (_, op) = b.head_op().unwrap();
11891×
        let mut view = op.view.clone();
11901×
        let feature_heads = b.heads_of(&view, "main").unwrap();
11911×
        view.bookmarks.insert(
11921×
            "feature".into(),
11931×
            b.store.put_state(&StateObject::new(feature_heads)).unwrap(),
1194
        );
11951×
        b.record_op(&[b.head_op().unwrap().0], "bookmark create feature", view)
11961×
            .unwrap();
1197
        // A's second push to main; B's push of feature lands in between.
11981×
        write(&root.join("a"), "f.txt", b"aa\n");
11991×
        a.commit("second", &[]).unwrap();
12001×
        let a_old = ra
12011×
            .refs()
12021×
            .unwrap()
12031×
            .refs
12041×
            .iter()
12051×
            .find(|r| r.name == "main")
12061×
            .unwrap()
1207
            .state;
12081×
        let b_remote = std::sync::Arc::new(std::sync::Mutex::new(Some(rb)));
12091×
        let b_handle = b_remote.clone();
12101×
        HOOK.with(|h| {
12111×
            *h.borrow_mut() = Some(Box::new(move || {
12121×
                let rb = b_handle.lock().unwrap().take().unwrap();
12131×
                let r = push_bookmark(&b, &rb, "feature", None);
12141×
                assert!(r[0].ok, "{r:?}");
12151×
            }))
12161×
        });
12171×
        let r = push_bookmark(&a, &ra, "main", Some(a_old));
12181×
        assert!(r[0].ok, "a's push is applied over b's: {r:?}");
12191×
        let refs = ra.refs().unwrap();
12202×
        let names: Vec<&str> = refs.refs.iter().map(|r| r.name.as_str()).collect();
12211×
        assert_eq!(names, ["feature", "main"], "both landed");
12221×
        assert_eq!(inner.list("bundles/").unwrap().len(), 3);
12231×
        let _ = b_remote;
12241×
    }
1225
1226
    #[test]
12271×
    fn an_interrupted_push_leaves_the_store_usable() {
12281×
        hermetic();
12291×
        let dir = tempfile::tempdir().unwrap();
12301×
        let root = dir.path();
12311×
        let inner = std::sync::Arc::new(MemoryStore::new());
12321×
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
12331×
        let shared = Shared::new(inner.clone());
12341×
        let fail = &shared.fail_uploads as *const std::sync::atomic::AtomicBool;
12351×
        let ra = remote_for(&a, shared, root, "a");
12361×
        assert!(push_bookmark(&a, &ra, "main", None)[0].ok);
12371×
        let old = ra.refs().unwrap().refs[0].state;
1238
        // The upload fails: nothing in the store changes, the mirror is
1239
        // honest, and the next push gets through.
12401×
        write(&root.join("a"), "f.txt", b"aa\n");
12411×
        let h2 = a.commit("second", &[]).unwrap();
12421×
        unsafe { (*fail).store(true, std::sync::atomic::Ordering::SeqCst) };
12431×
        let e = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
12441×
            push_bookmark(&a, &ra, "main", Some(old))
12451×
        }));
12461×
        assert!(e.is_err(), "the push reports the failed upload");
12471×
        assert_eq!(inner.list("bundles/").unwrap().len(), 1);
12481×
        let refs = ra.refs().unwrap();
12491×
        assert_eq!(refs.refs[0].state, old, "the record did not move");
12501×
        unsafe { (*fail).store(false, std::sync::atomic::Ordering::SeqCst) };
12511×
        let r = push_bookmark(&a, &ra, "main", Some(old));
12521×
        assert!(r[0].ok, "{r:?}");
12531×
        assert_eq!(ra.refs().unwrap().refs[0].heads, vec![h2]);
12541×
    }
1255
1256
    #[test]
12571×
    fn bundles_out_of_order_or_gone_are_handled() {
12581×
        hermetic();
12591×
        let dir = tempfile::tempdir().unwrap();
12601×
        let root = dir.path();
12611×
        let inner = std::sync::Arc::new(MemoryStore::new());
12621×
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
12631×
        let ra = remote_for(&a, Shared::new(inner.clone()), root, "a");
12641×
        assert!(push_bookmark(&a, &ra, "main", None)[0].ok);
12651×
        let old = ra.refs().unwrap().refs[0].state;
12661×
        write(&root.join("a"), "f.txt", b"aa\n");
12671×
        let h2 = a.commit("second", &[]).unwrap();
12681×
        assert!(push_bookmark(&a, &ra, "main", Some(old))[0].ok);
1269
        // The second bundle (which needs the first) renamed to sort first.
12701×
        let keys: Vec<String> = inner
12711×
            .list("bundles/")
12721×
            .unwrap()
12731×
            .into_iter()
12741×
            .map(|e| e.key)
12751×
            .collect();
12761×
        assert_eq!(keys.len(), 2);
12771×
        let bytes = inner.get(&keys[1]).unwrap();
12781×
        inner.delete(&keys[1]).unwrap();
12791×
        inner
12801×
            .put_new("bundles/00000000-first.bundle", &bytes)
12811×
            .unwrap();
12821×
        let shared = Shared::new(inner.clone());
12831×
        let c = Repo::init_with_id(&root.join("c"), a.id().unwrap()).unwrap();
12841×
        let rc = remote_for(&c, shared, root, "c");
12851×
        assert_eq!(
12861×
            rc.refs().unwrap().refs[0].heads,
12871×
            vec![h2],
1288
            "the dependent bundle waited for its dependency"
1289
        );
1290
        // A bundle gone between the listing and the read is skipped.
12911×
        let shared = Shared::new(inner.clone());
12921×
        *shared.vanish.lock().unwrap() = Some(keys[0].clone());
12931×
        let d = Repo::init_with_id(&root.join("d"), a.id().unwrap()).unwrap();
12941×
        let rd = remote_for(&d, shared, root, "d");
1295
        // Without the first bundle the second cannot be imported: the sync
1296
        // says so rather than failing silently.
12971×
        let e = rd.refs().expect_err("missing changes").to_string();
12981×
        assert!(e.contains("cannot be imported"), "{e}");
12991×
    }
1300
1301
    #[test]
13021×
    fn gc_merges_bundles_and_prunes_after_the_grace_period() {
13031×
        hermetic();
13041×
        let dir = tempfile::tempdir().unwrap();
13051×
        let root = dir.path();
13061×
        let url = format!("dir://{}", root.join("bucket").display());
13071×
        let store = LocalStore::open(root.join("bucket")).unwrap();
13081×
        let mut a = Repo::init(&root.join("a")).unwrap();
13091×
        add_remote(&mut a, "bucket", &url);
13103×
        for i in 1..=3 {
13113×
            write(&root.join("a"), "f.txt", format!("{i}\n").as_bytes());
13123×
            a.commit(&format!("change {i}"), &[]).unwrap();
13133×
            assert!(a.push("bucket", false).unwrap()[0].ok);
1314
        }
13151×
        a.tag_create("v3", &a.bookmark_heads().unwrap(), "three")
13161×
            .unwrap();
13171×
        assert!(a.push("bucket", false).unwrap().iter().all(|r| r.ok));
13181×
        assert_eq!(
13191×
            keys(&store, "bundles/").len(),
1320
            4,
1321
            "three pushes and the tag's"
1322
        );
1323
        // A gc folds the four into one; the old ones wait out the grace.
13241×
        let r = a.gc_remote("bucket", false).unwrap();
13251×
        assert_eq!(
13261×
            (r.merged, r.changes, r.deleted, r.deferred),
1327
            (4, 3, 0, 4),
1328
            "{r:?}"
1329
        );
13301×
        assert!(r.bundle_bytes > 0);
13311×
        assert_eq!(keys(&store, "bundles/").len(), 5);
13321×
        let b = Repo::clone(&url, &root.join("b")).unwrap();
13331×
        assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"3\n");
13341×
        assert_eq!(b.tags().unwrap().len(), 1);
1335
        // Past the grace period they go, and a clone needs only the one.
13361×
        let r = a.gc_remote("bucket", true).unwrap();
13371×
        assert_eq!((r.merged, r.deleted, r.deferred), (0, 4, 0), "{r:?}");
13381×
        assert_eq!(keys(&store, "bundles/").len(), 1);
13391×
        let c = Repo::clone(&url, &root.join("c")).unwrap();
13401×
        assert_eq!(c.bookmark_heads().unwrap(), a.bookmark_heads().unwrap());
1341
        // The earlier clone, which saw the old bundles, carries on.
13421×
        write(&root.join("b"), "f.txt", b"4\n");
13431×
        b.commit("four", &[]).unwrap();
13441×
        assert!(b.push("origin", false).unwrap()[0].ok);
13451×
        assert!(a.pull("bucket").unwrap().changed);
13461×
        assert_eq!(fs::read(root.join("a/f.txt")).unwrap(), b"4\n");
1347
        // Nothing more to merge: a second gc is a no-op.
13481×
        let r = a.gc_remote("bucket", true).unwrap();
13491×
        assert_eq!((r.merged, r.deleted), (2, 2));
13501×
        assert!(a.gc_remote("nope", true).is_err());
13511×
    }
1352
1353
    #[test]
13541×
    fn markers_parse_and_refuse_newer_formats() {
13551×
        let id = [7u8; 16];
13561×
        assert_eq!(parse_marker(&marker(id), "s").unwrap(), id);
13571×
        let e = parse_marker(
13581×
            b"format = 9\nrepo = \"07070707070707070707070707070707\"\n",
13591×
            "s",
13601×
        )
13611×
        .unwrap_err()
13621×
        .to_string();
13631×
        assert!(e.contains("format 9"), "{e}");
13641×
        assert!(parse_marker(b"hello", "s").is_err());
13651×
    }
1366
}