From 9deff4b68d86ad3a222e9326d39c1d7e699dac05 Mon Sep 17 00:00:00 2001 From: Seongmin Lee Date: Sat, 10 Oct 2026 00:49:56 +0900 Subject: [PATCH] knot-pack: send keepalives during upload-pack object selection Large clones (e.g. nixpkgs) spend more than a minute selecting objects before sending any pack data. Proxies and clients might drop the connection for timeout. Like git's uploadpack.keepAlive, run selection on a worker thread and write an empty data-band packet every 5s until it finishes. Keepalives only go out on sideband connections, after the `packfile` line, so packfile-uris requests still select silently. Signed-off-by: Seongmin Lee --- knot2/crates/knot-pack/src/upload.rs | 125 ++++++++++++++++++++++----- 1 file changed, 103 insertions(+), 22 deletions(-) diff --git a/knot2/crates/knot-pack/src/upload.rs b/knot2/crates/knot-pack/src/upload.rs index d11f95a2..cdf2cd07 100644 --- a/knot2/crates/knot-pack/src/upload.rs +++ b/knot2/crates/knot-pack/src/upload.rs @@ -17,6 +17,7 @@ use crate::{HaveOids, WantOids}; const AGENT: &[u8] = b"agent=knot/0\n"; const SELECTION_MAX_OBJECTS: ObjectCount = ObjectCount::new(16_000_000); const SELECTION_TIME_BUDGET: Duration = Duration::from_secs(600); +const KEEPALIVE: Duration = Duration::from_secs(5); #[derive(Debug, Clone, Copy)] pub struct SelectionLimits { @@ -636,28 +637,44 @@ pub fn stream_pack( 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 = 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() { + 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)?; } - write_packfile_header(opts, sink)?; let count = ObjectCount::new(selection.send.len()); emit_preamble(progress, messages, knot, count, sink)?; { @@ -854,14 +871,16 @@ fn stream_full_clone( ) -> Result<(), PackError> { let limits = selection_limits(); let stall = limits.time_budget; - 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 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 count = ObjectCount::new(pack.len()); emit_preamble(progress, messages, knot, count, sink)?; { @@ -874,6 +893,38 @@ fn stream_full_clone( emit_total(progress, messages, count, sink) } +fn with_keepalive( + repo: &Repo, + keepalive: bool, + interval: Duration, + sink: &mut dyn FnMut(&[u8]) -> io::Result<()>, + work: impl FnOnce(&Repo) -> Result + Send, +) -> Result { + if !keepalive { + return work(repo); + } + let path = repo.path().to_owned(); + std::thread::scope(|scope| { + let (done, finished) = std::sync::mpsc::channel::<()>(); + let worker = scope.spawn(move || { + let _done = done; + // knot_git::Repo is not thread-safe. + work(&Repo::open(path)?) + }); + let mut alive = Ok(()); + while alive.is_ok() + && finished.recv_timeout(interval) == Err(std::sync::mpsc::RecvTimeoutError::Timeout) + { + alive = sink(b"0005\x01"); + } + let out = worker + .join() + .unwrap_or_else(|panic| std::panic::resume_unwind(panic)); + alive?; + out + }) +} + struct PackSink<'a> { side_band: bool, sink: &'a mut dyn FnMut(&[u8]) -> io::Result<()>, @@ -977,3 +1028,33 @@ fn parse_oids(lines: &[&[u8]], prefix: &[u8]) -> Vec { }) .collect() } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn slow_work_is_kept_alive_with_empty_data_bands() { + let dir = tempfile::tempdir().unwrap(); + gix::init_bare(dir.path()).unwrap(); + let repo = Repo::open(dir.path()).unwrap(); + let interval = Duration::from_millis(20); + let mut out = Vec::new(); + let value = with_keepalive( + &repo, + true, + interval, + &mut |chunk| { + out.extend_from_slice(chunk); + Ok(()) + }, + |_| { + std::thread::sleep(interval * 5); + Ok(7) + }, + ) + .unwrap(); + assert_eq!(value, 7); + assert!(out.len() >= 5 && out.chunks(5).all(|packet| packet == b"0005\x01")); + } +} -- 2.54.0