Push a long history in parts cb8b8082fa73

- **A long push goes in parts (#95).** `oxy push` sent a whole history in one request, held in memory on both sides; the full VS Code import would have passed a hub's 4 GiB limit. It now sends parts of about `push.batch-bytes` (default 256 MiB) in dependency order, objects alone, and the last part moves the bookmark; the spinner says which part is going. A server stores a part and reports it as `objects`.

    /// `push.batch-bytes`: about how many bytes of changes and their texts

    /// one request of a long push carries (spec §16.3).

    pub push_batch_bytes: u64,

            push_batch_bytes: 256 << 20,

            "push.batch-bytes" => {

                self.push_batch_bytes = value.parse().ok().filter(|n| *n > 0).ok_or_else(|| {

                    RepoError::Msg(format!("expected a number of bytes, got {value:?}"))

                })?;

            }

pub use sync::{FetchReport, PullReport, PushReport, RefUpdated, OBJECTS_RESULT};

/// The result a push of objects alone reports (a batch of a long history):

/// not a reference.

pub const OBJECTS_RESULT: &str = "objects";

        // A push of objects alone (one batch of a long history, oxygen#95)

        // names no reference: its outcome is reported as `objects`.

        let objects_only = updates.is_empty() && tags.is_empty();

            Ok(imported) => {

                if objects_only {

                    results.push(RefResult {

                        name: OBJECTS_RESULT.into(),

                        ok: true,

                        reason: format!("{} changes stored", imported.len()),

                    });

                }

                report.imported = imported;

            }

                if objects_only {

                    results.push(RefResult {

                        name: OBJECTS_RESULT.into(),

                        ok: false,

                        reason: e.to_string(),

                    });

                }

    /// `changes`, in dependency order, cut into batches of about

    /// `push.batch-bytes` by what each change and its texts and blobs weigh.

    /// There is always one, perhaps empty.

    fn push_batches(&self, changes: &[ChangeHash]) -> Result<Vec<Vec<ChangeHash>>> {

        let blob_len = |r: &ObjectRef| -> u64 {

            self.store

                .get_typed(r, ObjectType::Blob)

                .ok()

                .and_then(|b| Blob::decode(&b).ok())

                .map_or(0, |b| b.length)

        };

        let (mut out, mut batch, mut bytes) = (Vec::new(), Vec::new(), 0u64);

        for h in changes {

            let raw = self.store.get_typed(h, ObjectType::Change)?;

            let (c, texts) = Change::decode_skeleton(&raw, *h)?;

            let mut weight = raw.len() as u64;

            for t in &texts {

                if let crate::objects::TextRef::Blob(r, _, _) = t {

                    weight += blob_len(r);

                }

            }

            for e in &c.edits {

                if let Edit::Replace { new: Some(b), .. } = e {

                    weight += blob_len(b);

                }

            }

            if !batch.is_empty() && bytes + weight > self.config.push_batch_bytes {

                out.push(std::mem::take(&mut batch));

                bytes = 0;

            }

            batch.push(*h);

            bytes += weight;

        }

        out.push(batch);

        Ok(out)

    }

        &self,

        remote_name: &str,

        bookmark: &str,

        force: bool,

    ) -> Result<Vec<RefResult>> {

        self.push_bookmark_with(remote_name, bookmark, force, &mut |_| {})

    }

    /// `push_bookmark`, saying how a long push is going: one line per

    /// batch sent (oxygen#95).

    pub fn push_bookmark_with(

        progress: crate::store::Progress,

            // and adopt anonymous ones (zero author, spec §14.4). The seals

            // are stored, so each batch's bundle carries those it needs.

                let c = self.store.get_change_meta(h)?;

                self.sign_attest(&Attest::seal(self.id()?, chunk.to_vec()))?;

        }

        // A long history goes in batches, in dependency order, so neither

        // side holds it whole and no request passes the remote's limit:

        // all but the last carry objects alone; the last moves the bookmark

        // (oxygen#95).

        let batches = self.push_batches(&to_send)?;

        let (last, earlier) = batches.split_last().expect("at least one batch");

        for (i, batch) in earlier.iter().enumerate() {

            progress(&format!(

                "part {} of {}: {} changes",

                i + 1,

                batches.len(),

                batch.len()

            ));

            let results = remote.push(self.bundle_for(batch)?, vec![], None, vec![])?;

            match results.iter().find(|r| r.name == OBJECTS_RESULT) {

                Some(r) if r.ok => {}

                Some(r) => {

                    return fail(format!(

                        "{remote_name} refused part {} of the push: {}",

                        i + 1,

                        r.reason

                    ))

                }

                None => {

                    return fail(format!(

                        "{remote_name} does not take a push in parts, and this one is too long for one: update the server's oxy"

                    ))

                }

            }

        }

        if batches.len() > 1 {

            progress(&format!(

                "part {} of {}: {} changes",

                batches.len(),

                batches.len(),

                last.len()

            ));

        let mut bundle = self.bundle_for(last)?;

/// oxygen#95: a push longer than `push.batch-bytes` goes in parts, each

/// objects alone and in dependency order, and the last moves the bookmark.

/// The remote ends with every change and the same state, and says how each

/// part went.

#[test]

fn a_long_push_goes_in_parts() {

    hermetic();

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

    let root = dir.path();

    let origin = root.join("origin");

    Repo::init_bare_unbound(&origin).unwrap();

    let main = root.join("main");

    let mut repo = Repo::init(&main).unwrap();

    repo.config

        .set("remote.origin.url", &origin.to_string_lossy())

        .unwrap();

    // One byte: every change is a part of its own.

    repo.config.set("push.batch-bytes", "1").unwrap();

    repo.save_config().unwrap();

    for i in 0..6 {

        write(

            &main,

            &format!("f{i}.txt"),

            format!("file {i}\n").as_bytes(),

        );

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

    }

    let mut parts = Vec::new();

    let results = repo

        .push_bookmark_with("origin", "main", false, &mut |m| parts.push(m.to_string()))

        .unwrap();

    assert!(

        results.iter().any(|r| r.ok && r.name == "main"),

        "{results:?}"

    );

    assert_eq!(parts.len(), 6, "{parts:?}");

    assert_eq!(parts[0], "part 1 of 6: 1 changes");

    let remote = Repo::open(&origin).unwrap();

    let heads = repo.bookmark_heads().unwrap();

    assert_eq!(remote.bookmark_heads().unwrap(), heads);

    assert_eq!(

        remote.log_state(&heads).unwrap().len(),

        repo.log_state(&heads).unwrap().len()

    );

    // A push of objects alone reports how it went, as `objects`.

    let got = remote

        .receive_push(Bundle::default(), vec![], None, vec![])

        .unwrap();

    assert_eq!(got.len(), 1);

    assert!(got[0].ok && got[0].name == OBJECTS_RESULT, "{got:?}");

}

                // The request named references, all refused here: it was no

                // push of objects alone, whatever the repository saw.

                if !refused.is_empty() {

                    report

                        .results

                        .retain(|r| r.name != oxy_core::repo::OBJECTS_RESULT);

                }

**A push in parts.** A push may carry no reference update and no tag: its changes and objects alone, which the server validates as above and stores, reporting the outcome as one result named `objects` (`ok`, and as its reason how many changes it stored or why it refused them). A client sends a long history this way, in dependency order, in parts of about `push.batch-bytes` (default 256 MiB) of changes and their texts and blobs, and the last part carries the reference updates, so that neither side holds the whole history at once and no request passes the server's body limit. A part already stored is harmless if the push stops, and a server MAY collect its changes under §21 if no reference ever reaches them.

    let bookmark_name = bookmark.map_or_else(|| repo.config.bookmark.clone(), String::from);

    // A long push goes in parts: the spinner says which.

    let results = repo.push_bookmark_with(&remote, &bookmark_name, force, &mut |m| {

        sp.set(&format!("Pushing to {remote}… {m}"))

    })?;