A push no longer passes a proxy's request limit (closes #133) d3b270909c78

- **A push no longer passes a proxy's request limit (#133).** A push to a hub behind Cloudflare, which takes at most 100 MB in one request, was refused `HTTP 413` with Cloudflare's HTML page printed raw: parts of a long push were about 256 MiB, and one change's blobs went in one request however many there were. Parts are now about 64 MiB (`push.batch-bytes`), a part whose blobs pass that sends them first in parts of objects alone, an HTML error page is shown by its title, and a 413 says how to push in smaller parts.

                // perhaps) is shown as it is, never read for hints. A

                // proxy's HTML error page is shown by its title only

                // (oxygen#133).

                let msg = format!("{status}: {}", plain_error_text(&text));

    } else if m.contains("http 413") {

        "the request was larger than the server, or a proxy in front of it, takes at once (Cloudflare's limit is 100 MB): push in smaller parts with `oxy config push.batch-bytes 33554432`, then push again"

/// An error response's text as it is, unless it is an HTML page (a proxy's,

/// such as Cloudflare's 413): then its title, or a note that it was one.

fn plain_error_text(text: &str) -> String {

    let t = text.trim();

    let lower = t.to_ascii_lowercase();

    if !(lower.starts_with("<!doctype html") || lower.starts_with("<html")) {

        return t.to_string();

    }

    let title = lower

        .find("<title>")

        .and_then(|i| {

            lower[i + 7..]

                .find("</title>")

                .map(|j| t[i + 7..i + 7 + j].trim())

        })

        .filter(|s| !s.is_empty());

    match title {

        Some(s) => format!("{s} (an HTML error page, from a proxy in front of the server perhaps)"),

        None => "an HTML error page (from a proxy in front of the server perhaps)".into(),

    }

}

    /// oxygen#133: a proxy's HTML error page is shown by its title, and a

    /// server's own text as it is; a 413 says to push in smaller parts.

    #[test]

    fn html_error_pages_are_shown_by_their_title() {

        let cloudflare = "<html>\r\n<head><title>413 Request Entity Too Large</title></head>\r\n<body><center><h1>413</h1></center><hr><center>cloudflare</center></body></html>";

        assert_eq!(

            plain_error_text(cloudflare),

            "413 Request Entity Too Large (an HTML error page, from a proxy in front of the server perhaps)"

        );

        assert_eq!(

            plain_error_text("  refused: a secret\nat a.txt:3 "),

            "refused: a secret\nat a.txt:3"

        );

        assert!(hint_for("https://hub/x: HTTP 413: too large")

            .unwrap()

            .contains("push.batch-bytes"));

    }

            // Well under the 100 MB a proxy in front of a hub may take

            // (oxygen#133).

            push_batch_bytes: 64 << 20,

    /// Send `bundle`'s blobs and chunks ahead of it, in parts of about

    /// `push.batch-bytes` of objects alone, when the whole would be larger

    /// (oxygen#133): one change can bring more blobs than one request may

    /// carry, and a proxy in front of a hub refuses a request past its limit

    /// (Cloudflare's is 100 MB). What is left, the changes with their

    /// identities and states, is returned to send as before. A receiver

    /// stores objects as they come, and checks a change's blobs are stored.

    fn send_objects_ahead(

        &self,

        remote: &dyn crate::protocol::Remote,

        remote_name: &str,

        mut bundle: Bundle,

        progress: crate::store::Progress,

    ) -> Result<Bundle> {

        let limit = self.config.push_batch_bytes;

        let changes: u64 = bundle

            .changes

            .iter()

            .map(|(c, _)| c.encode().len() as u64)

            .sum();

        let objects: u64 = bundle.objects.iter().map(|(_, b)| b.len() as u64).sum();

        if changes + objects <= limit {

            return Ok(bundle);

        }

        let (mut kept, mut parts, mut part, mut bytes) = (Vec::new(), Vec::new(), Vec::new(), 0u64);

        for (ty, body) in std::mem::take(&mut bundle.objects) {

            if !matches!(ty, ObjectType::Blob | ObjectType::Chunk) {

                kept.push((ty, body));

                continue;

            }

            if !part.is_empty() && bytes + body.len() as u64 > limit {

                parts.push(std::mem::take(&mut part));

                bytes = 0;

            }

            bytes += body.len() as u64;

            part.push((ty, body));

        }

        if !part.is_empty() {

            parts.push(part);

        }

        let n = parts.len();

        for (i, objects) in parts.into_iter().enumerate() {

            progress(&format!(

                "objects {} of {n}: {} blobs and chunks",

                i + 1,

                objects.len()

            ));

            let results = remote.push(

                Bundle {

                    objects,

                    ..Default::default()

                },

                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 objects part {} of {n}: {}",

                        i + 1,

                        r.reason

                    ))

                }

                None => {

                    return fail(format!(

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

                    ))

                }

            }

        }

        bundle.objects = kept;

        Ok(bundle)

    }

            let bundle = self.send_objects_ahead(

                remote.as_ref(),

                remote_name,

                self.bundle_for(batch)?,

                &mut *progress,

            )?;

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

        let mut bundle = self.send_objects_ahead(

            remote.as_ref(),

            remote_name,

            self.bundle_for(last)?,

            &mut *progress,

        )?;

/// oxygen#133: one change whose blobs pass `push.batch-bytes` sends them

/// first, in parts of objects alone, so no request is larger than about

/// that (a proxy in front of a hub refuses one past 100 MB); the change

/// and the bookmark follow, and a clone has every file.

#[test]

fn a_heavy_change_sends_its_blobs_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();

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

    repo.save_config().unwrap();

    let mut x: u64 = 0x9e3779b97f4a7c15;

    let mut noise = |n: usize| -> Vec<u8> {

        (0..n)

            .map(|_| {

                x ^= x << 13;

                x ^= x >> 7;

                x ^= x << 17;

                (x >> 24) as u8

            })

            .collect()

    };

    let files: Vec<Vec<u8>> = (0..4).map(|_| noise(40_000)).collect();

    for (i, f) in files.iter().enumerate() {

        write(&main, &format!("bin{i}.dat"), f);

    }

    repo.commit("four blobs", &[]).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!(

        parts.len() >= 3 && parts[0].starts_with("objects 1 of"),

        "{parts:?}"

    );

    let back = Repo::clone(&origin.to_string_lossy(), &root.join("back")).unwrap();

    for (i, f) in files.iter().enumerate() {

        assert_eq!(&fs::read(root.join(format!("back/bin{i}.dat"))).unwrap(), f);

    }

    drop(back);

}

**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 64 MiB, under the 100 MB a proxy in front of a hub may take) of changes and their texts and blobs, and the last part carries the reference updates; when one part's changes bring more than that in blobs, the blobs and chunks go first in parts of their own, objects alone (oxygen#133), 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.