OXYGENHUB
◆ v0.4.6 oxygen / crates / oxy-core / src / repo / strata.rs
51.4 KiB · Rust · last changed Strata: a push lands over what landed meanwhile; oxy gc --remote 1 month ago source blame history raw
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 {
47
    fn from(e: StoreError) -> RepoError {
48
        RepoError::Msg(e.to_string())
49
    }
50
}
51
52
/// The `oxygen.toml` marker: two lines, written once.
53
fn marker(id: [u8; 16]) -> Vec<u8> {
54
    format!("format = {FORMAT}\nrepo = \"{}\"\n", hex::encode(id)).into_bytes()
55
}
56
57
fn parse_marker(bytes: &[u8], what: &str) -> Result<[u8; 16]> {
58
    let text = String::from_utf8_lossy(bytes);
59
    let mut format = None;
60
    let mut repo = None;
61
    for line in text.lines() {
62
        let Some((k, v)) = line.split_once('=') else {
63
            continue;
64
        };
65
        let v = v.trim().trim_matches('"');
66
        match k.trim() {
67
            "format" => format = v.parse::<u32>().ok(),
68
            "repo" => {
69
                repo = hex::decode(v)
70
                    .ok()
71
                    .and_then(|b| <[u8; 16]>::try_from(b).ok())
72
            }
73
            _ => {}
74
        }
75
    }
76
    match (format, repo) {
77
        (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
        )),
80
        (Some(_), Some(id)) => Ok(id),
81
        _ => fail(format!("{what}: its oxygen.toml is not a Strata marker")),
82
    }
83
}
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 {
98
    fn drop(&mut self) {
99
        if self.temporary {
100
            let _ = fs::remove_dir_all(&self.dir);
101
        }
102
    }
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 {
113
    pub fn open(store: Box<dyn Store>, options: &RemoteOptions) -> Result<BucketRemote> {
114
        let (dir, temporary) = match &options.cache_dir {
115
            Some(d) => (d.clone(), false),
116
            None => {
117
                let mut nonce = [0u8; 6];
118
                rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut nonce);
119
                (
120
                    std::env::temp_dir().join(format!(
121
                        "oxy-strata-{}-{}",
122
                        std::process::id(),
123
                        hex::encode(nonce)
124
                    )),
125
                    true,
126
                )
127
            }
128
        };
129
        Ok(BucketRemote {
130
            store,
131
            dir,
132
            temporary,
133
            claimed_id: options.repo_id,
134
            mirror: RefCell::new(None),
135
        })
136
    }
137
138
    fn what(&self) -> String {
139
        self.store.describe()
140
    }
141
142
    /// The repository id the store holds: `None` until a first push.
143
    fn stored_id(&self) -> Result<Option<[u8; 16]>> {
144
        match self.store.get(MARKER_KEY) {
145
            Ok(b) => Ok(Some(parse_marker(&b, &self.what())?)),
146
            Err(StoreError::NotFound(_)) => Ok(None),
147
            Err(e) => Err(e.into()),
148
        }
149
    }
150
151
    fn mirror_dir(&self) -> PathBuf {
152
        self.dir.join("mirror")
153
    }
154
155
    fn seen_path(&self) -> PathBuf {
156
        self.dir.join("seen")
157
    }
158
159
    /// The bundles already imported into the mirror.
160
    fn seen(&self) -> BTreeSet<String> {
161
        fs::read_to_string(self.seen_path())
162
            .map(|s| s.lines().map(str::to_string).collect())
163
            .unwrap_or_default()
164
    }
165
166
    fn mark_seen(&self, key: &str) -> Result<()> {
167
        let mut seen = self.seen();
168
        if seen.insert(key.to_string()) {
169
            let text: String = seen.iter().map(|k| format!("{k}\n")).collect();
170
            crate::durable::write_atomic(&self.seen_path(), text.as_bytes())?;
171
        }
172
        Ok(())
173
    }
174
175
    /// Open the mirror for repository `id`, creating it on first use.
176
    fn open_mirror(&self, id: [u8; 16]) -> Result<()> {
177
        if self.mirror.borrow().is_some() {
178
            return Ok(());
179
        }
180
        let dir = self.mirror_dir();
181
        let repo = if dir.join(super::OXY_DIR).is_dir() {
182
            let repo = Repo::open(&dir)?;
183
            if repo.id()? != id {
184
                // The store was emptied and reused for another repository:
185
                // start the mirror over.
186
                drop(repo);
187
                fs::remove_dir_all(&self.dir)?;
188
                fs::create_dir_all(&self.dir)?;
189
                Repo::init_bare_with_id(&dir, id)?
190
            } else {
191
                repo
192
            }
193
        } else {
194
            fs::create_dir_all(&self.dir)?;
195
            Repo::init_bare_with_id(&dir, id)?
196
        };
197
        *self.mirror.borrow_mut() = Some(repo);
198
        Ok(())
199
    }
200
201
    fn with_mirror<T>(&self, f: impl FnOnce(&Repo) -> Result<T>) -> Result<T> {
202
        let m = self.mirror.borrow();
203
        match m.as_ref() {
204
            Some(repo) => f(repo),
205
            None => fail(format!(
206
                "{} holds no repository yet: push one to it first",
207
                self.what()
208
            )),
209
        }
210
    }
211
212
    /// Bring the mirror level with the store.
213
    fn sync(&self) -> Result<Synced> {
214
        let Some(id) = self.stored_id()? else {
215
            return Ok(Synced {
216
                version: None,
217
                id: None,
218
            });
219
        };
220
        self.open_mirror(id)?;
221
        let record = self.store.read_ref(REFS_KEY)?;
222
        let (resp, version) = match record {
223
            Some((bytes, version)) => {
224
                let (rid, resp) = wire::decode_refs(&bytes)?;
225
                if rid != id {
226
                    return fail(format!(
227
                        "{}: its refs record names a different repository than its oxygen.toml; the store is damaged",
228
                        self.what()
229
                    ));
230
                }
231
                (resp, Some(version))
232
            }
233
            None => (RefsResponse::default(), None),
234
        };
235
        self.with_mirror(|mirror| {
236
            // The policy first: bundles are then imported under it.
237
            if let Some(sp) = &resp.policy {
238
                let have = mirror.current_policy()?.map(|(_, p)| p.policy.serial);
239
                if have.is_none_or(|s| s < sp.policy.serial) {
240
                    mirror
241
                        .adopt_policy_chain(&resp.history, sp, "strata sync")
242
                        .map_err(|e| RepoError::Msg(format!("policy from {}: {e}", self.what())))?;
243
                }
244
            }
245
            self.import_new_bundles(mirror)?;
246
            self.set_view(mirror, &resp)?;
247
            Ok(())
248
        })?;
249
        Ok(Synced {
250
            version,
251
            id: Some(id),
252
        })
253
    }
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.
259
    fn import_new_bundles(&self, mirror: &Repo) -> Result<()> {
260
        let seen = self.seen();
261
        let mut pending: Vec<String> = self
262
            .store
263
            .list(BUNDLES_PREFIX)?
264
            .into_iter()
265
            .map(|e| e.key)
266
            .filter(|k| !seen.contains(k))
267
            .collect();
268
        let strict = mirror.current_policy()?.is_some();
269
        while !pending.is_empty() {
270
            let mut deferred = Vec::new();
271
            let mut last_error = None;
272
            let before = pending.len();
273
            for key in pending.drain(..) {
274
                let bytes = match self.store.get(&key) {
275
                    Ok(b) => b,
276
                    // Listed, then gone: a gc merged it into a newer
277
                    // bundle, which this listing holds too.
278
                    Err(StoreError::NotFound(_)) => continue,
279
                    Err(e) => return Err(e.into()),
280
                };
281
                let bundle = wire::decode_bundle(&bytes)
282
                    .map_err(|e| RepoError::Msg(format!("{}: {key}: {e}", self.what())))?;
283
                match mirror.import_bundle(&bundle, strict) {
284
                    Ok(_) => self.mark_seen(&key)?,
285
                    Err(RepoError::Msg(m)) if m.contains("neither stored nor in the bundle") => {
286
                        last_error = Some(m);
287
                        deferred.push(key);
288
                    }
289
                    Err(e) => return Err(RepoError::Msg(format!("{}: {key}: {e}", self.what()))),
290
                }
291
            }
292
            if deferred.len() == before {
293
                // No progress: something those bundles need was never
294
                // pushed, or a push is still in progress.
295
                return fail(format!(
296
                    "{}: {} cannot be imported ({}); a push may be in progress, so try again, else the store is missing changes",
297
                    self.what(),
298
                    deferred.join(", "),
299
                    last_error.unwrap_or_default()
300
                ));
301
            }
302
            pending = deferred;
303
        }
304
        Ok(())
305
    }
306
307
    /// Make the mirror's view say what the record says.
308
    fn set_view(&self, mirror: &Repo, resp: &RefsResponse) -> Result<()> {
309
        let (op_ref, op) = mirror.head_op()?;
310
        let mut view = op.view.clone();
311
        view.bookmarks.clear();
312
        for r in &resp.refs {
313
            for h in &r.heads {
314
                if !mirror.store.has(h) {
315
                    return fail(format!(
316
                        "{}: bookmark {:?} names change {}, which no bundle there holds; a push may be in progress, so try again",
317
                        self.what(),
318
                        r.name,
319
                        h.short(12)
320
                    ));
321
                }
322
            }
323
            let state = mirror.store.put_state(&StateObject::new(r.heads.clone()))?;
324
            if state != r.state {
325
                return fail(format!(
326
                    "{}: bookmark {:?} names a state that does not match its heads; the store is damaged",
327
                    self.what(),
328
                    r.name
329
                ));
330
            }
331
            if let Some(sa) = &r.attestation {
332
                let current = mirror.latest_ref_attest(&r.name)?.map(|(_, a)| a);
333
                if current.as_ref() != Some(sa) {
334
                    let a = mirror.store.put_attest(&sa.attest)?;
335
                    for sig in &sa.sigs {
336
                        mirror.store.put_signature(&a, sig)?;
337
                    }
338
                    mirror.set_ref_attest(&r.name, &a)?;
339
                }
340
            }
341
            view.bookmarks.insert(r.name.clone(), state);
342
        }
343
        let policy = mirror.current_policy()?.map(|(_, sp)| sp.policy);
344
        for t in &resp.tags {
345
            mirror.store.put_state(&StateObject::new(t.heads.clone()))?;
346
            mirror
347
                .adopt_tag(&mut view, t, policy.as_ref())
348
                .map_err(|e| RepoError::Msg(format!("{}: {e}", self.what())))?;
349
        }
350
        if view != op.view {
351
            mirror.record_op(&[op_ref], "strata sync", view)?;
352
        }
353
        Ok(())
354
    }
355
356
    /// The record for the mirror's current references.
357
    fn record(&self, id: [u8; 16]) -> Result<Vec<u8>> {
358
        self.with_mirror(|mirror| {
359
            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.
362
            resp.default_bookmark = None;
363
            Ok(wire::encode_refs(id, &resp))
364
        })
365
    }
366
367
    /// Store a bundle under the next sequence number; returns its key.
368
    fn upload(&self, bundle: &Bundle) -> Result<String> {
369
        let bytes = wire::encode_bundle(bundle);
370
        let hash = blake3::hash(&bytes).to_hex();
371
        let seq = self.store.list(BUNDLES_PREFIX)?.len() + 1;
372
        let key = format!("{BUNDLES_PREFIX}{seq:08}-{}.bundle", &hash[..16]);
373
        match self.store.put_new(&key, &bytes) {
374
            Ok(()) | Err(StoreError::Exists(_)) => {}
375
            Err(e) => return Err(e.into()),
376
        }
377
        self.mark_seen(&key)?;
378
        Ok(key)
379
    }
380
}
381
382
impl Remote for BucketRemote {
383
    fn repo_id(&self) -> Result<[u8; 16]> {
384
        Ok(self.stored_id()?.unwrap_or(UNBOUND_ID))
385
    }
386
387
    fn refs(&self) -> Result<RefsResponse> {
388
        let synced = self.sync()?;
389
        if synced.id.is_none() {
390
            return Ok(RefsResponse::default());
391
        }
392
        self.with_mirror(|mirror| {
393
            let mut resp = mirror.remote_refs()?;
394
            resp.default_bookmark = None;
395
            Ok(resp)
396
        })
397
    }
398
399
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
400
        self.with_mirror(|m| m.negotiate(have, want))
401
    }
402
403
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
404
        self.with_mirror(|m| m.bundle_raw_for(changes, max_blob))
405
    }
406
407
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
408
        self.with_mirror(|m| m.objects_for(refs))
409
    }
410
411
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
412
        self.with_mirror(|m| m.checkpoint_bundle_for(heads))
413
    }
414
415
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
416
        self.with_mirror(|m| m.shallow_changes(heads, depth))
417
    }
418
419
    fn push(
420
        &self,
421
        bundle: Bundle,
422
        updates: Vec<RefUpdate>,
423
        policy: Option<SignedPolicy>,
424
        tags: Vec<TagRef>,
425
    ) -> Result<Vec<RefResult>> {
426
        let mut synced = self.sync()?;
427
        let id = match synced.id {
428
            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.
433
                let Some(id) = self.claimed_id else {
434
                    return fail(format!(
435
                        "{} is empty, and this client did not say which repository it is",
436
                        self.what()
437
                    ));
438
                };
439
                match self.store.put_new(MARKER_KEY, &marker(id)) {
440
                    Ok(()) => {}
441
                    Err(StoreError::Exists(_)) => {}
442
                    Err(e) => return Err(e.into()),
443
                }
444
                synced = self.sync()?;
445
                match synced.id {
446
                    Some(stored) if stored == id => id,
447
                    _ => {
448
                        return fail(format!(
449
                            "{} was just bound to another repository by a concurrent push",
450
                            self.what()
451
                        ))
452
                    }
453
                }
454
            }
455
        };
456
        if let Some(claimed) = self.claimed_id {
457
            if claimed != id {
458
                return fail(format!(
459
                    "{} holds a different repository (id {}) than this one ({}): a store holds one repository; use another prefix or bucket",
460
                    self.what(),
461
                    hex::encode(id),
462
                    hex::encode(claimed)
463
                ));
464
            }
465
        }
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.
472
        let mut uploaded = false;
473
        for attempt in 0..PUSH_ATTEMPTS {
474
            let (before_ref, before) = self.with_mirror(|m| m.head_op())?;
475
            let results = self.with_mirror(|m| {
476
                m.receive_push(
477
                    bundle.clone(),
478
                    updates.clone(),
479
                    policy.clone(),
480
                    tags.clone(),
481
                )
482
            })?;
483
            if !results.iter().any(|r| r.ok) {
484
                return Ok(results);
485
            }
486
            if !bundle.is_empty() && !uploaded {
487
                self.upload(&bundle)?;
488
                uploaded = true;
489
            }
490
            let record = self.record(id)?;
491
            match self
492
                .store
493
                .write_ref(REFS_KEY, &record, synced.version.as_ref())
494
            {
495
                Ok(_) => return Ok(results),
496
                Err(StoreError::Conflict(_)) => {
497
                    self.with_mirror(|m| {
498
                        let (now, _) = m.head_op()?;
499
                        if now != before_ref {
500
                            m.record_op(
501
                                &[now],
502
                                "strata push undone: the remote moved",
503
                                before.view.clone(),
504
                            )?;
505
                        }
506
                        Ok(())
507
                    })?;
508
                    synced = self.sync()?;
509
                    if attempt + 1 == PUSH_ATTEMPTS {
510
                        return Ok(results
511
                            .into_iter()
512
                            .map(|r| RefResult {
513
                                name: r.name,
514
                                ok: false,
515
                                reason: "the remote kept moving while pushing (other pushes landed first); pull, then push again".into(),
516
                            })
517
                            .collect());
518
                    }
519
                }
520
                Err(e) => return Err(e.into()),
521
            }
522
        }
523
        unreachable!("the loop returns")
524
    }
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
555
fn now_ms() -> u64 {
556
    std::time::SystemTime::now()
557
        .duration_since(std::time::UNIX_EPOCH)
558
        .map(|d| d.as_millis() as u64)
559
        .unwrap_or(0)
560
}
561
562
impl BucketRemote {
563
    /// The bundles a gc superseded, with when: one `KEY\tMILLIS` per line.
564
    fn trash(&self) -> Result<(Trash, Option<Version>)> {
565
        match self.store.read_ref(TRASH_KEY)? {
566
            Some((bytes, v)) => {
567
                let text = String::from_utf8_lossy(&bytes);
568
                let list = text
569
                    .lines()
570
                    .filter_map(|l| {
571
                        let (k, t) = l.split_once('\t')?;
572
                        Some((k.to_string(), t.parse().ok()?))
573
                    })
574
                    .collect();
575
                Ok((list, Some(v)))
576
            }
577
            None => Ok((Vec::new(), None)),
578
        }
579
    }
580
581
    fn write_trash(&self, list: &[(String, u64)], expected: Option<&Version>) -> Result<Version> {
582
        let text: String = list.iter().map(|(k, t)| format!("{k}\t{t}\n")).collect();
583
        Ok(self.store.write_ref(TRASH_KEY, text.as_bytes(), expected)?)
584
    }
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).
589
    pub fn gc(&self, grace: std::time::Duration) -> Result<RemoteGc> {
590
        let synced = self.sync()?;
591
        let Some(id) = synced.id else {
592
            return fail(format!("{} holds no repository yet", self.what()));
593
        };
594
        let mut report = RemoteGc::default();
595
        let existing: Vec<String> = self
596
            .store
597
            .list(BUNDLES_PREFIX)?
598
            .into_iter()
599
            .map(|e| e.key)
600
            .collect();
601
        let (mut trash, mut trash_version) = self.trash()?;
602
        let live: Vec<&String> = existing
603
            .iter()
604
            .filter(|k| !trash.iter().any(|(t, _)| t == *k))
605
            .collect();
606
        if live.len() > 1 {
607
            // Everything the record reaches, in dependency order, with
608
            // the state objects the bookmarks and tags name.
609
            let (bundle, changes) = self.with_mirror(|m| {
610
                let resp = m.remote_refs()?;
611
                let mut heads: Vec<ChangeHash> = Vec::new();
612
                for r in &resp.refs {
613
                    heads.extend(r.heads.iter().copied());
614
                }
615
                for t in &resp.tags {
616
                    heads.extend(t.heads.iter().copied());
617
                }
618
                heads.sort();
619
                heads.dedup();
620
                let changes = m.negotiate(&[], &heads)?;
621
                let mut bundle = m.bundle_for(&changes)?;
622
                for r in &resp.refs {
623
                    bundle.objects.push((
624
                        ObjectType::State,
625
                        StateObject::new(r.heads.clone()).encode(),
626
                    ));
627
                }
628
                for t in &resp.tags {
629
                    bundle.objects.push((
630
                        ObjectType::State,
631
                        StateObject::new(t.heads.clone()).encode(),
632
                    ));
633
                }
634
                Ok((bundle, changes.len()))
635
            })?;
636
            report.changes = changes;
637
            report.bundle_bytes = wire::encode_bundle(&bundle).len();
638
            let merged_key = self.upload(&bundle)?;
639
            let now = now_ms();
640
            for k in live {
641
                if *k != merged_key {
642
                    trash.push((k.clone(), now));
643
                    report.merged += 1;
644
                }
645
            }
646
            trash_version = Some(self.write_trash(&trash, trash_version.as_ref())?);
647
        }
648
        // What the grace period has released.
649
        let now = now_ms();
650
        let mut kept = Vec::new();
651
        for (k, when) in trash {
652
            if now.saturating_sub(when) >= grace.as_millis() as u64 {
653
                self.store.delete(&k)?;
654
                report.deleted += 1;
655
            } else {
656
                kept.push((k, when));
657
                report.deferred += 1;
658
            }
659
        }
660
        if report.deleted > 0 {
661
            self.write_trash(&kept, trash_version.as_ref())?;
662
        }
663
        let _ = id;
664
        Ok(report)
665
    }
666
}
667
668
impl Repo {
669
    /// `oxy gc --remote NAME`: a gc against a remote kept on a store.
670
    pub fn gc_remote(&self, name: &str, now: bool) -> Result<RemoteGc> {
671
        let Some(url) = self.config.remotes.get(name).cloned() else {
672
            return fail(format!("no remote {name:?}"));
673
        };
674
        let options = self.remote_options(name);
675
        let Some(store) = crate::protocol::open_store(&url, &options) else {
676
            return fail(format!(
677
                "gc --remote works on a remote kept on a store (s3://, gs://, az://, dir://); {name} is {url}"
678
            ));
679
        };
680
        let remote = BucketRemote::open(store?, &options)?;
681
        let grace = if now {
682
            std::time::Duration::ZERO
683
        } else {
684
            REMOTE_GC_GRACE
685
        };
686
        remote.gc(grace)
687
    }
688
}
689
690
/// Open a remote on `store` for the local repository described by `options`.
691
pub fn open_store_remote(
692
    store: Box<dyn Store>,
693
    options: &RemoteOptions,
694
) -> Result<Box<dyn Remote>> {
695
    Ok(Box::new(BucketRemote::open(store, options)?))
696
}
697
698
/// Whether `path` holds a Strata store rather than a repository.
699
pub fn is_store_dir(path: &Path) -> bool {
700
    path.join(MARKER_KEY).is_file() || path.join(REFS_KEY).is_file()
701
}
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
711
    fn hermetic() {
712
        static ONCE: std::sync::Once = std::sync::Once::new();
713
        static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();
714
        ONCE.call_once(|| {
715
            let d = tempfile::tempdir().unwrap();
716
            std::env::set_var("OXY_HOME", d.path());
717
            DIR.set(d).ok();
718
        });
719
    }
720
721
    fn write(root: &Path, rel: &str, content: &[u8]) {
722
        let p = root.join(rel);
723
        fs::create_dir_all(p.parent().unwrap()).unwrap();
724
        fs::write(p, content).unwrap();
725
    }
726
727
    fn add_remote(repo: &mut Repo, name: &str, url: &str) {
728
        repo.config.set(&format!("remote.{name}.url"), url).unwrap();
729
        repo.save_config().unwrap();
730
    }
731
732
    fn keys(store: &dyn Store, prefix: &str) -> Vec<String> {
733
        store
734
            .list(prefix)
735
            .unwrap()
736
            .into_iter()
737
            .map(|e| e.key)
738
            .collect()
739
    }
740
741
    #[test]
742
    fn push_clone_and_pull_through_a_directory_store() {
743
        hermetic();
744
        let dir = tempfile::tempdir().unwrap();
745
        let root = dir.path();
746
        let url = format!("dir://{}", root.join("bucket").display());
747
        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.
751
        let mut a = Repo::init(&root.join("a"))
752
            .unwrap()
753
            .with_home(root.join("home-a"));
754
        a.create_identity("Ada", "[email protected]", IdentityScope::Repo)
755
            .unwrap();
756
        add_remote(&mut a, "bucket", &url);
757
        write(&root.join("a"), "f.txt", b"a\n");
758
        let h1 = a.commit("first", &[]).unwrap();
759
        a.tag_create("v1", &[h1], "one").unwrap();
760
        let results = a.push("bucket", false).unwrap();
761
        assert!(results.iter().all(|r| r.ok), "{results:?}");
762
        assert_eq!(
763
            keys(&store, ""),
764
            [
765
                "bundles/00000001-".to_string() + &keys(&store, "bundles/")[0][17..],
766
                "oxygen.toml".into(),
767
                "refs".into()
768
            ]
769
        );
770
        assert_eq!(
771
            parse_marker(&store.get("oxygen.toml").unwrap(), "t").unwrap(),
772
            a.id().unwrap()
773
        );
774
        let again = a.push("bucket", false).unwrap();
775
        assert!(
776
            again[0].ok && again[0].reason.contains("up to date"),
777
            "{again:?}"
778
        );
779
        assert_eq!(
780
            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.
787
        let b = Repo::clone(&url, &root.join("b")).unwrap();
788
        assert_eq!(b.id().unwrap(), a.id().unwrap());
789
        assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"a\n");
790
        assert_eq!(b.bookmark_heads().unwrap(), vec![h1]);
791
        assert_eq!(
792
            b.tags()
793
                .unwrap()
794
                .iter()
795
                .map(|t| t.name.clone())
796
                .collect::<Vec<_>>(),
797
            ["v1"]
798
        );
799
        assert!(matches!(
800
            b.authenticate_change(&h1).unwrap(),
801
            crate::repo::SignatureStatus::Valid { sealed: true, .. }
802
        ));
803
        assert!(
804
            root.join("b/.oxy/remotes/origin/strata/mirror/.oxy")
805
                .is_dir(),
806
            "the mirror lives with the remote's state"
807
        );
808
        assert!(
809
            !std::env::temp_dir()
810
                .read_dir()
811
                .unwrap()
812
                .flatten()
813
                .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.
818
        write(&root.join("b"), "f.txt", b"a\nb\n");
819
        let h2 = b.commit("second", &[]).unwrap();
820
        assert!(b.push("origin", false).unwrap()[0].ok);
821
        assert_eq!(keys(&store, "bundles/").len(), 2);
822
        let pr = a.pull("bucket").unwrap();
823
        assert!(pr.changed && pr.checked_out);
824
        assert_eq!(fs::read(root.join("a/f.txt")).unwrap(), b"a\nb\n");
825
        write(&root.join("a"), "g.txt", b"g\n");
826
        let h3 = a.commit("third", &[]).unwrap();
827
        assert!(a.push("bucket", false).unwrap()[0].ok);
828
        let pr = b.pull("origin").unwrap();
829
        assert!(pr.changed);
830
        // g.txt is new, so its change commutes with B's: two heads.
831
        let mut want = vec![h2, h3];
832
        want.sort();
833
        assert_eq!(b.bookmark_heads().unwrap(), want);
834
        assert_eq!(fs::read(root.join("b/g.txt")).unwrap(), b"g\n");
835
836
        // A third clone gets everything from the three bundles.
837
        let c = Repo::clone(&url, &root.join("c")).unwrap();
838
        assert_eq!(c.bookmark_heads().unwrap(), want);
839
        assert_eq!(fs::read(root.join("c/f.txt")).unwrap(), b"a\nb\n");
840
    }
841
842
    #[test]
843
    fn a_push_behind_the_remote_is_refused_before_anything_is_uploaded() {
844
        hermetic();
845
        let dir = tempfile::tempdir().unwrap();
846
        let root = dir.path();
847
        let url = format!("dir://{}", root.join("bucket").display());
848
        let store = LocalStore::open(root.join("bucket")).unwrap();
849
        let mut a = Repo::init(&root.join("a")).unwrap();
850
        add_remote(&mut a, "bucket", &url);
851
        write(&root.join("a"), "f.txt", b"a\n");
852
        a.commit("first", &[]).unwrap();
853
        assert!(a.push("bucket", false).unwrap()[0].ok);
854
        let b = Repo::clone(&url, &root.join("b")).unwrap();
855
        write(&root.join("b"), "f.txt", b"b\n");
856
        b.commit("from b", &[]).unwrap();
857
        assert!(b.push("origin", false).unwrap()[0].ok);
858
        write(&root.join("a"), "g.txt", b"g\n");
859
        a.commit("from a", &[]).unwrap();
860
        let e = a.push("bucket", false).unwrap_err().to_string();
861
        assert!(e.contains("pull first"), "{e}");
862
        assert_eq!(keys(&store, "bundles/").len(), 2);
863
        assert!(a.pull("bucket").unwrap().changed);
864
        assert!(a.push("bucket", false).unwrap()[0].ok);
865
        assert_eq!(keys(&store, "bundles/").len(), 3);
866
    }
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 {
877
        fn describe(&self) -> String {
878
            "racing".into()
879
        }
880
        fn has(&self, key: &str) -> crate::strata::Result<bool> {
881
            self.inner.has(key)
882
        }
883
        fn get(&self, key: &str) -> crate::strata::Result<Vec<u8>> {
884
            self.inner.get(key)
885
        }
886
        fn get_range(&self, key: &str, range: Range<u64>) -> crate::strata::Result<Vec<u8>> {
887
            self.inner.get_range(key, range)
888
        }
889
        fn put_new(&self, key: &str, bytes: &[u8]) -> crate::strata::Result<()> {
890
            self.inner.put_new(key, bytes)
891
        }
892
        fn list(&self, prefix: &str) -> crate::strata::Result<Vec<Entry>> {
893
            self.inner.list(prefix)
894
        }
895
        fn delete(&self, key: &str) -> crate::strata::Result<()> {
896
            self.inner.delete(key)
897
        }
898
        fn read_ref(&self, key: &str) -> crate::strata::Result<Option<(Vec<u8>, Version)>> {
899
            self.inner.read_ref(key)
900
        }
901
        fn write_ref(
902
            &self,
903
            key: &str,
904
            bytes: &[u8],
905
            expected: Option<&Version>,
906
        ) -> crate::strata::Result<Version> {
907
            let mut raced = self.raced.lock().unwrap();
908
            if !*raced && expected.is_some() {
909
                *raced = true;
910
                // Someone else lands a record: the same references with a
911
                // different detail, so the version moves.
912
                let (current, v) = self.inner.read_ref(key)?.unwrap();
913
                let (id, mut resp) =
914
                    wire::decode_refs(&current).map_err(|e| StoreError::Backend(e.to_string()))?;
915
                resp.default_bookmark = Some("raced".into());
916
                self.inner
917
                    .write_ref(key, &wire::encode_refs(id, &resp), Some(&v))?;
918
            }
919
            self.inner.write_ref(key, bytes, expected)
920
        }
921
    }
922
923
    #[test]
924
    fn a_record_moved_during_a_push_is_applied_over_what_landed() {
925
        hermetic();
926
        let dir = tempfile::tempdir().unwrap();
927
        let root = dir.path();
928
        let a = Repo::init(&root.join("a")).unwrap();
929
        write(&root.join("a"), "f.txt", b"a\n");
930
        let h1 = a.commit("first", &[]).unwrap();
931
        let store = Box::new(Racing {
932
            inner: MemoryStore::new(),
933
            raced: Mutex::new(false),
934
        });
935
        let options = RemoteOptions {
936
            repo_id: Some(a.id().unwrap()),
937
            cache_dir: Some(root.join("cache")),
938
            ..Default::default()
939
        };
940
        let remote = BucketRemote::open(store, &options).unwrap();
941
        // The first push creates the record (expected None: no race yet).
942
        let state = a.store.put_state(&StateObject::new(vec![h1])).unwrap();
943
        let empty = a.store.put_state(&StateObject::new(vec![])).unwrap();
944
        let mut bundle = a.bundle_for(&[h1]).unwrap();
945
        bundle
946
            .objects
947
            .push((ObjectType::State, StateObject::new(vec![h1]).encode()));
948
        let update = |old, new| RefUpdate {
949
            name: "main".into(),
950
            old_state: old,
951
            new_state: new,
952
            force: false,
953
            attestation: None,
954
        };
955
        let r = remote
956
            .push(bundle, vec![update(empty, state)], None, vec![])
957
            .unwrap();
958
        assert!(r[0].ok, "{r:?}");
959
        // The second push meets the race: refused, nothing moved.
960
        write(&root.join("a"), "f.txt", b"ab\n");
961
        let h2 = a.commit("second", &[]).unwrap();
962
        let state2 = a.store.put_state(&StateObject::new(vec![h2])).unwrap();
963
        let mut bundle = a.bundle_for(&[h2]).unwrap();
964
        bundle
965
            .objects
966
            .push((ObjectType::State, StateObject::new(vec![h2]).encode()));
967
        let r = remote
968
            .push(bundle, vec![update(state, state2)], None, vec![])
969
            .unwrap();
970
        assert!(r[0].ok, "applied over what landed meanwhile: {r:?}");
971
        let refs = remote.refs().unwrap();
972
        assert_eq!(
973
            refs.refs[0].heads,
974
            vec![h2],
975
            "the store and the mirror both moved"
976
        );
977
    }
978
979
    #[test]
980
    fn a_store_holds_one_repository() {
981
        hermetic();
982
        let dir = tempfile::tempdir().unwrap();
983
        let root = dir.path();
984
        let url = format!("dir://{}", root.join("bucket").display());
985
        let mut a = Repo::init(&root.join("a")).unwrap();
986
        add_remote(&mut a, "bucket", &url);
987
        write(&root.join("a"), "f.txt", b"a\n");
988
        a.commit("first", &[]).unwrap();
989
        assert!(a.push("bucket", false).unwrap()[0].ok);
990
        let mut other = Repo::init(&root.join("other")).unwrap();
991
        add_remote(&mut other, "bucket", &url);
992
        write(&root.join("other"), "f.txt", b"o\n");
993
        other.commit("mine", &[]).unwrap();
994
        let e = other.push("bucket", false).unwrap_err().to_string();
995
        assert!(
996
            e.contains("different repository") || e.contains("another repository"),
997
            "{e}"
998
        );
999
        assert_eq!(
1000
            keys(&LocalStore::open(root.join("bucket")).unwrap(), "bundles/").len(),
1001
            1
1002
        );
1003
    }
1004
1005
    #[test]
1006
    fn a_damaged_bundle_is_caught_and_an_empty_store_cannot_be_cloned() {
1007
        hermetic();
1008
        let dir = tempfile::tempdir().unwrap();
1009
        let root = dir.path();
1010
        let url = format!("dir://{}", root.join("bucket").display());
1011
        let e = Repo::clone(&url, &root.join("nothing"))
1012
            .err()
1013
            .expect("refused")
1014
            .to_string();
1015
        assert!(e.contains("empty and unbound"), "{e}");
1016
        let mut a = Repo::init(&root.join("a")).unwrap();
1017
        add_remote(&mut a, "bucket", &url);
1018
        write(&root.join("a"), "f.txt", b"a\n");
1019
        a.commit("first", &[]).unwrap();
1020
        assert!(a.push("bucket", false).unwrap()[0].ok);
1021
        let store = LocalStore::open(root.join("bucket")).unwrap();
1022
        let key = keys(&store, "bundles/").remove(0);
1023
        let mut bytes = store.get(&key).unwrap();
1024
        let mid = bytes.len() / 2;
1025
        bytes[mid] ^= 0xff;
1026
        store.delete(&key).unwrap();
1027
        store.put_new(&key, &bytes).unwrap();
1028
        let e = Repo::clone(&url, &root.join("b"))
1029
            .err()
1030
            .expect("refused")
1031
            .to_string();
1032
        assert!(
1033
            e.contains(&key) || e.contains("bundle") || e.contains("hash"),
1034
            "{e}"
1035
        );
1036
        assert!(!root.join("b/f.txt").exists());
1037
    }
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 {
1053
        fn new(inner: std::sync::Arc<MemoryStore>) -> Box<Shared> {
1054
            Box::new(Shared {
1055
                inner,
1056
                fail_uploads: Default::default(),
1057
                vanish: Mutex::new(None),
1058
            })
1059
        }
1060
    }
1061
1062
    impl Store for Shared {
1063
        fn describe(&self) -> String {
1064
            "shared".into()
1065
        }
1066
        fn has(&self, key: &str) -> crate::strata::Result<bool> {
1067
            self.inner.has(key)
1068
        }
1069
        fn get(&self, key: &str) -> crate::strata::Result<Vec<u8>> {
1070
            let mut vanish = self.vanish.lock().unwrap();
1071
            if vanish.as_deref() == Some(key) {
1072
                *vanish = None;
1073
                return Err(StoreError::NotFound(key.to_string()));
1074
            }
1075
            self.inner.get(key)
1076
        }
1077
        fn get_range(&self, key: &str, range: Range<u64>) -> crate::strata::Result<Vec<u8>> {
1078
            self.inner.get_range(key, range)
1079
        }
1080
        fn put_new(&self, key: &str, bytes: &[u8]) -> crate::strata::Result<()> {
1081
            if key.starts_with("bundles/")
1082
                && self.fail_uploads.load(std::sync::atomic::Ordering::SeqCst)
1083
            {
1084
                return Err(StoreError::Backend("shared: the disk is full".into()));
1085
            }
1086
            self.inner.put_new(key, bytes)
1087
        }
1088
        fn list(&self, prefix: &str) -> crate::strata::Result<Vec<Entry>> {
1089
            self.inner.list(prefix)
1090
        }
1091
        fn delete(&self, key: &str) -> crate::strata::Result<()> {
1092
            self.inner.delete(key)
1093
        }
1094
        fn read_ref(&self, key: &str) -> crate::strata::Result<Option<(Vec<u8>, Version)>> {
1095
            self.inner.read_ref(key)
1096
        }
1097
        fn write_ref(
1098
            &self,
1099
            key: &str,
1100
            bytes: &[u8],
1101
            expected: Option<&Version>,
1102
        ) -> crate::strata::Result<Version> {
1103
            if key == "refs" && expected.is_some() {
1104
                if let Some(hook) = HOOK.with(|h| h.borrow_mut().take()) {
1105
                    hook();
1106
                }
1107
            }
1108
            self.inner.write_ref(key, bytes, expected)
1109
        }
1110
    }
1111
1112
    fn repo_with_commit(root: &Path, name: &str, file: &str, text: &[u8], msg: &str) -> Repo {
1113
        let repo = Repo::init(&root.join(name)).unwrap();
1114
        write(&root.join(name), file, text);
1115
        repo.commit(msg, &[]).unwrap();
1116
        repo
1117
    }
1118
1119
    fn remote_for(repo: &Repo, store: Box<dyn Store>, root: &Path, name: &str) -> BucketRemote {
1120
        let options = RemoteOptions {
1121
            repo_id: Some(repo.id().unwrap()),
1122
            cache_dir: Some(root.join(format!("cache-{name}"))),
1123
            ..Default::default()
1124
        };
1125
        BucketRemote::open(store, &options).unwrap()
1126
    }
1127
1128
    fn push_bookmark(
1129
        repo: &Repo,
1130
        remote: &BucketRemote,
1131
        bookmark: &str,
1132
        old: Option<ObjectRef>,
1133
    ) -> Vec<RefResult> {
1134
        let heads = repo.heads_of(&repo.view().unwrap(), bookmark).unwrap();
1135
        let state = repo
1136
            .store
1137
            .put_state(&StateObject::new(heads.clone()))
1138
            .unwrap();
1139
        let empty = repo.store.put_state(&StateObject::new(vec![])).unwrap();
1140
        // Only what the remote lacks, as push_bookmark sends.
1141
        let have: Vec<ChangeHash> = remote
1142
            .refs()
1143
            .unwrap()
1144
            .refs
1145
            .iter()
1146
            .flat_map(|r| r.heads.clone())
1147
            .collect();
1148
        let changes = repo.negotiate(&have, &heads).unwrap();
1149
        let mut bundle = repo.bundle_for(&changes).unwrap();
1150
        bundle
1151
            .objects
1152
            .push((ObjectType::State, StateObject::new(heads).encode()));
1153
        let update = RefUpdate {
1154
            name: bookmark.into(),
1155
            old_state: old.unwrap_or(empty),
1156
            new_state: state,
1157
            force: false,
1158
            attestation: None,
1159
        };
1160
        remote.push(bundle, vec![update], None, vec![]).unwrap()
1161
    }
1162
1163
    #[test]
1164
    fn two_pushes_to_different_bookmarks_both_land() {
1165
        hermetic();
1166
        let dir = tempfile::tempdir().unwrap();
1167
        let root = dir.path();
1168
        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.
1171
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
1172
        let ra = remote_for(&a, Shared::new(inner.clone()), root, "a");
1173
        let r = push_bookmark(&a, &ra, "main", None);
1174
        assert!(r[0].ok, "{r:?}");
1175
        let b = Repo::init_with_id(&root.join("b"), a.id().unwrap()).unwrap();
1176
        let rb = remote_for(&b, Shared::new(inner.clone()), root, "b");
1177
        rb.refs().unwrap();
1178
        write(&root.join("b"), "g.txt", b"g\n");
1179
        b.commit("on b", &[]).unwrap();
1180
        let (_, op) = b.head_op().unwrap();
1181
        let mut view = op.view.clone();
1182
        let feature_heads = b.heads_of(&view, "main").unwrap();
1183
        view.bookmarks.insert(
1184
            "feature".into(),
1185
            b.store.put_state(&StateObject::new(feature_heads)).unwrap(),
1186
        );
1187
        b.record_op(&[b.head_op().unwrap().0], "bookmark create feature", view)
1188
            .unwrap();
1189
        // A's second push to main; B's push of feature lands in between.
1190
        write(&root.join("a"), "f.txt", b"aa\n");
1191
        a.commit("second", &[]).unwrap();
1192
        let a_old = ra
1193
            .refs()
1194
            .unwrap()
1195
            .refs
1196
            .iter()
1197
            .find(|r| r.name == "main")
1198
            .unwrap()
1199
            .state;
1200
        let b_remote = std::sync::Arc::new(std::sync::Mutex::new(Some(rb)));
1201
        let b_handle = b_remote.clone();
1202
        HOOK.with(|h| {
1203
            *h.borrow_mut() = Some(Box::new(move || {
1204
                let rb = b_handle.lock().unwrap().take().unwrap();
1205
                let r = push_bookmark(&b, &rb, "feature", None);
1206
                assert!(r[0].ok, "{r:?}");
1207
            }))
1208
        });
1209
        let r = push_bookmark(&a, &ra, "main", Some(a_old));
1210
        assert!(r[0].ok, "a's push is applied over b's: {r:?}");
1211
        let refs = ra.refs().unwrap();
1212
        let names: Vec<&str> = refs.refs.iter().map(|r| r.name.as_str()).collect();
1213
        assert_eq!(names, ["feature", "main"], "both landed");
1214
        assert_eq!(inner.list("bundles/").unwrap().len(), 3);
1215
        let _ = b_remote;
1216
    }
1217
1218
    #[test]
1219
    fn an_interrupted_push_leaves_the_store_usable() {
1220
        hermetic();
1221
        let dir = tempfile::tempdir().unwrap();
1222
        let root = dir.path();
1223
        let inner = std::sync::Arc::new(MemoryStore::new());
1224
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
1225
        let shared = Shared::new(inner.clone());
1226
        let fail = &shared.fail_uploads as *const std::sync::atomic::AtomicBool;
1227
        let ra = remote_for(&a, shared, root, "a");
1228
        assert!(push_bookmark(&a, &ra, "main", None)[0].ok);
1229
        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.
1232
        write(&root.join("a"), "f.txt", b"aa\n");
1233
        let h2 = a.commit("second", &[]).unwrap();
1234
        unsafe { (*fail).store(true, std::sync::atomic::Ordering::SeqCst) };
1235
        let e = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1236
            push_bookmark(&a, &ra, "main", Some(old))
1237
        }));
1238
        assert!(e.is_err(), "the push reports the failed upload");
1239
        assert_eq!(inner.list("bundles/").unwrap().len(), 1);
1240
        let refs = ra.refs().unwrap();
1241
        assert_eq!(refs.refs[0].state, old, "the record did not move");
1242
        unsafe { (*fail).store(false, std::sync::atomic::Ordering::SeqCst) };
1243
        let r = push_bookmark(&a, &ra, "main", Some(old));
1244
        assert!(r[0].ok, "{r:?}");
1245
        assert_eq!(ra.refs().unwrap().refs[0].heads, vec![h2]);
1246
    }
1247
1248
    #[test]
1249
    fn bundles_out_of_order_or_gone_are_handled() {
1250
        hermetic();
1251
        let dir = tempfile::tempdir().unwrap();
1252
        let root = dir.path();
1253
        let inner = std::sync::Arc::new(MemoryStore::new());
1254
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
1255
        let ra = remote_for(&a, Shared::new(inner.clone()), root, "a");
1256
        assert!(push_bookmark(&a, &ra, "main", None)[0].ok);
1257
        let old = ra.refs().unwrap().refs[0].state;
1258
        write(&root.join("a"), "f.txt", b"aa\n");
1259
        let h2 = a.commit("second", &[]).unwrap();
1260
        assert!(push_bookmark(&a, &ra, "main", Some(old))[0].ok);
1261
        // The second bundle (which needs the first) renamed to sort first.
1262
        let keys: Vec<String> = inner
1263
            .list("bundles/")
1264
            .unwrap()
1265
            .into_iter()
1266
            .map(|e| e.key)
1267
            .collect();
1268
        assert_eq!(keys.len(), 2);
1269
        let bytes = inner.get(&keys[1]).unwrap();
1270
        inner.delete(&keys[1]).unwrap();
1271
        inner
1272
            .put_new("bundles/00000000-first.bundle", &bytes)
1273
            .unwrap();
1274
        let shared = Shared::new(inner.clone());
1275
        let c = Repo::init_with_id(&root.join("c"), a.id().unwrap()).unwrap();
1276
        let rc = remote_for(&c, shared, root, "c");
1277
        assert_eq!(
1278
            rc.refs().unwrap().refs[0].heads,
1279
            vec![h2],
1280
            "the dependent bundle waited for its dependency"
1281
        );
1282
        // A bundle gone between the listing and the read is skipped.
1283
        let shared = Shared::new(inner.clone());
1284
        *shared.vanish.lock().unwrap() = Some(keys[0].clone());
1285
        let d = Repo::init_with_id(&root.join("d"), a.id().unwrap()).unwrap();
1286
        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.
1289
        let e = rd.refs().expect_err("missing changes").to_string();
1290
        assert!(e.contains("cannot be imported"), "{e}");
1291
    }
1292
1293
    #[test]
1294
    fn gc_merges_bundles_and_prunes_after_the_grace_period() {
1295
        hermetic();
1296
        let dir = tempfile::tempdir().unwrap();
1297
        let root = dir.path();
1298
        let url = format!("dir://{}", root.join("bucket").display());
1299
        let store = LocalStore::open(root.join("bucket")).unwrap();
1300
        let mut a = Repo::init(&root.join("a")).unwrap();
1301
        add_remote(&mut a, "bucket", &url);
1302
        for i in 1..=3 {
1303
            write(&root.join("a"), "f.txt", format!("{i}\n").as_bytes());
1304
            a.commit(&format!("change {i}"), &[]).unwrap();
1305
            assert!(a.push("bucket", false).unwrap()[0].ok);
1306
        }
1307
        a.tag_create("v3", &a.bookmark_heads().unwrap(), "three")
1308
            .unwrap();
1309
        assert!(a.push("bucket", false).unwrap().iter().all(|r| r.ok));
1310
        assert_eq!(
1311
            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.
1316
        let r = a.gc_remote("bucket", false).unwrap();
1317
        assert_eq!(
1318
            (r.merged, r.changes, r.deleted, r.deferred),
1319
            (4, 3, 0, 4),
1320
            "{r:?}"
1321
        );
1322
        assert!(r.bundle_bytes > 0);
1323
        assert_eq!(keys(&store, "bundles/").len(), 5);
1324
        let b = Repo::clone(&url, &root.join("b")).unwrap();
1325
        assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"3\n");
1326
        assert_eq!(b.tags().unwrap().len(), 1);
1327
        // Past the grace period they go, and a clone needs only the one.
1328
        let r = a.gc_remote("bucket", true).unwrap();
1329
        assert_eq!((r.merged, r.deleted, r.deferred), (0, 4, 0), "{r:?}");
1330
        assert_eq!(keys(&store, "bundles/").len(), 1);
1331
        let c = Repo::clone(&url, &root.join("c")).unwrap();
1332
        assert_eq!(c.bookmark_heads().unwrap(), a.bookmark_heads().unwrap());
1333
        // The earlier clone, which saw the old bundles, carries on.
1334
        write(&root.join("b"), "f.txt", b"4\n");
1335
        b.commit("four", &[]).unwrap();
1336
        assert!(b.push("origin", false).unwrap()[0].ok);
1337
        assert!(a.pull("bucket").unwrap().changed);
1338
        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.
1340
        let r = a.gc_remote("bucket", true).unwrap();
1341
        assert_eq!((r.merged, r.deleted), (2, 2));
1342
        assert!(a.gc_remote("nope", true).is_err());
1343
    }
1344
1345
    #[test]
1346
    fn markers_parse_and_refuse_newer_formats() {
1347
        let id = [7u8; 16];
1348
        assert_eq!(parse_marker(&marker(id), "s").unwrap(), id);
1349
        let e = parse_marker(
1350
            b"format = 9\nrepo = \"07070707070707070707070707070707\"\n",
1351
            "s",
1352
        )
1353
        .unwrap_err()
1354
        .to_string();
1355
        assert!(e.contains("format 9"), "{e}");
1356
        assert!(parse_marker(b"hello", "s").is_err());
1357
    }
1358
}