From e0ef7981da82fe1516cbdc2992b3023f4b2ca717 Mon Sep 17 00:00:00 2001 From: Seongmin Lee Date: Sat, 10 Oct 2026 21:02:28 +0900 Subject: [PATCH] knot-pack: keep upload-pack alive until the pack is fully sent The keepalive only covered object selection. On a full clone the selection is a quick bitmap walk, and the silence comes afterwards: write_pack counts all 9.4M objects before writing its first byte, so Cloudflare dropped the connection after 60s. Run the whole response on the worker and forward its output through a channel, sending an empty data band whenever it's quiet for 5s, like git's uploadpack.keepAlive. Signed-off-by: Seongmin Lee --- .cargo/config.toml | 3 + knot2/crates/knot-pack/src/upload.rs | 196 ++++++++++++---------- knot2/crates/knot-pack/tests/hardening.rs | 12 +- 3 files changed, 109 insertions(+), 102 deletions(-) create mode 100644 .cargo/config.toml diff --git a/.cargo/config.toml b/.cargo/config.toml new file mode 100644 index 00000000..59cfbcd3 --- /dev/null +++ b/.cargo/config.toml @@ -0,0 +1,3 @@ +[target.aarch64-apple-darwin] +linker = "arm64-apple-darwin-cc" + diff --git a/knot2/crates/knot-pack/src/upload.rs b/knot2/crates/knot-pack/src/upload.rs index cdf2cd07..ab6fd25c 100644 --- a/knot2/crates/knot-pack/src/upload.rs +++ b/knot2/crates/knot-pack/src/upload.rs @@ -367,7 +367,11 @@ fn plan_v2_fetch(repo: &Repo, lines: &[&[u8]]) -> Result packfile_uri_candidates(repo, lines), + false => Vec::new(), + }, emit_packfile_header: true, }, }) @@ -620,78 +624,69 @@ pub fn stream_pack( sink: &mut dyn FnMut(&[u8]) -> io::Result<()>, ) -> Result<(), PackError> { let progress = opts.side_band && !opts.no_progress; - if opts.shallow_commits.is_none() - && haves.is_empty() - && opts.filter == Filter::None - && opts.packfile_uris.is_empty() - { - if let Ok(Some(pack)) = knot_git::verbatim_clone_pack(repo, wants.wants()) { + // the `packfile` section opens before keepalives start, unless `packfile-uris` must precede it + if opts.packfile_uris.is_empty() { + write_packfile_header(opts, sink)?; + } + with_keepalive(repo, opts.side_band, KEEPALIVE, sink, |repo, sink| { + if opts.shallow_commits.is_none() + && haves.is_empty() + && opts.filter == Filter::None + && opts.packfile_uris.is_empty() + { + if let Ok(Some(pack)) = knot_git::verbatim_clone_pack(repo, wants.wants()) { + return stream_verbatim_pack(pack, opts, progress, messages, knot, sink); + } + if let Ok(Some(oids)) = + knot_git::reachable_via_bitmap(repo, wants.wants(), Haves::new(&[])) + { + return stream_object_set(repo, oids, opts, progress, messages, knot, sink); + } + return stream_full_clone(repo, wants.as_slice(), opts, progress, messages, knot, sink); + } + let budget = selection_budget(); + let mut selection = match &opts.shallow_commits { + Some(commits) => repo.select_shallow_objects( + wants.wants(), + ShallowCommits::new(commits), + haves.haves(), + opts.filter, + budget, + )?, + None => repo.select_pack_objects_filtered( + wants.wants(), + haves.haves(), + opts.filter, + budget, + )?, + }; + if !opts.packfile_uris.is_empty() { + offload_packfile_uris( + &mut selection.send, + &opts.packfile_uris, + opts.sideband_all, + sink, + )?; write_packfile_header(opts, sink)?; - return stream_verbatim_pack(pack, opts, progress, messages, knot, sink); } - if let Ok(Some(oids)) = knot_git::reachable_via_bitmap(repo, wants.wants(), Haves::new(&[])) + let count = ObjectCount::new(selection.send.len()); + emit_preamble(progress, messages, knot, count, sink)?; { - write_packfile_header(opts, sink)?; - return stream_object_set(repo, oids, opts, progress, messages, knot, sink); + let mut pack_sink = PackSink { + side_band: opts.side_band, + sink: &mut *sink, + }; + let thin_bases = opts.thin.then_some(&selection.client_has); + objects::write_pack( + &repo.objects_dir(), + selection.send, + thin_bases, + &mut pack_sink, + repo.object_format().kind(), + )?; } - write_packfile_header(opts, sink)?; - return stream_full_clone(repo, wants.as_slice(), opts, progress, messages, knot, sink); - } - // keepalives may only follow the `packfile` line, which must follow `packfile-uris` - let early_header = opts.packfile_uris.is_empty(); - if early_header { - write_packfile_header(opts, sink)?; - } - let budget = selection_budget(); - let mut selection = with_keepalive( - repo, - opts.side_band && early_header, - KEEPALIVE, - sink, - |repo| { - Ok(match &opts.shallow_commits { - Some(commits) => repo.select_shallow_objects( - wants.wants(), - ShallowCommits::new(commits), - haves.haves(), - opts.filter, - budget, - )?, - None => repo.select_pack_objects_filtered( - wants.wants(), - haves.haves(), - opts.filter, - budget, - )?, - }) - }, - )?; - if !early_header { - offload_packfile_uris( - &mut selection.send, - &opts.packfile_uris, - opts.sideband_all, - sink, - )?; - write_packfile_header(opts, sink)?; - } - let count = ObjectCount::new(selection.send.len()); - emit_preamble(progress, messages, knot, count, sink)?; - { - let mut pack_sink = PackSink { - side_band: opts.side_band, - sink: &mut *sink, - }; - let thin_bases = opts.thin.then_some(&selection.client_has); - objects::write_pack( - &repo.objects_dir(), - selection.send, - thin_bases, - &mut pack_sink, - repo.object_format().kind(), - )?; - } - emit_total(progress, messages, count, sink) + emit_total(progress, messages, count, sink) + }) } fn emit_progress( @@ -871,16 +866,14 @@ fn stream_full_clone( ) -> Result<(), PackError> { let limits = selection_limits(); let stall = limits.time_budget; - let pack = with_keepalive(repo, opts.side_band, KEEPALIVE, sink, |repo| { - let roots = repo.clone_roots(wants, PackBudget::new(limits.max_objects, stall))?; - objects::count_expanded( - &repo.objects_dir(), - roots, - limits.max_objects, - stall, - repo.object_format().kind(), - ) - })?; + let roots = repo.clone_roots(wants, PackBudget::new(limits.max_objects, stall))?; + let pack = objects::count_expanded( + &repo.objects_dir(), + roots, + limits.max_objects, + stall, + repo.object_format().kind(), + )?; let count = ObjectCount::new(pack.len()); emit_preamble(progress, messages, knot, count, sink)?; { @@ -893,34 +886,45 @@ fn stream_full_clone( emit_total(progress, messages, count, sink) } +/// Runs `work` on a worker thread, forwarding its output to `sink` and sending an empty data +/// band whenever it stays silent for `interval`. fn with_keepalive( repo: &Repo, keepalive: bool, interval: Duration, sink: &mut dyn FnMut(&[u8]) -> io::Result<()>, - work: impl FnOnce(&Repo) -> Result + Send, + work: impl FnOnce(&Repo, &mut dyn FnMut(&[u8]) -> io::Result<()>) -> Result + Send, ) -> Result { if !keepalive { - return work(repo); + return work(repo, sink); } let path = repo.path().to_owned(); std::thread::scope(|scope| { - let (done, finished) = std::sync::mpsc::channel::<()>(); + let (tx, rx) = std::sync::mpsc::sync_channel::>(16); let worker = scope.spawn(move || { - let _done = done; + let mut send = |data: &[u8]| { + tx.send(data.to_vec()) + .map_err(|_| io::Error::from(io::ErrorKind::BrokenPipe)) + }; // knot_git::Repo is not thread-safe. - work(&Repo::open(path)?) + work(&Repo::open(path)?, &mut send) }); - let mut alive = Ok(()); - while alive.is_ok() - && finished.recv_timeout(interval) == Err(std::sync::mpsc::RecvTimeoutError::Timeout) - { - alive = sink(b"0005\x01"); - } + let forwarded = loop { + let sent = match rx.recv_timeout(interval) { + Ok(data) => sink(&data), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => sink(b"0005\x01"), + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break Ok(()), + }; + if let Err(error) = sent { + break Err(error); + } + }; + // unblocks the worker if the client went away + drop(rx); let out = worker .join() .unwrap_or_else(|panic| std::panic::resume_unwind(panic)); - alive?; + forwarded?; out }) } @@ -1048,13 +1052,19 @@ mod tests { out.extend_from_slice(chunk); Ok(()) }, - |_| { + |_, sink| { + sink(b"first")?; std::thread::sleep(interval * 5); + sink(b"last")?; Ok(7) }, ) .unwrap(); assert_eq!(value, 7); - assert!(out.len() >= 5 && out.chunks(5).all(|packet| packet == b"0005\x01")); + let middle = out + .strip_prefix(b"first") + .and_then(|rest| rest.strip_suffix(b"last")) + .unwrap(); + assert!(middle.len() >= 5 && middle.chunks(5).all(|packet| packet == b"0005\x01")); } } diff --git a/knot2/crates/knot-pack/tests/hardening.rs b/knot2/crates/knot-pack/tests/hardening.rs index d8cfd7f8..f348921c 100644 --- a/knot2/crates/knot-pack/tests/hardening.rs +++ b/knot2/crates/knot-pack/tests/hardening.rs @@ -884,15 +884,9 @@ fn v2_sideband_all_band_frames_negotiation_and_packfile_uris() { .unwrap(); let plain_uri_payloads = pkt_payloads(&plain_uris); assert!( - plain(&plain_uri_payloads, b"packfile-uris\n"), - "without sideband-all packfile-uris header is a plain pkt-line, not band 1" - ); - assert!( - plain( - &plain_uri_payloads, - format!("{packhash} {uri}\n").as_bytes() - ), - "configured blob uri must be advertised verbatim" + !plain(&plain_uri_payloads, b"packfile-uris\n") + && !banded(&plain_uri_payloads, b"packfile-uris\n"), + "packfile-uris are ignored without sideband-all" ); let banded_uris = knot_pack::upload_pack( -- 2.54.0