Strata: a push lands over what landed meanwhile; oxy gc --remote d8d9367878a6

            let before = pending.len();

                let bytes = match self.store.get(&key) {

                    Ok(b) => b,

                    // Listed, then gone: a gc merged it into a newer

                    // bundle, which this listing holds too.

                    Err(StoreError::NotFound(_)) => continue,

                    Err(e) => return Err(e.into()),

                };

            if deferred.len() == before {

        // The bundle is checked against the mirror and uploaded once; the

        // record is then moved from the version read. If it moved

        // meanwhile (another push landed), the mirror is put back, brought

        // level again, and the updates applied over what landed: two

        // pushes to different bookmarks both get through, and one to the

        // same bookmark is told it is stale, as a hub would say.

        let mut uploaded = false;

        for attempt in 0..PUSH_ATTEMPTS {

            let (before_ref, before) = self.with_mirror(|m| m.head_op())?;

            let results = self.with_mirror(|m| {

                m.receive_push(

                    bundle.clone(),

                    updates.clone(),

                    policy.clone(),

                    tags.clone(),

                )

            })?;

            if !results.iter().any(|r| r.ok) {

                return Ok(results);

            }

            if !bundle.is_empty() && !uploaded {

                self.upload(&bundle)?;

                uploaded = true;

            }

            let record = self.record(id)?;

            match self

                .store

                .write_ref(REFS_KEY, &record, synced.version.as_ref())

            {

                Ok(_) => return Ok(results),

                Err(StoreError::Conflict(_)) => {

                    self.with_mirror(|m| {

                        let (now, _) = m.head_op()?;

                        if now != before_ref {

                            m.record_op(

                                &[now],

                                "strata push undone: the remote moved",

                                before.view.clone(),

                            )?;

                        }

                        Ok(())

                    })?;

                    synced = self.sync()?;

                    if attempt + 1 == PUSH_ATTEMPTS {

                        return Ok(results

                            .into_iter()

                            .map(|r| RefResult {

                                name: r.name,

                                ok: false,

                                reason: "the remote kept moving while pushing (other pushes landed first); pull, then push again".into(),

                            })

                            .collect());

                    }

                }

                Err(e) => return Err(e.into()),

            }

        unreachable!("the loop returns")

    }

}

/// How often a push is applied over what landed meanwhile before giving up.

const PUSH_ATTEMPTS: usize = 4;

/// The grace period before `gc` removes a bundle it merged: a reader that

/// listed it before the merge can still fetch it.

pub const REMOTE_GC_GRACE: std::time::Duration = std::time::Duration::from_secs(7 * 86_400);

/// Key of the record naming the bundles a gc merged, with when.

const TRASH_KEY: &str = "gc";

/// The superseded bundles and when (milliseconds) each was superseded.

type Trash = Vec<(String, u64)>;

/// What a gc against a store did.

#[derive(Clone, Debug, Default, PartialEq, Eq)]

pub struct RemoteGc {

    /// Bundles folded into one new bundle.

    pub merged: usize,

    /// The merged bundle's size, bytes.

    pub bundle_bytes: usize,

    /// Changes in the merged bundle.

    pub changes: usize,

    /// Superseded bundles removed, past the grace period.

    pub deleted: usize,

    /// Superseded bundles kept for now by the grace period.

    pub deferred: usize,

}

fn now_ms() -> u64 {

    std::time::SystemTime::now()

        .duration_since(std::time::UNIX_EPOCH)

        .map(|d| d.as_millis() as u64)

        .unwrap_or(0)

}

impl BucketRemote {

    /// The bundles a gc superseded, with when: one `KEY\tMILLIS` per line.

    fn trash(&self) -> Result<(Trash, Option<Version>)> {

        match self.store.read_ref(TRASH_KEY)? {

            Some((bytes, v)) => {

                let text = String::from_utf8_lossy(&bytes);

                let list = text

                    .lines()

                    .filter_map(|l| {

                        let (k, t) = l.split_once('\t')?;

                        Some((k.to_string(), t.parse().ok()?))

                    })

                    .collect();

                Ok((list, Some(v)))

            }

            None => Ok((Vec::new(), None)),

    }

    fn write_trash(&self, list: &[(String, u64)], expected: Option<&Version>) -> Result<Version> {

        let text: String = list.iter().map(|(k, t)| format!("{k}\t{t}\n")).collect();

        Ok(self.store.write_ref(TRASH_KEY, text.as_bytes(), expected)?)

    }

    /// Fold every bundle into one holding what the record reaches, and

    /// remove the bundles that made a gc redundant once `grace` has

    /// passed since they were superseded (stores.md §6).

    pub fn gc(&self, grace: std::time::Duration) -> Result<RemoteGc> {

        let synced = self.sync()?;

        let Some(id) = synced.id else {

            return fail(format!("{} holds no repository yet", self.what()));

        };

        let mut report = RemoteGc::default();

        let existing: Vec<String> = self

            .list(BUNDLES_PREFIX)?

            .into_iter()

            .map(|e| e.key)

            .collect();

        let (mut trash, mut trash_version) = self.trash()?;

        let live: Vec<&String> = existing

            .iter()

            .filter(|k| !trash.iter().any(|(t, _)| t == *k))

            .collect();

        if live.len() > 1 {

            // Everything the record reaches, in dependency order, with

            // the state objects the bookmarks and tags name.

            let (bundle, changes) = self.with_mirror(|m| {

                let resp = m.remote_refs()?;

                let mut heads: Vec<ChangeHash> = Vec::new();

                for r in &resp.refs {

                    heads.extend(r.heads.iter().copied());

                }

                for t in &resp.tags {

                    heads.extend(t.heads.iter().copied());

                }

                heads.sort();

                heads.dedup();

                let changes = m.negotiate(&[], &heads)?;

                let mut bundle = m.bundle_for(&changes)?;

                for r in &resp.refs {

                    bundle.objects.push((

                        ObjectType::State,

                        StateObject::new(r.heads.clone()).encode(),

                    ));

                }

                for t in &resp.tags {

                    bundle.objects.push((

                        ObjectType::State,

                        StateObject::new(t.heads.clone()).encode(),

                    ));

                }

                Ok((bundle, changes.len()))

            })?;

            report.changes = changes;

            report.bundle_bytes = wire::encode_bundle(&bundle).len();

            let merged_key = self.upload(&bundle)?;

            let now = now_ms();

            for k in live {

                if *k != merged_key {

                    trash.push((k.clone(), now));

                    report.merged += 1;

                }

            }

            trash_version = Some(self.write_trash(&trash, trash_version.as_ref())?);

        }

        // What the grace period has released.

        let now = now_ms();

        let mut kept = Vec::new();

        for (k, when) in trash {

            if now.saturating_sub(when) >= grace.as_millis() as u64 {

                self.store.delete(&k)?;

                report.deleted += 1;

            } else {

                kept.push((k, when));

                report.deferred += 1;

        }

        if report.deleted > 0 {

            self.write_trash(&kept, trash_version.as_ref())?;

        let _ = id;

        Ok(report)

impl Repo {

    /// `oxy gc --remote NAME`: a gc against a remote kept on a store.

    pub fn gc_remote(&self, name: &str, now: bool) -> Result<RemoteGc> {

        let Some(url) = self.config.remotes.get(name).cloned() else {

            return fail(format!("no remote {name:?}"));

        };

        let options = self.remote_options(name);

        let Some(store) = crate::protocol::open_store(&url, &options) else {

            return fail(format!(

                "gc --remote works on a remote kept on a store (s3://, gs://, az://, dir://); {name} is {url}"

            ));

        };

        let remote = BucketRemote::open(store?, &options)?;

        let grace = if now {

            std::time::Duration::ZERO

        } else {

            REMOTE_GC_GRACE

        };

        remote.gc(grace)

    }

}

    fn a_record_moved_during_a_push_is_applied_over_what_landed() {

        assert!(r[0].ok, "applied over what landed meanwhile: {r:?}");

            vec![h2],

            "the store and the mirror both moved"

    }

    /// A store shared by two remotes in one test, with a hook run once at

    /// the first conditional write: what another push landing between a

    /// read of the record and its compare-and-swap looks like.

    struct Shared {

        inner: std::sync::Arc<MemoryStore>,

        fail_uploads: std::sync::atomic::AtomicBool,

        vanish: Mutex<Option<String>>,

    }

    thread_local! {

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

    }

    impl Shared {

        fn new(inner: std::sync::Arc<MemoryStore>) -> Box<Shared> {

            Box::new(Shared {

                inner,

                fail_uploads: Default::default(),

                vanish: Mutex::new(None),

            })

        }

    }

    impl Store for Shared {

        fn describe(&self) -> String {

            "shared".into()

        }

        fn has(&self, key: &str) -> crate::strata::Result<bool> {

            self.inner.has(key)

        }

        fn get(&self, key: &str) -> crate::strata::Result<Vec<u8>> {

            let mut vanish = self.vanish.lock().unwrap();

            if vanish.as_deref() == Some(key) {

                *vanish = None;

                return Err(StoreError::NotFound(key.to_string()));

            }

            self.inner.get(key)

        }

        fn get_range(&self, key: &str, range: Range<u64>) -> crate::strata::Result<Vec<u8>> {

            self.inner.get_range(key, range)

        }

        fn put_new(&self, key: &str, bytes: &[u8]) -> crate::strata::Result<()> {

            if key.starts_with("bundles/")

                && self.fail_uploads.load(std::sync::atomic::Ordering::SeqCst)

            {

                return Err(StoreError::Backend("shared: the disk is full".into()));

            }

            self.inner.put_new(key, bytes)

        }

        fn list(&self, prefix: &str) -> crate::strata::Result<Vec<Entry>> {

            self.inner.list(prefix)

        }

        fn delete(&self, key: &str) -> crate::strata::Result<()> {

            self.inner.delete(key)

        }

        fn read_ref(&self, key: &str) -> crate::strata::Result<Option<(Vec<u8>, Version)>> {

            self.inner.read_ref(key)

        }

        fn write_ref(

            &self,

            key: &str,

            bytes: &[u8],

            expected: Option<&Version>,

        ) -> crate::strata::Result<Version> {

            if key == "refs" && expected.is_some() {

                if let Some(hook) = HOOK.with(|h| h.borrow_mut().take()) {

                    hook();

                }

            }

            self.inner.write_ref(key, bytes, expected)

        }

    }

    fn repo_with_commit(root: &Path, name: &str, file: &str, text: &[u8], msg: &str) -> Repo {

        let repo = Repo::init(&root.join(name)).unwrap();

        write(&root.join(name), file, text);

        repo.commit(msg, &[]).unwrap();

        repo

    }

    fn remote_for(repo: &Repo, store: Box<dyn Store>, root: &Path, name: &str) -> BucketRemote {

        let options = RemoteOptions {

            repo_id: Some(repo.id().unwrap()),

            cache_dir: Some(root.join(format!("cache-{name}"))),

            ..Default::default()

        };

        BucketRemote::open(store, &options).unwrap()

    }

    fn push_bookmark(

        repo: &Repo,

        remote: &BucketRemote,

        bookmark: &str,

        old: Option<ObjectRef>,

    ) -> Vec<RefResult> {

        let heads = repo.heads_of(&repo.view().unwrap(), bookmark).unwrap();

        let state = repo

            .store

            .put_state(&StateObject::new(heads.clone()))

            .unwrap();

        let empty = repo.store.put_state(&StateObject::new(vec![])).unwrap();

        // Only what the remote lacks, as push_bookmark sends.

        let have: Vec<ChangeHash> = remote

            .refs()

            .unwrap()

            .refs

            .iter()

            .flat_map(|r| r.heads.clone())

            .collect();

        let changes = repo.negotiate(&have, &heads).unwrap();

        let mut bundle = repo.bundle_for(&changes).unwrap();

        bundle

            .objects

            .push((ObjectType::State, StateObject::new(heads).encode()));

        let update = RefUpdate {

            name: bookmark.into(),

            old_state: old.unwrap_or(empty),

            new_state: state,

            force: false,

            attestation: None,

        };

        remote.push(bundle, vec![update], None, vec![]).unwrap()

    }

    #[test]

    fn two_pushes_to_different_bookmarks_both_land() {

        hermetic();

        let dir = tempfile::tempdir().unwrap();

        let root = dir.path();

        let inner = std::sync::Arc::new(MemoryStore::new());

        // A pushes main; B (a clone of the same repository) pushes a

        // second bookmark between A's read and A's compare-and-swap.

        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");

        let ra = remote_for(&a, Shared::new(inner.clone()), root, "a");

        let r = push_bookmark(&a, &ra, "main", None);

        assert!(r[0].ok, "{r:?}");

        let b = Repo::init_with_id(&root.join("b"), a.id().unwrap()).unwrap();

        let rb = remote_for(&b, Shared::new(inner.clone()), root, "b");

        rb.refs().unwrap();

        write(&root.join("b"), "g.txt", b"g\n");

        b.commit("on b", &[]).unwrap();

        let (_, op) = b.head_op().unwrap();

        let mut view = op.view.clone();

        let feature_heads = b.heads_of(&view, "main").unwrap();

        view.bookmarks.insert(

            "feature".into(),

            b.store.put_state(&StateObject::new(feature_heads)).unwrap(),

        );

        b.record_op(&[b.head_op().unwrap().0], "bookmark create feature", view)

            .unwrap();

        // A's second push to main; B's push of feature lands in between.

        write(&root.join("a"), "f.txt", b"aa\n");

        a.commit("second", &[]).unwrap();

        let a_old = ra

            .refs()

            .unwrap()

            .refs

            .iter()

            .find(|r| r.name == "main")

            .unwrap()

            .state;

        let b_remote = std::sync::Arc::new(std::sync::Mutex::new(Some(rb)));

        let b_handle = b_remote.clone();

        HOOK.with(|h| {

            *h.borrow_mut() = Some(Box::new(move || {

                let rb = b_handle.lock().unwrap().take().unwrap();

                let r = push_bookmark(&b, &rb, "feature", None);

                assert!(r[0].ok, "{r:?}");

            }))

        });

        let r = push_bookmark(&a, &ra, "main", Some(a_old));

        assert!(r[0].ok, "a's push is applied over b's: {r:?}");

        let refs = ra.refs().unwrap();

        let names: Vec<&str> = refs.refs.iter().map(|r| r.name.as_str()).collect();

        assert_eq!(names, ["feature", "main"], "both landed");

        assert_eq!(inner.list("bundles/").unwrap().len(), 3);

        let _ = b_remote;

    }

    #[test]

    fn an_interrupted_push_leaves_the_store_usable() {

        hermetic();

        let dir = tempfile::tempdir().unwrap();

        let root = dir.path();

        let inner = std::sync::Arc::new(MemoryStore::new());

        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");

        let shared = Shared::new(inner.clone());

        let fail = &shared.fail_uploads as *const std::sync::atomic::AtomicBool;

        let ra = remote_for(&a, shared, root, "a");

        assert!(push_bookmark(&a, &ra, "main", None)[0].ok);

        let old = ra.refs().unwrap().refs[0].state;

        // The upload fails: nothing in the store changes, the mirror is

        // honest, and the next push gets through.

        write(&root.join("a"), "f.txt", b"aa\n");

        let h2 = a.commit("second", &[]).unwrap();

        unsafe { (*fail).store(true, std::sync::atomic::Ordering::SeqCst) };

        let e = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {

            push_bookmark(&a, &ra, "main", Some(old))

        }));

        assert!(e.is_err(), "the push reports the failed upload");

        assert_eq!(inner.list("bundles/").unwrap().len(), 1);

        let refs = ra.refs().unwrap();

        assert_eq!(refs.refs[0].state, old, "the record did not move");

        unsafe { (*fail).store(false, std::sync::atomic::Ordering::SeqCst) };

        let r = push_bookmark(&a, &ra, "main", Some(old));

        assert!(r[0].ok, "{r:?}");

        assert_eq!(ra.refs().unwrap().refs[0].heads, vec![h2]);

    }

    #[test]

    fn bundles_out_of_order_or_gone_are_handled() {

        hermetic();

        let dir = tempfile::tempdir().unwrap();

        let root = dir.path();

        let inner = std::sync::Arc::new(MemoryStore::new());

        let a = repo_with_commit(root, "a", "f.txt", b"a\n", "first");

        let ra = remote_for(&a, Shared::new(inner.clone()), root, "a");

        assert!(push_bookmark(&a, &ra, "main", None)[0].ok);

        let old = ra.refs().unwrap().refs[0].state;

        write(&root.join("a"), "f.txt", b"aa\n");

        let h2 = a.commit("second", &[]).unwrap();

        assert!(push_bookmark(&a, &ra, "main", Some(old))[0].ok);

        // The second bundle (which needs the first) renamed to sort first.

        let keys: Vec<String> = inner

            .list("bundles/")

            .unwrap()

            .into_iter()

            .map(|e| e.key)

            .collect();

        assert_eq!(keys.len(), 2);

        let bytes = inner.get(&keys[1]).unwrap();

        inner.delete(&keys[1]).unwrap();

        inner

            .put_new("bundles/00000000-first.bundle", &bytes)

            .unwrap();

        let shared = Shared::new(inner.clone());

        let c = Repo::init_with_id(&root.join("c"), a.id().unwrap()).unwrap();

        let rc = remote_for(&c, shared, root, "c");

        assert_eq!(

            rc.refs().unwrap().refs[0].heads,

            vec![h2],

            "the dependent bundle waited for its dependency"

        );

        // A bundle gone between the listing and the read is skipped.

        let shared = Shared::new(inner.clone());

        *shared.vanish.lock().unwrap() = Some(keys[0].clone());

        let d = Repo::init_with_id(&root.join("d"), a.id().unwrap()).unwrap();

        let rd = remote_for(&d, shared, root, "d");

        // Without the first bundle the second cannot be imported: the sync

        // says so rather than failing silently.

        let e = rd.refs().expect_err("missing changes").to_string();

        assert!(e.contains("cannot be imported"), "{e}");

    }

    #[test]

    fn gc_merges_bundles_and_prunes_after_the_grace_period() {

        hermetic();

        let dir = tempfile::tempdir().unwrap();

        let root = dir.path();

        let url = format!("dir://{}", root.join("bucket").display());

        let store = LocalStore::open(root.join("bucket")).unwrap();

        let mut a = Repo::init(&root.join("a")).unwrap();

        add_remote(&mut a, "bucket", &url);

        for i in 1..=3 {

            write(&root.join("a"), "f.txt", format!("{i}\n").as_bytes());

            a.commit(&format!("change {i}"), &[]).unwrap();

            assert!(a.push("bucket", false).unwrap()[0].ok);

        }

        a.tag_create("v3", &a.bookmark_heads().unwrap(), "three")

            .unwrap();

        assert!(a.push("bucket", false).unwrap().iter().all(|r| r.ok));

        assert_eq!(

            keys(&store, "bundles/").len(),

            4,

            "three pushes and the tag's"

        );

        // A gc folds the four into one; the old ones wait out the grace.

        let r = a.gc_remote("bucket", false).unwrap();

        assert_eq!(

            (r.merged, r.changes, r.deleted, r.deferred),

            (4, 3, 0, 4),

            "{r:?}"

        );

        assert!(r.bundle_bytes > 0);

        assert_eq!(keys(&store, "bundles/").len(), 5);

        let b = Repo::clone(&url, &root.join("b")).unwrap();

        assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"3\n");

        assert_eq!(b.tags().unwrap().len(), 1);

        // Past the grace period they go, and a clone needs only the one.

        let r = a.gc_remote("bucket", true).unwrap();

        assert_eq!((r.merged, r.deleted, r.deferred), (0, 4, 0), "{r:?}");

        assert_eq!(keys(&store, "bundles/").len(), 1);

        let c = Repo::clone(&url, &root.join("c")).unwrap();

        assert_eq!(c.bookmark_heads().unwrap(), a.bookmark_heads().unwrap());

        // The earlier clone, which saw the old bundles, carries on.

        write(&root.join("b"), "f.txt", b"4\n");

        b.commit("four", &[]).unwrap();

        assert!(b.push("origin", false).unwrap()[0].ok);

        assert!(a.pull("bucket").unwrap().changed);

        assert_eq!(fs::read(root.join("a/f.txt")).unwrap(), b"4\n");

        // Nothing more to merge: a second gc is a no-op.

        let r = a.gc_remote("bucket", true).unwrap();

        assert_eq!((r.merged, r.deleted), (2, 2));

        assert!(a.gc_remote("nope", true).is_err());

  store says, brought level, and the push's updates are applied over

  what landed (up to four times). Two pushes to different bookmarks

  both get through; one to the same bookmark is told it is stale, as a

  hub says: pull, then push again.

- **`oxy gc --remote NAME`** folds every bundle into one holding what

  the record reaches, and records the superseded ones in a `gc`

  record with the time. They are removed only once seven days have

  passed (`--now` ignores that), so a reader that listed them before

  the fold can still fetch them; a reader that finds one gone skips it,

  since the merged bundle holds the same changes.

        /// Instead: fold the bundles of a remote kept on a store into one, and remove the ones a gc superseded 7 days ago (or now, with --now)

        #[arg(long, value_name = "NAME")]

        remote: Option<String>,

        Cmd::Gc { now, remote } => match remote {

            Some(name) => workspace::gc_remote(&name, now),

            None => workspace::gc(now),

        },

/// `oxy gc --remote NAME`: fold a store remote's bundles into one.

pub fn gc_remote(name: &str, now: bool) -> Result<i32> {

    let repo = open()?;

    let sp = ui::spinner(&format!("Folding {name}'s bundles…"));

    let r = repo.gc_remote(name, now)?;

    sp.clear();

    if r.merged > 0 {

        println!(

            "{} {} into one of {} bytes ({})",

            ui::done("Folded"),

            plural(r.merged, "bundle"),

            r.bundle_bytes,

            plural(r.changes, "change")

        );

    } else {

        println!(

            "{} {}",

            ui::note(),

            ui::dim("nothing to fold: one bundle holds everything")

        );

    }

    println!(

        "  {} {}",

        ui::note(),

        ui::dim(format!(

            "removed {}{}",

            plural(r.deleted, "superseded bundle"),

            if r.deferred > 0 {

                format!(

                    "; {} kept for readers that may still fetch them (use --now)",

                    plural(r.deferred, "bundle")

                )

            } else {

                String::new()

            }

        ))

    );

    Ok(0)

}

    // Two pushes made two bundles; a gc folds them into one, and a clone

    // after it is whole.

    let out = ok(&a, &home, &["gc", "--remote", "usb", "--now"]);

    assert!(

        out.contains("Folded 2 bundles") && out.contains("removed 2 superseded bundles"),

        "{out}"

    );

    assert_eq!(

        std::fs::read_dir(bucket.join("bundles")).unwrap().count(),

        1

    );

    let d = tmp.path().join("d");

    ok(tmp.path(), &home, &["clone", &url, d.to_str().unwrap()]);

    assert_eq!(

        std::fs::read_to_string(d.join("f.txt")).unwrap(),

        "hello\nagain\n"

    );

    let (code, _, err) = oxy(&a, &home, &["gc", "--remote", "nope"]);

    assert_ne!(code, 0);

    assert!(err.contains("no remote"), "{err}");