Fetch every bookmark in one exchange, and none when nothing is new: a no-op pull of 160 bookmarks 169 s to 3 s, a clone 324 s to 26 s (closes #158) 2f0b4b0651ea

- **Fixed: fetch, pull and clone are far faster with many bookmarks (#158).** They fetched each bookmark in its own exchange with the hub, and changes that bookmarks share were sent again for each one; a pull with nothing new still asked for a bundle per bookmark. Now every bookmark's heads are fetched in one exchange, and none when nothing is new. On a 47 MB repository with 160 bookmarks: a pull with nothing new 169 s → 3.5 s, a clone 324 s → 26 s. `OXY_TIMINGS` now also shows the refs, transparency log and per-bookmark checks of a fetch.

// Bundles a local remote served on this thread, for tests that count

// round trips (oxygen#158).

#[cfg(test)]

thread_local! {

    pub(crate) static LOCAL_FETCHES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };

}

        #[cfg(test)]

        LOCAL_FETCHES.with(|n| n.set(n.get() + 1));

        let resp = {

            let _t = crate::timing::phase("fetch: refs");

            remote.refs()?

        };

        let log_phase = crate::timing::phase("fetch: transparency log");

        drop(log_phase);

        // One negotiation and one bundle for every bookmark's heads at

        // once (oxygen#158): a bookmark at a time cost a round trip each and

        // sent again the changes bookmarks share, as `have` did not grow

        // between them (160 bookmarks: 160 bundles, 49k changes encoded for

        // a few hundred). Heads already here, with every attestation's

        // signer known, need no round trip at all.

        let mut wanted: Vec<ChangeHash> = resp

            .refs

            .iter()

            .flat_map(|r| r.heads.iter().copied())

            .filter(|h| !self.store.has(h))

            .collect();

        wanted.sort();

        wanted.dedup();

        // A bundle also carries the identities and certificates of the keys

        // that signed the remote's attestations (spec §16.2): with no change

        // wanted, ask anyway when one of those signers is unknown here.

        let unknown_signer = resp

            .refs

            .iter()

            .filter_map(|r| r.attestation.as_ref())

            .chain(resp.tags.iter().filter_map(|t| t.attestation.as_ref()))

            .flat_map(|a| a.sigs.iter().map(|s| s.key))

            .any(|k| {

                self.store.get_identity(&k).is_err()

                    && !matches!(self.store.certificate_of(&k), Ok(Some(_)))

            });

        let shallow = self.is_shallow();

        let imported: Vec<ChangeHash> = if wanted.is_empty() && !unknown_signer {

            Vec::new()

        } else {

                remote.negotiate(&have, &wanted)?

            // Changes are not individually re-verified here: each bookmark's

            let _t = crate::timing::phase("fetch: import");

                self.import_bundle_shallow(&bundle, false)?

            } else {

                self.import_bundle(&bundle, false)?

        };

        if shallow {

            self.refresh_boundary(&imported)?;

        }

        if max_blob.is_some() {

            self.register_missing_blobs(&imported, remote_name)?;

        }

        let boundary = if shallow {

            self.boundary()

        } else {

            BTreeSet::new()

        };

        // Each bookmark's new changes: those just imported that its heads

        // reach, for the report.

        let mut new_deps: HashMap<ChangeHash, Vec<ChangeHash>> = HashMap::new();

        for h in &imported {

            if let Ok(m) = self.store.get_change_meta(h) {

                new_deps.insert(*h, m.deps);

        }

        let new_in = |heads: &[ChangeHash]| -> usize {

            let mut seen: HashSet<ChangeHash> = HashSet::new();

            let mut stack: Vec<ChangeHash> = heads

                .iter()

                .filter(|h| new_deps.contains_key(*h))

                .copied()

                .collect();

            while let Some(h) = stack.pop() {

                if seen.insert(h) {

                    stack.extend(

                        new_deps[&h]

                            .iter()

                            .filter(|d| new_deps.contains_key(*d) && !seen.contains(*d)),

                    );

                }

            }

            seen.len()

        };

        let checks = crate::timing::phase("fetch: check each bookmark");

        for r in resp.refs {

            let n = new_in(&r.heads);

            report.push((r.name, r.heads, n));

        drop(checks);

        let tags_phase = crate::timing::phase("fetch: tags");

        drop(tags_phase);

#[test]

fn a_fetch_of_many_bookmarks_takes_shared_changes_once() {

    // oxygen#158: every bookmark's heads are fetched in one exchange, so

    // changes bookmarks share are not sent again for each; each bookmark

    // still says how many of its changes were new, and a fetch with

    // nothing new imports nothing.

    hermetic();

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

    let root = dir.path();

    drop(Repo::init(&root.join("origin")).unwrap());

    let url = root.join("origin").to_string_lossy().to_string();

    let mut a = Repo::clone(&url, &root.join("a"))

        .unwrap()

        .with_home(root.join("home-a"));

    let w = root.join("a");

    write(&w, "f.txt", b"one\n");

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

    write(&w, "f.txt", b"one\ntwo\n");

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

    a.push("origin", false).unwrap();

    // Five bookmarks on main's history, each one change of its own.

    for i in 0..5 {

        a.bookmark_create(&format!("feat{i}"), None).unwrap();

        a.bookmark_switch(&format!("feat{i}"), false).unwrap();

        write(&w, &format!("feat{i}.txt"), format!("{i}\n").as_bytes());

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

        a.push("origin", false).unwrap();

        a.bookmark_switch("main", false).unwrap();

    }

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

        .unwrap()

        .with_home(root.join("home-b"));

    let report = b.fetch("origin").unwrap();

    assert_eq!(report.refs.len(), 6, "every bookmark: {:?}", report.refs);

    assert!(

        report.refs.iter().all(|(_, _, n)| *n == 0),

        "the clone brought them all: {:?}",

        report.refs

    );

    // One new change on one bookmark: counted there, nowhere else.

    a.bookmark_switch("feat3", false).unwrap();

    write(&w, "feat3.txt", b"3\nmore\n");

    a.commit("feat 3 again", &[]).unwrap();

    a.push("origin", false).unwrap();

    let fetches = || crate::protocol::LOCAL_FETCHES.with(|n| n.get());

    let at = fetches();

    let report = b.fetch("origin").unwrap();

    assert_eq!(fetches() - at, 1, "one bundle for all six bookmarks");

    let counts: Vec<(String, usize)> = report

        .refs

        .iter()

        .map(|(n, _, c)| (n.clone(), *c))

        .collect();

    assert!(counts.contains(&("feat3".to_string(), 1)), "{counts:?}");

    assert_eq!(

        counts.iter().filter(|(_, c)| *c > 0).count(),

        1,

        "only feat3 had anything new: {counts:?}"

    );

    // Nothing new: no bundle at all, every bookmark still listed.

    let at = fetches();

    let report = b.fetch("origin").unwrap();

    assert_eq!(fetches() - at, 0, "nothing to fetch, no round trip");

    assert_eq!(report.refs.len(), 6);

    assert!(

        report.refs.iter().all(|(_, _, n)| *n == 0),

        "{:?}",

        report.refs

    );

}