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

@@ -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<BufReader<ChildStdout>>,
+}
+
+/// 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<Self> {
+        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;
+    }
+}

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<FetchWorker> = 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<String, (String, anyhow::Error)> {
     Ok(oid)
 }
 
-async fn handle_download(msg: &Value) -> Value {
-    match download_inner(msg).await {
+async fn handle_download(msg: &Value, worker: &mut Option<FetchWorker>) -> 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<FetchWorker>,
+) -> 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