From 3215f231d80752266d2438f760c9dc244f1fc828 Mon Sep 17 00:00:00 2001 From: "Maciek \"mab122\" Bator" Date: Fri, 17 Jul 2026 23:42:17 +0200 Subject: [PATCH] Fetch multiple LFS objects through one long-lived worker, not per-object download_inner used to spawn a fresh `rad lfs fetch` subprocess for every object, each one independently loading the signer -- for a private repo, checking out N files meant N separate passphrase prompts (or N failures without RAD_PASSPHRASE). Same bug class the pre-commit hook had before it was batched into `rad lfs precommit`, just on the download side. Fetch can't batch the same way: git-lfs's custom-transfer protocol requests objects one at a time, so unlike the pre-commit hook (which knows every staged file upfront), there's no point where the full object set is known before starting. Instead, `fetch_worker::FetchWorker` spawns `rad lfs fetch-batch` (new, radicle-heartwood-lfs) lazily on first use and keeps it alive for the rest of the session, one request per download, one JSON response line back -- so the signer is loaded (and the passphrase prompted for) at most once per session. If the worker process itself dies (broken pipe, unexpected exit) -- as opposed to a legitimate per-object failure like a missing note or wrong passphrase, which isn't retried -- one respawn-and-retry is attempted before giving up. --- src/fetch_worker.rs | 103 ++++++++++++++++++++++++++++++++++++++++++++ src/main.rs | 85 ++++++++++++++++++++++-------------- 2 files changed, 156 insertions(+), 32 deletions(-) create mode 100644 src/fetch_worker.rs diff --git a/src/fetch_worker.rs b/src/fetch_worker.rs new file mode 100644 index 0000000..eecc556 --- /dev/null +++ b/src/fetch_worker.rs @@ -0,0 +1,103 @@ +//! Manages a long-lived `rad lfs fetch-batch` child process, spawned +//! lazily on first use and kept alive for the whole transfer session, so +//! a private repo's passphrase is only prompted for once per session +//! rather than once per downloaded object. +//! +//! This can't be a simple read-until-EOF batch the way `rad lfs +//! precommit` is (which knows every staged file upfront): git-lfs's +//! custom-transfer protocol requests objects one at a time by default, +//! so we only learn about the next object once the previous one has been +//! answered. Instead, the worker stays alive across the whole session and +//! answers one request at a time as they arrive -- see `rad lfs +//! fetch-batch`'s own doc comment for the line protocol. + +use std::process::Stdio; + +use anyhow::Context as _; +use serde_json::Value; +use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader, Lines}; +use tokio::process::{Child, ChildStdin, ChildStdout, Command}; + +pub struct FetchWorker { + child: Child, + stdin: ChildStdin, + lines: Lines>, +} + +/// Distinguishes a legitimate per-object failure (wrong passphrase, +/// missing note, decryption failure -- retrying with a fresh worker +/// process won't help) from the worker process itself having died +/// (broken pipe, unexpected exit -- worth one respawn-and-retry, since +/// that's likely a transient problem, not the specific object's fault). +pub enum FetchError { + WorkerDied(anyhow::Error), + ObjectFailed(String), +} + +impl FetchWorker { + pub async fn spawn() -> anyhow::Result { + let mut child = Command::new("rad") + .args(["lfs", "fetch-batch"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .kill_on_drop(true) + .spawn() + .context("failed to spawn `rad lfs fetch-batch`")?; + + let stdin = child.stdin.take().context("child has no stdin")?; + let stdout = child.stdout.take().context("child has no stdout")?; + let lines = BufReader::new(stdout).lines(); + + Ok(Self { child, stdin, lines }) + } + + pub async fn fetch(&mut self, oid: &str, size: i64, out: &str) -> Result<(), FetchError> { + let request = format!("{oid} {size} {out}\n"); + self.stdin + .write_all(request.as_bytes()) + .await + .context("failed to write to `rad lfs fetch-batch`") + .map_err(FetchError::WorkerDied)?; + self.stdin + .flush() + .await + .context("failed to flush `rad lfs fetch-batch` stdin") + .map_err(FetchError::WorkerDied)?; + + let line = self + .lines + .next_line() + .await + .context("failed to read from `rad lfs fetch-batch`") + .map_err(FetchError::WorkerDied)? + .ok_or_else(|| { + FetchError::WorkerDied(anyhow::anyhow!( + "`rad lfs fetch-batch` closed its output unexpectedly" + )) + })?; + + let value: Value = serde_json::from_str(&line) + .with_context(|| format!("invalid response from `rad lfs fetch-batch`: {line:?}")) + .map_err(FetchError::WorkerDied)?; + + if value.get("ok").and_then(Value::as_bool) == Some(true) { + Ok(()) + } else { + let message = value + .get("error") + .and_then(Value::as_str) + .unwrap_or("unknown error") + .to_string(); + Err(FetchError::ObjectFailed(message)) + } + } + + pub async fn shutdown(mut self) { + // Dropping `stdin` closes the pipe, which is what makes the + // worker's `for line in stdin.lock().lines()` loop see EOF and + // exit cleanly on its own. + drop(self.stdin); + let _ = self.child.wait().await; + } +} diff --git a/src/main.rs b/src/main.rs index cdfdbad..bfcbada 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,3 +1,4 @@ +mod fetch_worker; mod kubo; use anyhow::Context as _; @@ -5,6 +6,7 @@ use serde_json::{Value, json}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::process::Command; +use fetch_worker::{FetchError, FetchWorker}; use kubo::Kubo; #[tokio::main] @@ -17,6 +19,10 @@ async fn main() -> anyhow::Result<()> { let mut lines = BufReader::new(stdin).lines(); let mut stdout = tokio::io::stdout(); + // Lazily spawned on the first "download" event, kept alive for the + // rest of the session -- see `fetch_worker`'s doc comment for why. + let mut fetch_worker: Option = None; + while let Some(line) = lines.next_line().await? { if line.trim().is_empty() { continue; @@ -37,7 +43,7 @@ async fn main() -> anyhow::Result<()> { write_line(&mut stdout, &resp).await?; } "download" => { - let resp = handle_download(&msg).await; + let resp = handle_download(&msg, &mut fetch_worker).await; write_line(&mut stdout, &resp).await?; } "terminate" => break, @@ -51,6 +57,10 @@ async fn main() -> anyhow::Result<()> { } } + if let Some(worker) = fetch_worker { + worker.shutdown().await; + } + Ok(()) } @@ -104,8 +114,8 @@ async fn upload_inner(msg: &Value) -> Result { Ok(oid) } -async fn handle_download(msg: &Value) -> Value { - match download_inner(msg).await { +async fn handle_download(msg: &Value, worker: &mut Option) -> Value { + match download_inner(msg, worker).await { Ok((oid, path)) => json!({"event": "complete", "oid": oid, "path": path}), Err((oid, e)) => json!({ "event": "complete", @@ -115,9 +125,15 @@ async fn handle_download(msg: &Value) -> Value { } } -/// Delegates entirely to `rad lfs fetch`, which looks up the note, fetches -/// from IPFS, and decrypts (if needed) on the Radicle side. -async fn download_inner(msg: &Value) -> Result<(String, String), (String, anyhow::Error)> { +/// Delegates to the long-lived `rad lfs fetch-batch` worker (see +/// `fetch_worker`), which looks up the note, fetches from IPFS, and +/// decrypts (if needed) on the Radicle side -- kept alive across every +/// download in this session rather than re-spawned per object, so a +/// private repo's passphrase is only prompted for once per session. +async fn download_inner( + msg: &Value, + worker: &mut Option, +) -> Result<(String, String), (String, anyhow::Error)> { let oid = field_str(msg, "oid").map_err(|e| (String::new(), e))?; let size = field_i64(msg, "size").map_err(|e| (oid.clone(), e))?; @@ -132,35 +148,40 @@ async fn download_inner(msg: &Value) -> Result<(String, String), (String, anyhow // construction. let tmp_dir = git_lfs_tmp_dir().await.map_err(|e| (oid.clone(), e))?; let tmp_path = tmp_dir.join(format!("rad-lfs-{oid}")); + let out = tmp_path.to_string_lossy().to_string(); - let output = Command::new("rad") - .args([ - "lfs", - "fetch", - "--oid", - &oid, - "--size", - &size.to_string(), - "--out", - ]) - .arg(&tmp_path) - .output() - .await - .map_err(|e| { - ( - oid.clone(), - anyhow::Error::from(e).context("failed to spawn `rad lfs fetch`"), - ) - })?; - - if !output.status.success() { - return Err(( - oid, - anyhow::anyhow!("rad lfs fetch failed: {}", String::from_utf8_lossy(&output.stderr)), - )); + if worker.is_none() { + *worker = Some( + FetchWorker::spawn() + .await + .map_err(|e| (oid.clone(), e.context("failed to start `rad lfs fetch-batch`")))?, + ); } - Ok((oid, tmp_path.to_string_lossy().to_string())) + // One respawn-and-retry if the worker process itself died (broken + // pipe, unexpected exit) -- a per-object failure (wrong passphrase, + // no note, decryption failure) is not retried, since a fresh worker + // wouldn't change that outcome. + match worker.as_mut().expect("just populated above").fetch(&oid, size, &out).await { + Ok(()) => Ok((oid, out)), + Err(FetchError::ObjectFailed(message)) => Err((oid, anyhow::anyhow!(message))), + Err(FetchError::WorkerDied(err)) => { + worker.take(); + let mut fresh = FetchWorker::spawn() + .await + .map_err(|e| (oid.clone(), e.context("failed to restart `rad lfs fetch-batch`")))?; + let result = fresh.fetch(&oid, size, &out).await; + *worker = Some(fresh); + match result { + Ok(()) => Ok((oid, out)), + Err(FetchError::ObjectFailed(message)) => Err((oid, anyhow::anyhow!(message))), + Err(FetchError::WorkerDied(retry_err)) => Err(( + oid, + retry_err.context(format!("`rad lfs fetch-batch` worker died twice in a row (first: {err:#})")), + )), + } + } + } } /// Resolves (and creates, if needed) `.git/lfs/tmp` for the current