lfs: add fetch-batch, a long-lived worker for multi-object downloads
`rad lfs fetch` still exists for single-object/manual use, but `radicle-lfs-transfer` used to spawn a fresh one per object during `git lfs pull`/checkout -- for a private repo, each one independently loaded the signer, so N files meant N passphrase prompts. Same bug class `rad lfs precommit` already fixed for commits, unfixed here until now. Unlike precommit (which knows every staged file upfront and can read stdin until EOF), fetch can't collect a batch before starting: git-lfs's custom-transfer protocol requests objects one at a time. `rad lfs fetch-batch` instead stays alive for as long as its caller keeps its stdin open, answering one "<oid> <size> <out-path>" request at a time with a JSON response line, reusing the same unwrapped signer across all of them -- paired with radicle-lfs-transfer's new FetchWorker, which spawns this once per transfer session instead of once per object. Also factors `fetch::run`'s core logic out into `fetch_object`, taking a `&mut Option<MemorySigner>` the same way `store::store_object` already does, so both the single-shot `rad lfs fetch` and the new batch worker share one implementation.
This commit is contained in:
parent
4ab6fc2499
commit
7c6d342ef9
|
|
@ -2,6 +2,7 @@
|
||||||
|
|
||||||
pub mod backfill;
|
pub mod backfill;
|
||||||
pub mod fetch;
|
pub mod fetch;
|
||||||
|
pub mod fetch_batch;
|
||||||
pub mod init;
|
pub mod init;
|
||||||
pub mod precommit;
|
pub mod precommit;
|
||||||
pub mod rekey;
|
pub mod rekey;
|
||||||
|
|
@ -22,5 +23,6 @@ pub fn run(args: Args, ctx: impl term::Context) -> anyhow::Result<()> {
|
||||||
Command::Rekey => self::rekey::run(ctx),
|
Command::Rekey => self::rekey::run(ctx),
|
||||||
Command::Precommit => self::precommit::run(ctx),
|
Command::Precommit => self::precommit::run(ctx),
|
||||||
Command::Backfill => self::backfill::run(ctx),
|
Command::Backfill => self::backfill::run(ctx),
|
||||||
|
Command::FetchBatch => self::fetch_batch::run(ctx),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -55,4 +55,10 @@ pub enum Command {
|
||||||
/// without going through the pre-commit hook (e.g. `--no-verify`, or
|
/// without going through the pre-commit hook (e.g. `--no-verify`, or
|
||||||
/// the IPFS daemon/`ipfs`/`rad` weren't available at commit time)
|
/// the IPFS daemon/`ipfs`/`rad` weren't available at commit time)
|
||||||
Backfill,
|
Backfill,
|
||||||
|
/// Long-lived worker that fetches many LFS objects across one process
|
||||||
|
/// (invoked by the custom transfer agent; not for interactive use).
|
||||||
|
/// Reads "<oid> <size> <out-path>" lines from stdin, one at a time,
|
||||||
|
/// writing a JSON response line for each.
|
||||||
|
#[command(hide = true)]
|
||||||
|
FetchBatch,
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -3,10 +3,15 @@
|
||||||
//! content from the local IPFS (Kubo) node, decrypting it first if the
|
//! content from the local IPFS (Kubo) node, decrypting it first if the
|
||||||
//! repository is private.
|
//! repository is private.
|
||||||
|
|
||||||
use std::path::PathBuf;
|
use std::path::{Path, PathBuf};
|
||||||
|
|
||||||
use anyhow::{Context as _, anyhow};
|
use anyhow::{Context as _, anyhow};
|
||||||
|
|
||||||
|
use radicle::Profile;
|
||||||
|
use radicle::git::raw::Repository;
|
||||||
|
use radicle::identity::RepoId;
|
||||||
|
use radicle_crypto::ssh::keystore::MemorySigner;
|
||||||
|
|
||||||
use crate::ipfs;
|
use crate::ipfs;
|
||||||
use crate::lfs_crypto;
|
use crate::lfs_crypto;
|
||||||
use crate::terminal as term;
|
use crate::terminal as term;
|
||||||
|
|
@ -16,13 +21,32 @@ pub fn run(oid: String, size: i64, out: PathBuf, ctx: impl term::Context) -> any
|
||||||
let (repo, rid) = radicle::rad::cwd()
|
let (repo, rid) = radicle::rad::cwd()
|
||||||
.context("`rad lfs fetch` must be run inside a Radicle repository working copy")?;
|
.context("`rad lfs fetch` must be run inside a Radicle repository working copy")?;
|
||||||
|
|
||||||
|
let mut signer = None;
|
||||||
|
fetch_object(&profile, &repo, rid, &oid, size, &out, &mut signer)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Fetches one LFS object's content from IPFS (decrypting it first for
|
||||||
|
/// private repos) and writes it to `out`. Factored out of [`run`] so
|
||||||
|
/// `rad lfs fetch-batch` can process many objects across a single
|
||||||
|
/// process, loading (and prompting for) the signer at most once across
|
||||||
|
/// however many objects it ends up handling, instead of once per object.
|
||||||
|
#[allow(clippy::too_many_arguments)]
|
||||||
|
pub fn fetch_object(
|
||||||
|
profile: &Profile,
|
||||||
|
repo: &Repository,
|
||||||
|
rid: RepoId,
|
||||||
|
oid: &str,
|
||||||
|
size: i64,
|
||||||
|
out: &Path,
|
||||||
|
signer: &mut Option<MemorySigner>,
|
||||||
|
) -> anyhow::Result<()> {
|
||||||
let pointer_text =
|
let pointer_text =
|
||||||
format!("version https://git-lfs.github.com/spec/v1\noid sha256:{oid}\nsize {size}\n");
|
format!("version https://git-lfs.github.com/spec/v1\noid sha256:{oid}\nsize {size}\n");
|
||||||
let blob_oid = repo
|
let blob_oid = repo
|
||||||
.blob(pointer_text.as_bytes())
|
.blob(pointer_text.as_bytes())
|
||||||
.context("failed to write LFS pointer blob")?;
|
.context("failed to write LFS pointer blob")?;
|
||||||
|
|
||||||
let candidates = lfs_crypto::find_envelopes(&repo, blob_oid)
|
let candidates = lfs_crypto::find_envelopes(repo, blob_oid)
|
||||||
.with_context(|| format!("failed to read LFS notes for oid {oid}"))?;
|
.with_context(|| format!("failed to read LFS notes for oid {oid}"))?;
|
||||||
if candidates.is_empty() {
|
if candidates.is_empty() {
|
||||||
anyhow::bail!(
|
anyhow::bail!(
|
||||||
|
|
@ -43,13 +67,16 @@ pub fn run(oid: String, size: i64, out: PathBuf, ctx: impl term::Context) -> any
|
||||||
let result = match &envelope.enc {
|
let result = match &envelope.enc {
|
||||||
None => ipfs::cat(&envelope.cid),
|
None => ipfs::cat(&envelope.cid),
|
||||||
Some(enc) => ipfs::cat(&envelope.cid).and_then(|ciphertext| {
|
Some(enc) => ipfs::cat(&envelope.cid).and_then(|ciphertext| {
|
||||||
let signer = lfs_crypto::load_signer(&profile)?;
|
if signer.is_none() {
|
||||||
lfs_crypto::decrypt_with(&ciphertext, enc, &signer, profile.did(), &rid, &oid)
|
*signer = Some(lfs_crypto::load_signer(profile)?);
|
||||||
|
}
|
||||||
|
let signer = signer.as_ref().expect("just populated above");
|
||||||
|
lfs_crypto::decrypt_with(&ciphertext, enc, signer, profile.did(), &rid, oid)
|
||||||
}),
|
}),
|
||||||
};
|
};
|
||||||
match result {
|
match result {
|
||||||
Ok(bytes) => {
|
Ok(bytes) => {
|
||||||
std::fs::write(&out, bytes)
|
std::fs::write(out, bytes)
|
||||||
.with_context(|| format!("failed to write `{}`", out.display()))?;
|
.with_context(|| format!("failed to write `{}`", out.display()))?;
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,67 @@
|
||||||
|
//! `rad lfs fetch-batch` — long-lived plumbing worker invoked by the Git
|
||||||
|
//! LFS custom transfer agent (`rad-lfs-transfer`) to download multiple LFS
|
||||||
|
//! objects across a single process, kept alive for the lifetime of one
|
||||||
|
//! `git lfs pull`/checkout session -- specifically so a private repo's
|
||||||
|
//! passphrase is only prompted for once per session, not once per file.
|
||||||
|
//!
|
||||||
|
//! Unlike `rad lfs precommit`/`backfill` (which know every file upfront
|
||||||
|
//! and can just read stdin until EOF), this can't collect a batch before
|
||||||
|
//! starting: git-lfs's custom-transfer protocol requests objects one at a
|
||||||
|
//! time by default (`rad-lfs-transfer` doesn't negotiate the "concurrent"
|
||||||
|
//! capability), so the caller only knows about the next object once the
|
||||||
|
//! previous one has been answered. This process instead stays alive and
|
||||||
|
//! answers requests one at a time as they arrive, reusing the same
|
||||||
|
//! unwrapped signer across all of them.
|
||||||
|
//!
|
||||||
|
//! Protocol (line-delimited, UTF-8, on stdin/stdout):
|
||||||
|
//! request: "<oid> <size> <out-path>\n"
|
||||||
|
//! response: a JSON object, one line per request, in the same order:
|
||||||
|
//! `{"ok":true}` or `{"ok":false,"error":"<message>"}`
|
||||||
|
//! The process exits cleanly on stdin EOF (the parent closing its write
|
||||||
|
//! end signals the transfer session is over).
|
||||||
|
|
||||||
|
use std::io::{BufRead as _, Write as _};
|
||||||
|
use std::path::PathBuf;
|
||||||
|
|
||||||
|
use anyhow::Context as _;
|
||||||
|
use serde_json::json;
|
||||||
|
|
||||||
|
use crate::terminal as term;
|
||||||
|
|
||||||
|
use super::fetch::fetch_object;
|
||||||
|
|
||||||
|
pub fn run(ctx: impl term::Context) -> anyhow::Result<()> {
|
||||||
|
let profile = ctx.profile()?;
|
||||||
|
let (repo, rid) = radicle::rad::cwd()
|
||||||
|
.context("`rad lfs fetch-batch` must be run inside a Radicle repository working copy")?;
|
||||||
|
|
||||||
|
let mut signer = None;
|
||||||
|
let stdin = std::io::stdin();
|
||||||
|
let mut stdout = std::io::stdout();
|
||||||
|
|
||||||
|
for line in stdin.lock().lines() {
|
||||||
|
let line = line.context("failed to read stdin")?;
|
||||||
|
// Each line is "<oid> <size> <out-path>" -- oid is a fixed-length
|
||||||
|
// hex string and size is decimal, so splitting on the first two
|
||||||
|
// spaces leaves the path (which may itself contain spaces) intact
|
||||||
|
// as the remainder. Mirrors `rad lfs precommit`'s input format.
|
||||||
|
let mut parts = line.splitn(3, ' ');
|
||||||
|
let (Some(oid), Some(size), Some(out)) = (parts.next(), parts.next(), parts.next())
|
||||||
|
else {
|
||||||
|
anyhow::bail!("malformed `rad lfs fetch-batch` input line: {line:?}");
|
||||||
|
};
|
||||||
|
let Ok(size) = size.parse::<i64>() else {
|
||||||
|
anyhow::bail!("invalid size in `rad lfs fetch-batch` input: {size:?}");
|
||||||
|
};
|
||||||
|
let out = PathBuf::from(out);
|
||||||
|
|
||||||
|
let response = match fetch_object(&profile, &repo, rid, oid, size, &out, &mut signer) {
|
||||||
|
Ok(()) => json!({"ok": true}),
|
||||||
|
Err(err) => json!({"ok": false, "error": format!("{err:#}")}),
|
||||||
|
};
|
||||||
|
writeln!(stdout, "{response}").context("failed to write response")?;
|
||||||
|
stdout.flush().context("failed to flush stdout")?;
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
Loading…
Reference in New Issue