OXYGENHUB
◆ v0.5.0 oxygen / crates / oxy-core / src / repo / strata.rs
51.6 KiB · Rust · last changed Readers never wait: one head per handle, retry on a miss, retire before deleting 18 days ago source blame history raw
9f9e72b9 Andy McCall ↶ 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| {
767ac6cd Andy McCall 236
            // Another process may have brought the mirror level since this
237
            // handle read its head.
238
            mirror.refresh()?;
9f9e72b9 Andy McCall ↶ 239
            // The policy first: bundles are then imported under it.
240
            if let Some(sp) = &resp.policy {
241
                let have = mirror.current_policy()?.map(|(_, p)| p.policy.serial);
242
                if have.is_none_or(|s| s < sp.policy.serial) {
243
                    mirror
244
                        .adopt_policy_chain(&resp.history, sp, "strata sync")
245
                        .map_err(|e| RepoError::Msg(format!("policy from {}: {e}", self.what())))?;
246
                }
247
            }
248
            self.import_new_bundles(mirror)?;
249
            self.set_view(mirror, &resp)?;
250
            Ok(())
251
        })?;
252
        Ok(Synced {
253
            version,
254
            id: Some(id),
255
        })
256
    }
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.
262
    fn import_new_bundles(&self, mirror: &Repo) -> Result<()> {
263
        let seen = self.seen();
264
        let mut pending: Vec<String> = self
265
            .store
266
            .list(BUNDLES_PREFIX)?
267
            .into_iter()
268
            .map(|e| e.key)
269
            .filter(|k| !seen.contains(k))
270
            .collect();
271
        let strict = mirror.current_policy()?.is_some();
272
        while !pending.is_empty() {
273
            let mut deferred = Vec::new();
274
            let mut last_error = None;
d8d93678 Andy McCall ↶ 275
            let before = pending.len();
9f9e72b9 Andy McCall ↶ 276
            for key in pending.drain(..) {
d8d93678 Andy McCall ↶ 277
                let bytes = match self.store.get(&key) {
278
                    Ok(b) => b,
279
                    // Listed, then gone: a gc merged it into a newer
280
                    // bundle, which this listing holds too.
281
                    Err(StoreError::NotFound(_)) => continue,
282
                    Err(e) => return Err(e.into()),
283
                };
9f9e72b9 Andy McCall ↶ 284
                let bundle = wire::decode_bundle(&bytes)
285
                    .map_err(|e| RepoError::Msg(format!("{}: {key}: {e}", self.what())))?;
286
                match mirror.import_bundle(&bundle, strict) {
287
                    Ok(_) => self.mark_seen(&key)?,
288
                    Err(RepoError::Msg(m)) if m.contains("neither stored nor in the bundle") => {
289
                        last_error = Some(m);
290
                        deferred.push(key);
291
                    }
292
                    Err(e) => return Err(RepoError::Msg(format!("{}: {key}: {e}", self.what()))),
293
                }
294
            }
d8d93678 Andy McCall ↶ 295
            if deferred.len() == before {
9f9e72b9 Andy McCall ↶ 296
                // No progress: something those bundles need was never
297
                // pushed, or a push is still in progress.
298
                return fail(format!(
299
                    "{}: {} cannot be imported ({}); a push may be in progress, so try again, else the store is missing changes",
300
                    self.what(),
301
                    deferred.join(", "),
302
                    last_error.unwrap_or_default()
303
                ));
304
            }
305
            pending = deferred;
306
        }
307
        Ok(())
308
    }
309
310
    /// Make the mirror's view say what the record says.
311
    fn set_view(&self, mirror: &Repo, resp: &RefsResponse) -> Result<()> {
312
        let (op_ref, op) = mirror.head_op()?;
313
        let mut view = op.view.clone();
314
        view.bookmarks.clear();
315
        for r in &resp.refs {
316
            for h in &r.heads {
317
                if !mirror.store.has(h) {
318
                    return fail(format!(
319
                        "{}: bookmark {:?} names change {}, which no bundle there holds; a push may be in progress, so try again",
320
                        self.what(),
321
                        r.name,
322
                        h.short(12)
323
                    ));
324
                }
325
            }
326
            let state = mirror.store.put_state(&StateObject::new(r.heads.clone()))?;
327
            if state != r.state {
328
                return fail(format!(
329
                    "{}: bookmark {:?} names a state that does not match its heads; the store is damaged",
330
                    self.what(),
331
                    r.name
332
                ));
333
            }
334
            if let Some(sa) = &r.attestation {
335
                let current = mirror.latest_ref_attest(&r.name)?.map(|(_, a)| a);
336
                if current.as_ref() != Some(sa) {
337
                    let a = mirror.store.put_attest(&sa.attest)?;
338
                    for sig in &sa.sigs {
339
                        mirror.store.put_signature(&a, sig)?;
340
                    }
341
                    mirror.set_ref_attest(&r.name, &a)?;
342
                }
343
            }
344
            view.bookmarks.insert(r.name.clone(), state);
345
        }
346
        let policy = mirror.current_policy()?.map(|(_, sp)| sp.policy);
347
        for t in &resp.tags {
348
            mirror.store.put_state(&StateObject::new(t.heads.clone()))?;
349
            mirror
350
                .adopt_tag(&mut view, t, policy.as_ref())
351
                .map_err(|e| RepoError::Msg(format!("{}: {e}", self.what())))?;
352
        }
353
        if view != op.view {
354
            mirror.record_op(&[op_ref], "strata sync", view)?;
355
        }
356
        Ok(())
357
    }
358
359
    /// The record for the mirror's current references.
360
    fn record(&self, id: [u8; 16]) -> Result<Vec<u8>> {
361
        self.with_mirror(|mirror| {
362
            let mut resp = mirror.remote_refs()?;
363
            // A mirror has no checkout; the store's clone picks a bookmark
364
            // as any other remote's does.
365
            resp.default_bookmark = None;
366
            Ok(wire::encode_refs(id, &resp))
367
        })
368
    }
369
370
    /// Store a bundle under the next sequence number; returns its key.
371
    fn upload(&self, bundle: &Bundle) -> Result<String> {
372
        let bytes = wire::encode_bundle(bundle);
373
        let hash = blake3::hash(&bytes).to_hex();
374
        let seq = self.store.list(BUNDLES_PREFIX)?.len() + 1;
375
        let key = format!("{BUNDLES_PREFIX}{seq:08}-{}.bundle", &hash[..16]);
376
        match self.store.put_new(&key, &bytes) {
377
            Ok(()) | Err(StoreError::Exists(_)) => {}
378
            Err(e) => return Err(e.into()),
379
        }
380
        self.mark_seen(&key)?;
381
        Ok(key)
382
    }
383
}
384
385
impl Remote for BucketRemote {
386
    fn repo_id(&self) -> Result<[u8; 16]> {
387
        Ok(self.stored_id()?.unwrap_or(UNBOUND_ID))
388
    }
389
390
    fn refs(&self) -> Result<RefsResponse> {
391
        let synced = self.sync()?;
392
        if synced.id.is_none() {
393
            return Ok(RefsResponse::default());
394
        }
395
        self.with_mirror(|mirror| {
396
            let mut resp = mirror.remote_refs()?;
397
            resp.default_bookmark = None;
398
            Ok(resp)
399
        })
400
    }
401
402
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
403
        self.with_mirror(|m| m.negotiate(have, want))
404
    }
405
406
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
407
        self.with_mirror(|m| m.bundle_raw_for(changes, max_blob))
408
    }
409
410
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
411
        self.with_mirror(|m| m.objects_for(refs))
412
    }
413
414
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
415
        self.with_mirror(|m| m.checkpoint_bundle_for(heads))
416
    }
417
418
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
419
        self.with_mirror(|m| m.shallow_changes(heads, depth))
420
    }
421
422
    fn push(
423
        &self,
424
        bundle: Bundle,
425
        updates: Vec<RefUpdate>,
426
        policy: Option<SignedPolicy>,
427
        tags: Vec<TagRef>,
428
    ) -> Result<Vec<RefResult>> {
429
        let mut synced = self.sync()?;
430
        let id = match synced.id {
431
            Some(id) => id,
432
            None => {
433
                // The first push binds the store to this repository. Two
434
                // first pushes race on the marker; the loser reads the
435
                // winner's and goes on only if it is the same repository.
436
                let Some(id) = self.claimed_id else {
437
                    return fail(format!(
438
                        "{} is empty, and this client did not say which repository it is",
439
                        self.what()
440
                    ));
441
                };
442
                match self.store.put_new(MARKER_KEY, &marker(id)) {
443
                    Ok(()) => {}
444
                    Err(StoreError::Exists(_)) => {}
445
                    Err(e) => return Err(e.into()),
446
                }
447
                synced = self.sync()?;
448
                match synced.id {
449
                    Some(stored) if stored == id => id,
450
                    _ => {
451
                        return fail(format!(
452
                            "{} was just bound to another repository by a concurrent push",
453
                            self.what()
454
                        ))
455
                    }
456
                }
457
            }
458
        };
459
        if let Some(claimed) = self.claimed_id {
460
            if claimed != id {
461
                return fail(format!(
462
                    "{} holds a different repository (id {}) than this one ({}): a store holds one repository; use another prefix or bucket",
463
                    self.what(),
464
                    hex::encode(id),
465
                    hex::encode(claimed)
466
                ));
467
            }
468
        }
d8d93678 Andy McCall ↶ 469
        // The bundle is checked against the mirror and uploaded once; the
470
        // record is then moved from the version read. If it moved
471
        // meanwhile (another push landed), the mirror is put back, brought
472
        // level again, and the updates applied over what landed: two
473
        // pushes to different bookmarks both get through, and one to the
474
        // same bookmark is told it is stale, as a hub would say.
475
        let mut uploaded = false;
476
        for attempt in 0..PUSH_ATTEMPTS {
477
            let (before_ref, before) = self.with_mirror(|m| m.head_op())?;
478
            let results = self.with_mirror(|m| {
479
                m.receive_push(
480
                    bundle.clone(),
481
                    updates.clone(),
482
                    policy.clone(),
483
                    tags.clone(),
484
                )
485
            })?;
486
            if !results.iter().any(|r| r.ok) {
487
                return Ok(results);
488
            }
489
            if !bundle.is_empty() && !uploaded {
490
                self.upload(&bundle)?;
491
                uploaded = true;
492
            }
493
            let record = self.record(id)?;
494
            match self
495
                .store
496
                .write_ref(REFS_KEY, &record, synced.version.as_ref())
497
            {
498
                Ok(_) => return Ok(results),
499
                Err(StoreError::Conflict(_)) => {
500
                    self.with_mirror(|m| {
501
                        let (now, _) = m.head_op()?;
502
                        if now != before_ref {
503
                            m.record_op(
504
                                &[now],
505
                                "strata push undone: the remote moved",
506
                                before.view.clone(),
507
                            )?;
508
                        }
509
                        Ok(())
510
                    })?;
511
                    synced = self.sync()?;
512
                    if attempt + 1 == PUSH_ATTEMPTS {
513
                        return Ok(results
514
                            .into_iter()
515
                            .map(|r| RefResult {
516
                                name: r.name,
517
                                ok: false,
518
                                reason: "the remote kept moving while pushing (other pushes landed first); pull, then push again".into(),
519
                            })
520
                            .collect());
521
                    }
522
                }
523
                Err(e) => return Err(e.into()),
524
            }
9f9e72b9 Andy McCall ↶ 525
        }
d8d93678 Andy McCall ↶ 526
        unreachable!("the loop returns")
527
    }
528
}
529
530
/// How often a push is applied over what landed meanwhile before giving up.
531
const PUSH_ATTEMPTS: usize = 4;
532
533
/// The grace period before `gc` removes a bundle it merged: a reader that
534
/// listed it before the merge can still fetch it.
535
pub const REMOTE_GC_GRACE: std::time::Duration = std::time::Duration::from_secs(7 * 86_400);
536
537
/// Key of the record naming the bundles a gc merged, with when.
538
const TRASH_KEY: &str = "gc";
539
540
/// The superseded bundles and when (milliseconds) each was superseded.
541
type Trash = Vec<(String, u64)>;
542
543
/// What a gc against a store did.
544
#[derive(Clone, Debug, Default, PartialEq, Eq)]
545
pub struct RemoteGc {
546
    /// Bundles folded into one new bundle.
547
    pub merged: usize,
548
    /// The merged bundle's size, bytes.
549
    pub bundle_bytes: usize,
550
    /// Changes in the merged bundle.
551
    pub changes: usize,
552
    /// Superseded bundles removed, past the grace period.
553
    pub deleted: usize,
554
    /// Superseded bundles kept for now by the grace period.
555
    pub deferred: usize,
556
}
557
558
fn now_ms() -> u64 {
559
    std::time::SystemTime::now()
560
        .duration_since(std::time::UNIX_EPOCH)
561
        .map(|d| d.as_millis() as u64)
562
        .unwrap_or(0)
563
}
564
565
impl BucketRemote {
566
    /// The bundles a gc superseded, with when: one `KEY\tMILLIS` per line.
567
    fn trash(&self) -> Result<(Trash, Option<Version>)> {
568
        match self.store.read_ref(TRASH_KEY)? {
569
            Some((bytes, v)) => {
570
                let text = String::from_utf8_lossy(&bytes);
571
                let list = text
572
                    .lines()
573
                    .filter_map(|l| {
574
                        let (k, t) = l.split_once('\t')?;
575
                        Some((k.to_string(), t.parse().ok()?))
576
                    })
577
                    .collect();
578
                Ok((list, Some(v)))
579
            }
580
            None => Ok((Vec::new(), None)),
9f9e72b9 Andy McCall ↶ 581
        }
d8d93678 Andy McCall ↶ 582
    }
583
584
    fn write_trash(&self, list: &[(String, u64)], expected: Option<&Version>) -> Result<Version> {
585
        let text: String = list.iter().map(|(k, t)| format!("{k}\t{t}\n")).collect();
586
        Ok(self.store.write_ref(TRASH_KEY, text.as_bytes(), expected)?)
587
    }
588
589
    /// Fold every bundle into one holding what the record reaches, and
590
    /// remove the bundles that made a gc redundant once `grace` has
591
    /// passed since they were superseded (stores.md §6).
592
    pub fn gc(&self, grace: std::time::Duration) -> Result<RemoteGc> {
593
        let synced = self.sync()?;
594
        let Some(id) = synced.id else {
595
            return fail(format!("{} holds no repository yet", self.what()));
596
        };
597
        let mut report = RemoteGc::default();
598
        let existing: Vec<String> = self
9f9e72b9 Andy McCall ↶ 599
            .store
d8d93678 Andy McCall ↶ 600
            .list(BUNDLES_PREFIX)?
601
            .into_iter()
602
            .map(|e| e.key)
603
            .collect();
604
        let (mut trash, mut trash_version) = self.trash()?;
605
        let live: Vec<&String> = existing
606
            .iter()
607
            .filter(|k| !trash.iter().any(|(t, _)| t == *k))
608
            .collect();
609
        if live.len() > 1 {
610
            // Everything the record reaches, in dependency order, with
611
            // the state objects the bookmarks and tags name.
612
            let (bundle, changes) = self.with_mirror(|m| {
613
                let resp = m.remote_refs()?;
614
                let mut heads: Vec<ChangeHash> = Vec::new();
615
                for r in &resp.refs {
616
                    heads.extend(r.heads.iter().copied());
617
                }
618
                for t in &resp.tags {
619
                    heads.extend(t.heads.iter().copied());
620
                }
621
                heads.sort();
622
                heads.dedup();
623
                let changes = m.negotiate(&[], &heads)?;
624
                let mut bundle = m.bundle_for(&changes)?;
625
                for r in &resp.refs {
626
                    bundle.objects.push((
627
                        ObjectType::State,
628
                        StateObject::new(r.heads.clone()).encode(),
629
                    ));
630
                }
631
                for t in &resp.tags {
632
                    bundle.objects.push((
633
                        ObjectType::State,
634
                        StateObject::new(t.heads.clone()).encode(),
635
                    ));
636
                }
637
                Ok((bundle, changes.len()))
638
            })?;
639
            report.changes = changes;
640
            report.bundle_bytes = wire::encode_bundle(&bundle).len();
641
            let merged_key = self.upload(&bundle)?;
642
            let now = now_ms();
643
            for k in live {
644
                if *k != merged_key {
645
                    trash.push((k.clone(), now));
646
                    report.merged += 1;
647
                }
648
            }
649
            trash_version = Some(self.write_trash(&trash, trash_version.as_ref())?);
650
        }
651
        // What the grace period has released.
652
        let now = now_ms();
653
        let mut kept = Vec::new();
654
        for (k, when) in trash {
655
            if now.saturating_sub(when) >= grace.as_millis() as u64 {
656
                self.store.delete(&k)?;
657
                report.deleted += 1;
658
            } else {
659
                kept.push((k, when));
660
                report.deferred += 1;
9f9e72b9 Andy McCall ↶ 661
            }
d8d93678 Andy McCall ↶ 662
        }
663
        if report.deleted > 0 {
664
            self.write_trash(&kept, trash_version.as_ref())?;
9f9e72b9 Andy McCall ↶ 665
        }
d8d93678 Andy McCall ↶ 666
        let _ = id;
667
        Ok(report)
9f9e72b9 Andy McCall ↶ 668
    }
669
}
670
d8d93678 Andy McCall ↶ 671
impl Repo {
672
    /// `oxy gc --remote NAME`: a gc against a remote kept on a store.
673
    pub fn gc_remote(&self, name: &str, now: bool) -> Result<RemoteGc> {
674
        let Some(url) = self.config.remotes.get(name).cloned() else {
675
            return fail(format!("no remote {name:?}"));
676
        };
677
        let options = self.remote_options(name);
678
        let Some(store) = crate::protocol::open_store(&url, &options) else {
679
            return fail(format!(
680
                "gc --remote works on a remote kept on a store (s3://, gs://, az://, dir://); {name} is {url}"
681
            ));
682
        };
683
        let remote = BucketRemote::open(store?, &options)?;
684
        let grace = if now {
685
            std::time::Duration::ZERO
686
        } else {
687
            REMOTE_GC_GRACE
688
        };
689
        remote.gc(grace)
690
    }
691
}
692
9f9e72b9 Andy McCall ↶ 693
/// Open a remote on `store` for the local repository described by `options`.
694
pub fn open_store_remote(
695
    store: Box<dyn Store>,
696
    options: &RemoteOptions,
697
) -> Result<Box<dyn Remote>> {
698
    Ok(Box::new(BucketRemote::open(store, options)?))
699
}
700
701
/// Whether `path` holds a Strata store rather than a repository.
702
pub fn is_store_dir(path: &Path) -> bool {
703
    path.join(MARKER_KEY).is_file() || path.join(REFS_KEY).is_file()
704
}
705
706
#[cfg(test)]
707
mod tests {
708
    use super::*;
709
    use crate::repo::IdentityScope;
710
    use crate::strata::{Entry, LocalStore, MemoryStore};
711
    use std::ops::Range;
712
    use std::sync::Mutex;
713
714
    fn hermetic() {
715
        static ONCE: std::sync::Once = std::sync::Once::new();
716
        static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();
717
        ONCE.call_once(|| {
718
            let d = tempfile::tempdir().unwrap();
719
            std::env::set_var("OXY_HOME", d.path());
720
            DIR.set(d).ok();
721
        });
722
    }
723
724
    fn write(root: &Path, rel: &str, content: &[u8]) {
725
        let p = root.join(rel);
726
        fs::create_dir_all(p.parent().unwrap()).unwrap();
727
        fs::write(p, content).unwrap();
728
    }
729
730
    fn add_remote(repo: &mut Repo, name: &str, url: &str) {
731
        repo.config.set(&format!("remote.{name}.url"), url).unwrap();
732
        repo.save_config().unwrap();
733
    }
734
735
    fn keys(store: &dyn Store, prefix: &str) -> Vec<String> {
736
        store
737
            .list(prefix)
738
            .unwrap()
739
            .into_iter()
740
            .map(|e| e.key)
741
            .collect()
742
    }
743
744
    #[test]
745
    fn push_clone_and_pull_through_a_directory_store() {
746
        hermetic();
747
        let dir = tempfile::tempdir().unwrap();
748
        let root = dir.path();
749
        let url = format!("dir://{}", root.join("bucket").display());
750
        let store = LocalStore::open(root.join("bucket")).unwrap();
751
752
        // A signs, commits, tags and pushes: the store gets the marker, one
753
        // bundle, and a record.
754
        let mut a = Repo::init(&root.join("a"))
755
            .unwrap()
756
            .with_home(root.join("home-a"));
757
        a.create_identity("Ada", "[email protected]", IdentityScope::Repo)
758
            .unwrap();
759
        add_remote(&mut a, "bucket", &url);
760
        write(&root.join("a"), "f.txt", b"a\n");
761
        let h1 = a.commit("first", &[]).unwrap();
762
        a.tag_create("v1", &[h1], "one").unwrap();
763
        let results = a.push("bucket", false).unwrap();
764
        assert!(results.iter().all(|r| r.ok), "{results:?}");
765
        assert_eq!(
766
            keys(&store, ""),
767
            [
768
                "bundles/00000001-".to_string() + &keys(&store, "bundles/")[0][17..],
769
                "oxygen.toml".into(),
770
                "refs".into()
771
            ]
772
        );
773
        assert_eq!(
774
            parse_marker(&store.get("oxygen.toml").unwrap(), "t").unwrap(),
775
            a.id().unwrap()
776
        );
777
        let again = a.push("bucket", false).unwrap();
778
        assert!(
779
            again[0].ok && again[0].reason.contains("up to date"),
780
            "{again:?}"
781
        );
782
        assert_eq!(
783
            keys(&store, "bundles/").len(),
784
            1,
785
            "a no-op push uploads nothing"
786
        );
787
788
        // B clones from the store: same id, same file, the tag, and the
789
        // seal on the change.
790
        let b = Repo::clone(&url, &root.join("b")).unwrap();
791
        assert_eq!(b.id().unwrap(), a.id().unwrap());
792
        assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"a\n");
793
        assert_eq!(b.bookmark_heads().unwrap(), vec![h1]);
794
        assert_eq!(
795
            b.tags()
796
                .unwrap()
797
                .iter()
798
                .map(|t| t.name.clone())
799
                .collect::<Vec<_>>(),
800
            ["v1"]
801
        );
802
        assert!(matches!(
803
            b.authenticate_change(&h1).unwrap(),
804
            crate::repo::SignatureStatus::Valid { sealed: true, .. }
805
        ));
806
        assert!(
807
            root.join("b/.oxy/remotes/origin/strata/mirror/.oxy")
808
                .is_dir(),
809
            "the mirror lives with the remote's state"
810
        );
811
        assert!(
812
            !std::env::temp_dir()
813
                .read_dir()
814
                .unwrap()
815
                .flatten()
816
                .any(|e| e.file_name().to_string_lossy().starts_with("oxy-strata-")),
817
            "a clone's temporary mirror is removed"
818
        );
819
820
        // B commits and pushes; A pulls it. A then commits and pushes.
821
        write(&root.join("b"), "f.txt", b"a\nb\n");
822
        let h2 = b.commit("second", &[]).unwrap();
823
        assert!(b.push("origin", false).unwrap()[0].ok);
824
        assert_eq!(keys(&store, "bundles/").len(), 2);
825
        let pr = a.pull("bucket").unwrap();
826
        assert!(pr.changed && pr.checked_out);
827
        assert_eq!(fs::read(root.join("a/f.txt")).unwrap(), b"a\nb\n");
828
        write(&root.join("a"), "g.txt", b"g\n");
829
        let h3 = a.commit("third", &[]).unwrap();
830
        assert!(a.push("bucket", false).unwrap()[0].ok);
831
        let pr = b.pull("origin").unwrap();
832
        assert!(pr.changed);
833
        // g.txt is new, so its change commutes with B's: two heads.
834
        let mut want = vec![h2, h3];
835
        want.sort();
836
        assert_eq!(b.bookmark_heads().unwrap(), want);
837
        assert_eq!(fs::read(root.join("b/g.txt")).unwrap(), b"g\n");
838
839
        // A third clone gets everything from the three bundles.
840
        let c = Repo::clone(&url, &root.join("c")).unwrap();
841
        assert_eq!(c.bookmark_heads().unwrap(), want);
842
        assert_eq!(fs::read(root.join("c/f.txt")).unwrap(), b"a\nb\n");
843
    }
844
845
    #[test]
846
    fn a_push_behind_the_remote_is_refused_before_anything_is_uploaded() {
847
        hermetic();
848
        let dir = tempfile::tempdir().unwrap();
849
        let root = dir.path();
850
        let url = format!("dir://{}", root.join("bucket").display());
851
        let store = LocalStore::open(root.join("bucket")).unwrap();
852
        let mut a = Repo::init(&root.join("a")).unwrap();
853
        add_remote(&mut a, "bucket", &url);
854
        write(&root.join("a"), "f.txt", b"a\n");
855
        a.commit("first", &[]).unwrap();
856
        assert!(a.push("bucket", false).unwrap()[0].ok);
857
        let b = Repo::clone(&url, &root.join("b")).unwrap();
858
        write(&root.join("b"), "f.txt", b"b\n");
859
        b.commit("from b", &[]).unwrap();
860
        assert!(b.push("origin", false).unwrap()[0].ok);
861
        write(&root.join("a"), "g.txt", b"g\n");
862
        a.commit("from a", &[]).unwrap();
863
        let e = a.push("bucket", false).unwrap_err().to_string();
864
        assert!(e.contains("pull first"), "{e}");
865
        assert_eq!(keys(&store, "bundles/").len(), 2);
866
        assert!(a.pull("bucket").unwrap().changed);
867
        assert!(a.push("bucket", false).unwrap()[0].ok);
868
        assert_eq!(keys(&store, "bundles/").len(), 3);
869
    }
870
871
    /// A store whose record moves under a writer the first time it is
872
    /// written to: what a concurrent push looks like between the read and
873
    /// the compare-and-swap.
874
    struct Racing {
875
        inner: MemoryStore,
876
        raced: Mutex<bool>,
877
    }
878
879
    impl Store for Racing {
880
        fn describe(&self) -> String {
881
            "racing".into()
882
        }
883
        fn has(&self, key: &str) -> crate::strata::Result<bool> {
884
            self.inner.has(key)
885
        }
886
        fn get(&self, key: &str) -> crate::strata::Result<Vec<u8>> {
887
            self.inner.get(key)
888
        }
889
        fn get_range(&self, key: &str, range: Range<u64>) -> crate::strata::Result<Vec<u8>> {
890
            self.inner.get_range(key, range)
891
        }
892
        fn put_new(&self, key: &str, bytes: &[u8]) -> crate::strata::Result<()> {
893
            self.inner.put_new(key, bytes)
894
        }
895
        fn list(&self, prefix: &str) -> crate::strata::Result<Vec<Entry>> {
896
            self.inner.list(prefix)
897
        }
898
        fn delete(&self, key: &str) -> crate::strata::Result<()> {
899
            self.inner.delete(key)
900
        }
901
        fn read_ref(&self, key: &str) -> crate::strata::Result<Option<(Vec<u8>, Version)>> {
902
            self.inner.read_ref(key)
903
        }
904
        fn write_ref(
905
            &self,
906
            key: &str,
907
            bytes: &[u8],
908
            expected: Option<&Version>,
909
        ) -> crate::strata::Result<Version> {
910
            let mut raced = self.raced.lock().unwrap();
911
            if !*raced && expected.is_some() {
912
                *raced = true;
913
                // Someone else lands a record: the same references with a
914
                // different detail, so the version moves.
915
                let (current, v) = self.inner.read_ref(key)?.unwrap();
916
                let (id, mut resp) =
917
                    wire::decode_refs(&current).map_err(|e| StoreError::Backend(e.to_string()))?;
918
                resp.default_bookmark = Some("raced".into());
919
                self.inner
920
                    .write_ref(key, &wire::encode_refs(id, &resp), Some(&v))?;
921
            }
922
            self.inner.write_ref(key, bytes, expected)
923
        }
924
    }
925
926
    #[test]
d8d93678 Andy McCall ↶ 927
    fn a_record_moved_during_a_push_is_applied_over_what_landed() {
9f9e72b9 Andy McCall ↶ 928
        hermetic();
929
        let dir = tempfile::tempdir().unwrap();
930
        let root = dir.path();
931
        let a = Repo::init(&root.join("a")).unwrap();
932
        write(&root.join("a"), "f.txt", b"a\n");
933
        let h1 = a.commit("first", &[]).unwrap();
934
        let store = Box::new(Racing {
935
            inner: MemoryStore::new(),
936
            raced: Mutex::new(false),
937
        });
938
        let options = RemoteOptions {
939
            repo_id: Some(a.id().unwrap()),
940
            cache_dir: Some(root.join("cache")),
941
            ..Default::default()
942
        };
943
        let remote = BucketRemote::open(store, &options).unwrap();
944
        // The first push creates the record (expected None: no race yet).
945
        let state = a.store.put_state(&StateObject::new(vec![h1])).unwrap();
946
        let empty = a.store.put_state(&StateObject::new(vec![])).unwrap();
947
        let mut bundle = a.bundle_for(&[h1]).unwrap();
948
        bundle
949
            .objects
950
            .push((ObjectType::State, StateObject::new(vec![h1]).encode()));
951
        let update = |old, new| RefUpdate {
952
            name: "main".into(),
953
            old_state: old,
954
            new_state: new,
955
            force: false,
956
            attestation: None,
957
        };
958
        let r = remote
959
            .push(bundle, vec![update(empty, state)], None, vec![])
960
            .unwrap();
961
        assert!(r[0].ok, "{r:?}");
962
        // The second push meets the race: refused, nothing moved.
963
        write(&root.join("a"), "f.txt", b"ab\n");
964
        let h2 = a.commit("second", &[]).unwrap();
965
        let state2 = a.store.put_state(&StateObject::new(vec![h2])).unwrap();
966
        let mut bundle = a.bundle_for(&[h2]).unwrap();
967
        bundle
968
            .objects
969
            .push((ObjectType::State, StateObject::new(vec![h2]).encode()));
970
        let r = remote
971
            .push(bundle, vec![update(state, state2)], None, vec![])
972
            .unwrap();
d8d93678 Andy McCall ↶ 973
        assert!(r[0].ok, "applied over what landed meanwhile: {r:?}");
9f9e72b9 Andy McCall ↶ 974
        let refs = remote.refs().unwrap();
975
        assert_eq!(
976
            refs.refs[0].heads,
d8d93678 Andy McCall ↶ 977
            vec![h2],
978
            "the store and the mirror both moved"
9f9e72b9 Andy McCall ↶ 979
        );
980
    }
981
982
    #[test]
983
    fn a_store_holds_one_repository() {
984
        hermetic();
985
        let dir = tempfile::tempdir().unwrap();
986
        let root = dir.path();
987
        let url = format!("dir://{}", root.join("bucket").display());
988
        let mut a = Repo::init(&root.join("a")).unwrap();
989
        add_remote(&mut a, "bucket", &url);
990
        write(&root.join("a"), "f.txt", b"a\n");
991
        a.commit("first", &[]).unwrap();
992
        assert!(a.push("bucket", false).unwrap()[0].ok);
993
        let mut other = Repo::init(&root.join("other")).unwrap();
994
        add_remote(&mut other, "bucket", &url);
995
        write(&root.join("other"), "f.txt", b"o\n");
996
        other.commit("mine", &[]).unwrap();
997
        let e = other.push("bucket", false).unwrap_err().to_string();
998
        assert!(
999
            e.contains("different repository") || e.contains("another repository"),
1000
            "{e}"
1001
        );
1002
        assert_eq!(
1003
            keys(&LocalStore::open(root.join("bucket")).unwrap(), "bundles/").len(),
1004
            1
1005
        );
1006
    }
1007
1008
    #[test]
1009
    fn a_damaged_bundle_is_caught_and_an_empty_store_cannot_be_cloned() {
1010
        hermetic();
1011
        let dir = tempfile::tempdir().unwrap();
1012
        let root = dir.path();
1013
        let url = format!("dir://{}", root.join("bucket").display());
1014
        let e = Repo::clone(&url, &root.join("nothing"))
1015
            .err()
1016
            .expect("refused")
1017
            .to_string();
1018
        assert!(e.contains("empty and unbound"), "{e}");
1019
        let mut a = Repo::init(&root.join("a")).unwrap();
1020
        add_remote(&mut a, "bucket", &url);
1021
        write(&root.join("a"), "f.txt", b"a\n");
1022
        a.commit("first", &[]).unwrap();
1023
        assert!(a.push("bucket", false).unwrap()[0].ok);
1024
        let store = LocalStore::open(root.join("bucket")).unwrap();
1025
        let key = keys(&store, "bundles/").remove(0);
1026
        let mut bytes = store.get(&key).unwrap();
1027
        let mid = bytes.len() / 2;
1028
        bytes[mid] ^= 0xff;
1029
        store.delete(&key).unwrap();
1030
        store.put_new(&key, &bytes).unwrap();
1031
        let e = Repo::clone(&url, &root.join("b"))
1032
            .err()
1033
            .expect("refused")
1034
            .to_string();
1035
        assert!(
1036
            e.contains(&key) || e.contains("bundle") || e.contains("hash"),
1037
            "{e}"
1038
        );
1039
        assert!(!root.join("b/f.txt").exists());
d8d93678 Andy McCall ↶ 1040
    }
1041
1042
    /// A store shared by two remotes in one test, with a hook run once at
1043
    /// the first conditional write: what another push landing between a
1044
    /// read of the record and its compare-and-swap looks like.
1045
    struct Shared {
1046
        inner: std::sync::Arc<MemoryStore>,
1047
        fail_uploads: std::sync::atomic::AtomicBool,
1048
        vanish: Mutex<Option<String>>,
1049
    }
1050
1051
    thread_local! {
1052
        static HOOK: std::cell::RefCell<Option<Box<dyn FnOnce()>>> = const { std::cell::RefCell::new(None) };
1053
    }
1054
1055
    impl Shared {
1056
        fn new(inner: std::sync::Arc<MemoryStore>) -> Box<Shared> {
1057
            Box::new(Shared {
1058
                inner,
1059
                fail_uploads: Default::default(),
1060
                vanish: Mutex::new(None),
1061
            })
1062
        }
1063
    }
1064
1065
    impl Store for Shared {
1066
        fn describe(&self) -> String {
1067
            "shared".into()
1068
        }
1069
        fn has(&self, key: &str) -> crate::strata::Result<bool> {
1070
            self.inner.has(key)
1071
        }
1072
        fn get(&self, key: &str) -> crate::strata::Result<Vec<u8>> {
1073
            let mut vanish = self.vanish.lock().unwrap();
1074
            if vanish.as_deref() == Some(key) {
1075
                *vanish = None;
1076
                return Err(StoreError::NotFound(key.to_string()));
1077
            }
1078
            self.inner.get(key)
1079
        }
1080
        fn get_range(&self, key: &str, range: Range<u64>) -> crate::strata::Result<Vec<u8>> {
1081
            self.inner.get_range(key, range)
1082
        }
1083
        fn put_new(&self, key: &str, bytes: &[u8]) -> crate::strata::Result<()> {
1084
            if key.starts_with("bundles/")
1085
                && self.fail_uploads.load(std::sync::atomic::Ordering::SeqCst)
1086
            {
1087
                return Err(StoreError::Backend("shared: the disk is full".into()));
1088
            }
1089
            self.inner.put_new(key, bytes)
1090
        }
1091
        fn list(&self, prefix: &str) -> crate::strata::Result<Vec<Entry>> {
1092
            self.inner.list(prefix)
1093
        }
1094
        fn delete(&self, key: &str) -> crate::strata::Result<()> {
1095
            self.inner.delete(key)
1096
        }
1097
        fn read_ref(&self, key: &str) -> crate::strata::Result<Option<(Vec<u8>, Version)>> {
1098
            self.inner.read_ref(key)
1099
        }
1100
        fn write_ref(
1101
            &self,
1102
            key: &str,
1103
            bytes: &[u8],
1104
            expected: Option<&Version>,
1105
        ) -> crate::strata::Result<Version> {
1106
            if key == "refs" && expected.is_some() {
1107
                if let Some(hook) = HOOK.with(|h| h.borrow_mut().take()) {
1108
                    hook();
1109
                }
1110
            }
1111
            self.inner.write_ref(key, bytes, expected)
1112
        }
1113
    }
1114
1115
    fn repo_with_commit(root: &Path, name: &str, file: &str, text: &[u8], msg: &str) -> Repo {
1116
        let repo = Repo::init(&root.join(name)).unwrap();
1117
        write(&root.join(name), file, text);
1118
        repo.commit(msg, &[]).unwrap();
1119
        repo
1120
    }
1121
1122
    fn remote_for(repo: &Repo, store: Box<dyn Store>, root: &Path, name: &str) -> BucketRemote {
1123
        let options = RemoteOptions {
1124
            repo_id: Some(repo.id().unwrap()),
1125
            cache_dir: Some(root.join(format!("cache-{name}"))),
1126
            ..Default::default()
1127
        };
1128
        BucketRemote::open(store, &options).unwrap()
1129
    }
1130
1131
    fn push_bookmark(
1132
        repo: &Repo,
1133
        remote: &BucketRemote,
1134
        bookmark: &str,
1135
        old: Option<ObjectRef>,
1136
    ) -> Vec<RefResult> {
1137
        let heads = repo.heads_of(&repo.view().unwrap(), bookmark).unwrap();
1138
        let state = repo
1139
            .store
1140
            .put_state(&StateObject::new(heads.clone()))
1141
            .unwrap();
1142
        let empty = repo.store.put_state(&StateObject::new(vec![])).unwrap();
1143
        // Only what the remote lacks, as push_bookmark sends.
1144
        let have: Vec<ChangeHash> = remote
1145
            .refs()
1146
            .unwrap()
1147
            .refs
1148
            .iter()
1149
            .flat_map(|r| r.heads.clone())
1150
            .collect();
1151
        let changes = repo.negotiate(&have, &heads).unwrap();
1152
        let mut bundle = repo.bundle_for(&changes).unwrap();
1153
        bundle
1154
            .objects
1155
            .push((ObjectType::State, StateObject::new(heads).encode()));
1156
        let update = RefUpdate {
1157
            name: bookmark.into(),
1158
            old_state: old.unwrap_or(empty),
1159
            new_state: state,
1160
            force: false,
1161
            attestation: None,
1162
        };
1163
        remote.push(bundle, vec![update], None, vec![]).unwrap()
1164
    }
1165
1166
    #[test]
1167
    fn two_pushes_to_different_bookmarks_both_land() {
1168
        hermetic();
1169
        let dir = tempfile::tempdir().unwrap();
1170
        let root = dir.path();
1171
        let inner = std::sync::Arc::new(MemoryStore::new());
1172
        // A pushes main; B (a clone of the same repository) pushes a
1173
        // second bookmark between A's read and A's compare-and-swap.
1174
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
1175
        let ra = remote_for(&a, Shared::new(inner.clone()), root, "a");
1176
        let r = push_bookmark(&a, &ra, "main", None);
1177
        assert!(r[0].ok, "{r:?}");
1178
        let b = Repo::init_with_id(&root.join("b"), a.id().unwrap()).unwrap();
1179
        let rb = remote_for(&b, Shared::new(inner.clone()), root, "b");
1180
        rb.refs().unwrap();
1181
        write(&root.join("b"), "g.txt", b"g\n");
1182
        b.commit("on b", &[]).unwrap();
1183
        let (_, op) = b.head_op().unwrap();
1184
        let mut view = op.view.clone();
1185
        let feature_heads = b.heads_of(&view, "main").unwrap();
1186
        view.bookmarks.insert(
1187
            "feature".into(),
1188
            b.store.put_state(&StateObject::new(feature_heads)).unwrap(),
1189
        );
1190
        b.record_op(&[b.head_op().unwrap().0], "bookmark create feature", view)
1191
            .unwrap();
1192
        // A's second push to main; B's push of feature lands in between.
1193
        write(&root.join("a"), "f.txt", b"aa\n");
1194
        a.commit("second", &[]).unwrap();
1195
        let a_old = ra
1196
            .refs()
1197
            .unwrap()
1198
            .refs
1199
            .iter()
1200
            .find(|r| r.name == "main")
1201
            .unwrap()
1202
            .state;
1203
        let b_remote = std::sync::Arc::new(std::sync::Mutex::new(Some(rb)));
1204
        let b_handle = b_remote.clone();
1205
        HOOK.with(|h| {
1206
            *h.borrow_mut() = Some(Box::new(move || {
1207
                let rb = b_handle.lock().unwrap().take().unwrap();
1208
                let r = push_bookmark(&b, &rb, "feature", None);
1209
                assert!(r[0].ok, "{r:?}");
1210
            }))
1211
        });
1212
        let r = push_bookmark(&a, &ra, "main", Some(a_old));
1213
        assert!(r[0].ok, "a's push is applied over b's: {r:?}");
1214
        let refs = ra.refs().unwrap();
1215
        let names: Vec<&str> = refs.refs.iter().map(|r| r.name.as_str()).collect();
1216
        assert_eq!(names, ["feature", "main"], "both landed");
1217
        assert_eq!(inner.list("bundles/").unwrap().len(), 3);
1218
        let _ = b_remote;
1219
    }
1220
1221
    #[test]
1222
    fn an_interrupted_push_leaves_the_store_usable() {
1223
        hermetic();
1224
        let dir = tempfile::tempdir().unwrap();
1225
        let root = dir.path();
1226
        let inner = std::sync::Arc::new(MemoryStore::new());
1227
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
1228
        let shared = Shared::new(inner.clone());
1229
        let fail = &shared.fail_uploads as *const std::sync::atomic::AtomicBool;
1230
        let ra = remote_for(&a, shared, root, "a");
1231
        assert!(push_bookmark(&a, &ra, "main", None)[0].ok);
1232
        let old = ra.refs().unwrap().refs[0].state;
1233
        // The upload fails: nothing in the store changes, the mirror is
1234
        // honest, and the next push gets through.
1235
        write(&root.join("a"), "f.txt", b"aa\n");
1236
        let h2 = a.commit("second", &[]).unwrap();
1237
        unsafe { (*fail).store(true, std::sync::atomic::Ordering::SeqCst) };
1238
        let e = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1239
            push_bookmark(&a, &ra, "main", Some(old))
1240
        }));
1241
        assert!(e.is_err(), "the push reports the failed upload");
1242
        assert_eq!(inner.list("bundles/").unwrap().len(), 1);
1243
        let refs = ra.refs().unwrap();
1244
        assert_eq!(refs.refs[0].state, old, "the record did not move");
1245
        unsafe { (*fail).store(false, std::sync::atomic::Ordering::SeqCst) };
1246
        let r = push_bookmark(&a, &ra, "main", Some(old));
1247
        assert!(r[0].ok, "{r:?}");
1248
        assert_eq!(ra.refs().unwrap().refs[0].heads, vec![h2]);
1249
    }
1250
1251
    #[test]
1252
    fn bundles_out_of_order_or_gone_are_handled() {
1253
        hermetic();
1254
        let dir = tempfile::tempdir().unwrap();
1255
        let root = dir.path();
1256
        let inner = std::sync::Arc::new(MemoryStore::new());
1257
        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");
1258
        let ra = remote_for(&a, Shared::new(inner.clone()), root, "a");
1259
        assert!(push_bookmark(&a, &ra, "main", None)[0].ok);
1260
        let old = ra.refs().unwrap().refs[0].state;
1261
        write(&root.join("a"), "f.txt", b"aa\n");
1262
        let h2 = a.commit("second", &[]).unwrap();
1263
        assert!(push_bookmark(&a, &ra, "main", Some(old))[0].ok);
1264
        // The second bundle (which needs the first) renamed to sort first.
1265
        let keys: Vec<String> = inner
1266
            .list("bundles/")
1267
            .unwrap()
1268
            .into_iter()
1269
            .map(|e| e.key)
1270
            .collect();
1271
        assert_eq!(keys.len(), 2);
1272
        let bytes = inner.get(&keys[1]).unwrap();
1273
        inner.delete(&keys[1]).unwrap();
1274
        inner
1275
            .put_new("bundles/00000000-first.bundle", &bytes)
1276
            .unwrap();
1277
        let shared = Shared::new(inner.clone());
1278
        let c = Repo::init_with_id(&root.join("c"), a.id().unwrap()).unwrap();
1279
        let rc = remote_for(&c, shared, root, "c");
1280
        assert_eq!(
1281
            rc.refs().unwrap().refs[0].heads,
1282
            vec![h2],
1283
            "the dependent bundle waited for its dependency"
1284
        );
1285
        // A bundle gone between the listing and the read is skipped.
1286
        let shared = Shared::new(inner.clone());
1287
        *shared.vanish.lock().unwrap() = Some(keys[0].clone());
1288
        let d = Repo::init_with_id(&root.join("d"), a.id().unwrap()).unwrap();
1289
        let rd = remote_for(&d, shared, root, "d");
1290
        // Without the first bundle the second cannot be imported: the sync
1291
        // says so rather than failing silently.
1292
        let e = rd.refs().expect_err("missing changes").to_string();
1293
        assert!(e.contains("cannot be imported"), "{e}");
1294
    }
1295
1296
    #[test]
1297
    fn gc_merges_bundles_and_prunes_after_the_grace_period() {
1298
        hermetic();
1299
        let dir = tempfile::tempdir().unwrap();
1300
        let root = dir.path();
1301
        let url = format!("dir://{}", root.join("bucket").display());
1302
        let store = LocalStore::open(root.join("bucket")).unwrap();
1303
        let mut a = Repo::init(&root.join("a")).unwrap();
1304
        add_remote(&mut a, "bucket", &url);
1305
        for i in 1..=3 {
1306
            write(&root.join("a"), "f.txt", format!("{i}\n").as_bytes());
1307
            a.commit(&format!("change {i}"), &[]).unwrap();
1308
            assert!(a.push("bucket", false).unwrap()[0].ok);
1309
        }
1310
        a.tag_create("v3", &a.bookmark_heads().unwrap(), "three")
1311
            .unwrap();
1312
        assert!(a.push("bucket", false).unwrap().iter().all(|r| r.ok));
1313
        assert_eq!(
1314
            keys(&store, "bundles/").len(),
1315
            4,
1316
            "three pushes and the tag's"
1317
        );
1318
        // A gc folds the four into one; the old ones wait out the grace.
1319
        let r = a.gc_remote("bucket", false).unwrap();
1320
        assert_eq!(
1321
            (r.merged, r.changes, r.deleted, r.deferred),
1322
            (4, 3, 0, 4),
1323
            "{r:?}"
1324
        );
1325
        assert!(r.bundle_bytes > 0);
1326
        assert_eq!(keys(&store, "bundles/").len(), 5);
1327
        let b = Repo::clone(&url, &root.join("b")).unwrap();
1328
        assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"3\n");
1329
        assert_eq!(b.tags().unwrap().len(), 1);
1330
        // Past the grace period they go, and a clone needs only the one.
1331
        let r = a.gc_remote("bucket", true).unwrap();
1332
        assert_eq!((r.merged, r.deleted, r.deferred), (0, 4, 0), "{r:?}");
1333
        assert_eq!(keys(&store, "bundles/").len(), 1);
1334
        let c = Repo::clone(&url, &root.join("c")).unwrap();
1335
        assert_eq!(c.bookmark_heads().unwrap(), a.bookmark_heads().unwrap());
1336
        // The earlier clone, which saw the old bundles, carries on.
1337
        write(&root.join("b"), "f.txt", b"4\n");
1338
        b.commit("four", &[]).unwrap();
1339
        assert!(b.push("origin", false).unwrap()[0].ok);
1340
        assert!(a.pull("bucket").unwrap().changed);
1341
        assert_eq!(fs::read(root.join("a/f.txt")).unwrap(), b"4\n");
1342
        // Nothing more to merge: a second gc is a no-op.
1343
        let r = a.gc_remote("bucket", true).unwrap();
1344
        assert_eq!((r.merged, r.deleted), (2, 2));
1345
        assert!(a.gc_remote("nope", true).is_err());
9f9e72b9 Andy McCall ↶ 1346
    }
1347
1348
    #[test]
1349
    fn markers_parse_and_refuse_newer_formats() {
1350
        let id = [7u8; 16];
1351
        assert_eq!(parse_marker(&marker(id), "s").unwrap(), id);
1352
        let e = parse_marker(
1353
            b"format = 9\nrepo = \"07070707070707070707070707070707\"\n",
1354
            "s",
1355
        )
1356
        .unwrap_err()
1357
        .to_string();
1358
        assert!(e.contains("format 9"), "{e}");
1359
        assert!(parse_marker(b"hello", "s").is_err());
1360
    }
1361
}