fix: use streaming for large caches
commit
984f84dfix: use streaming for large caches
Assisted-by: Claude:claude-fable-5
Reviews
No reviews of this commit yet — record a verdict below.
Start a review
crates/git-effect/src/cache.rs
@@ -11,9 +11,8 @@
//! persistent filesystem, which the toolchain extraction cache leans on
//! instead.
-use std::io::Write as _;
use std::path::Path;
-use std::process::{Command, Stdio};
+use std::process::Command;
use gix_hash::ObjectId;
@@ -69,43 +68,22 @@
return Ok(());
};
- let archive = Command::new("git")
+ let mut archive = Command::new("git");
+ archive
.arg("-C")
.arg(repo)
- .args(["archive", "--format=tar", &tree.to_string()])
- .output()
- .map_err(|e| format!("could not run git archive: {e}"))?;
- if !archive.status.success() {
- return Err(format!("git archive failed for cache {name}"));
- }
-
- let mut child = Command::new("sprite")
- .args([
- "exec",
- "-s",
- sprite,
- "--",
- "sh",
- "-c",
- &format!("tar -x -C {dir}"),
- ])
- .stdin(Stdio::piped())
- .spawn()
- .map_err(|e| format!("could not run the sprite CLI: {e}"))?;
- child
- .stdin
- .take()
- .ok_or("sprite exec did not accept stdin")?
- .write_all(&archive.stdout)
- .map_err(|e| format!("could not stream the cache into the sprite: {e}"))?;
- let status = child
- .wait()
- .map_err(|e| format!("sprite exec did not complete: {e}"))?;
- if status.success() {
- Ok(())
- } else {
- Err(format!("could not restore cache {name} in the sprite"))
- }
+ .args(["archive", "--format=tar", &tree.to_string()]);
+ let mut unpack = Command::new("sprite");
+ unpack.args([
+ "exec",
+ "-s",
+ sprite,
+ "--",
+ "sh",
+ "-c",
+ &format!("tar -x -C {dir}"),
+ ]);
+ crate::stream::pipe(archive, unpack, &format!("restoring cache {name}"))
}
/// Snapshot the sandbox's [`cache_dir`] for `name` back to [`cache_ref`],
@@ -119,43 +97,22 @@
/// @relation(checks.cache)
pub fn snapshot(repo: &Path, sprite: &str, name: &str) -> Result<(), String> {
let dir = cache_dir(name);
- let archive = Command::new("sprite")
- .args([
- "exec",
- "-s",
- sprite,
- "--",
- "sh",
- "-c",
- &format!("tar -C {dir} -cf - ."),
- ])
- .output()
- .map_err(|e| format!("could not run the sprite CLI: {e}"))?;
- if !archive.status.success() {
- return Err(format!("could not archive cache {name} from the sprite"));
- }
-
let scratch = tempfile::tempdir().map_err(|e| format!("could not create temp dir: {e}"))?;
let extracted = scratch.path().join("tree");
std::fs::create_dir(&extracted).map_err(|e| format!("could not create extraction dir: {e}"))?;
- let mut child = Command::new("tar")
- .args(["-x", "-C"])
- .arg(&extracted)
- .stdin(Stdio::piped())
- .spawn()
- .map_err(|e| format!("could not run tar: {e}"))?;
- child
- .stdin
- .take()
- .ok_or("tar did not accept stdin")?
- .write_all(&archive.stdout)
- .map_err(|e| format!("could not extract the cache archive: {e}"))?;
- let status = child
- .wait()
- .map_err(|e| format!("tar did not complete: {e}"))?;
- if !status.success() {
- return Err(format!("could not extract cache {name}'s archive"));
- }
+ let mut archive = Command::new("sprite");
+ archive.args([
+ "exec",
+ "-s",
+ sprite,
+ "--",
+ "sh",
+ "-c",
+ &format!("tar -C {dir} -cf - ."),
+ ]);
+ let mut extract = Command::new("tar");
+ extract.args(["-x", "-C"]).arg(&extracted);
+ crate::stream::pipe(archive, extract, &format!("snapshotting cache {name}"))?;
// A scratch index and an explicit work tree, so this builds a tree from
// the extracted directory without disturbing the repository's own
@@ -214,36 +171,14 @@
return Ok(());
};
- let archive = Command::new("git")
+ let mut archive = Command::new("git");
+ archive
.arg("-C")
.arg(repo)
- .args(["archive", "--format=tar", &tree.to_string()])
- .output()
- .map_err(|e| format!("could not run git archive: {e}"))?;
- if !archive.status.success() {
- return Err(format!("git archive failed for cache {name}"));
- }
-
- let mut child = Command::new("tar")
- .args(["-x", "-C"])
- .arg(dest)
- .stdin(Stdio::piped())
- .spawn()
- .map_err(|e| format!("could not run tar: {e}"))?;
- child
- .stdin
- .take()
- .ok_or("tar did not accept stdin")?
- .write_all(&archive.stdout)
- .map_err(|e| format!("could not extract cache {name}: {e}"))?;
- let status = child
- .wait()
- .map_err(|e| format!("tar did not complete: {e}"))?;
- if status.success() {
- Ok(())
- } else {
- Err(format!("could not restore cache {name}"))
- }
+ .args(["archive", "--format=tar", &tree.to_string()]);
+ let mut extract = Command::new("tar");
+ extract.args(["-x", "-C"]).arg(dest);
+ crate::stream::pipe(archive, extract, &format!("restoring cache {name}"))
}
/// [`snapshot`]'s local-backend equivalent: `src` is already a host
crates/git-effect/src/engine.rs
@@ -23,7 +23,7 @@
use std::collections::HashMap;
use std::collections::HashSet;
-use std::io::{Read, Write};
+use std::io::Read;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::mpsc::RecvTimeoutError;
@@ -817,36 +817,15 @@
///
/// @relation(checks.sandbox, compat.sprite, compat.git)
fn sync_tree(repo: &Path, sprite: &str, new: ObjectId) -> Result<(), String> {
- let archive = Command::new("git")
+ let mut archive = Command::new("git");
+ archive
.arg("-C")
.arg(repo)
- .args(["archive", "--format=tar", &new.to_string()])
- .output()
- .map_err(|e| format!("could not run git archive: {e}"))?;
- if !archive.status.success() {
- return Err(format!("git archive failed for {new}"));
- }
-
+ .args(["archive", "--format=tar", &new.to_string()]);
let script = unpack_script();
- let mut child = Command::new("sprite")
- .args(["exec", "-s", sprite, "--", "sh", "-c", &script])
- .stdin(Stdio::piped())
- .spawn()
- .map_err(|e| format!("could not run the sprite CLI: {e}"))?;
- child
- .stdin
- .take()
- .ok_or("sprite exec did not accept stdin")?
- .write_all(&archive.stdout)
- .map_err(|e| format!("could not stream the tree into the sprite: {e}"))?;
- let status = child
- .wait()
- .map_err(|e| format!("sprite exec did not complete: {e}"))?;
- if status.success() {
- Ok(())
- } else {
- Err("could not unpack the tree in the sprite".to_owned())
- }
+ let mut unpack = Command::new("sprite");
+ unpack.args(["exec", "-s", sprite, "--", "sh", "-c", &script]);
+ crate::stream::pipe(archive, unpack, &format!("syncing the tree at {new}"))
}
/// The in-sprite script that replaces [`WORKDIR`]'s contents with the tar
@@ -1012,39 +991,18 @@
return Ok(());
}
- let archive = Command::new("git")
+ let mut archive = Command::new("git");
+ archive
.arg("-C")
.arg(repo)
- .args(["archive", "--format=tar", &tree.to_string()])
- .output()
- .map_err(|e| format!("could not run git archive: {e}"))?;
- if !archive.status.success() {
- return Err(format!("git archive failed for toolchain {tree}"));
- }
-
+ .args(["archive", "--format=tar", &tree.to_string()]);
let tmp = format!("{dir}.tmp");
let script = format!(
"rm -rf {tmp} && mkdir -p {tmp} && tar -x -C {tmp} && rm -rf {dir} && mv {tmp} {dir}"
);
- let mut child = Command::new("sprite")
- .args(["exec", "-s", sprite, "--", "sh", "-c", &script])
- .stdin(Stdio::piped())
- .spawn()
- .map_err(|e| format!("could not run the sprite CLI: {e}"))?;
- child
- .stdin
- .take()
- .ok_or("sprite exec did not accept stdin")?
- .write_all(&archive.stdout)
- .map_err(|e| format!("could not stream the toolchain into the sprite: {e}"))?;
- let status = child
- .wait()
- .map_err(|e| format!("sprite exec did not complete: {e}"))?;
- if status.success() {
- Ok(())
- } else {
- Err(format!("could not extract toolchain {tree} in the sprite"))
- }
+ let mut unpack = Command::new("sprite");
+ unpack.args(["exec", "-s", sprite, "--", "sh", "-c", &script]);
+ crate::stream::pipe(archive, unpack, &format!("syncing toolchain {tree}"))
}
/// Fetch, sha256-verify, and extract a [`git_toolchain::Bin::Downloaded`]
crates/git-effect/src/lib.rs
@@ -23,6 +23,7 @@
pub mod executor;
pub mod local;
pub mod results;
+mod stream;
#[cfg(test)]
mod testutil;
crates/git-effect/src/local.rs
@@ -14,9 +14,7 @@
//! materialized bytes are identical no matter which backend runs it.
use std::collections::HashMap;
-use std::io::Write as _;
use std::path::{Path, PathBuf};
-use std::process::Stdio;
use gix_hash::ObjectId;
use std::process::Command;
@@ -69,36 +67,14 @@
/// `git archive | tar -x` straight onto the host filesystem — no sandbox CLI
/// involved, unlike the Sprite path's streamed unpack.
pub fn sync_tree(repo: &Path, sandbox: &Sandbox, new: ObjectId) -> Result<(), String> {
- let archive = Command::new("git")
+ let mut archive = Command::new("git");
+ archive
.arg("-C")
.arg(repo)
- .args(["archive", "--format=tar", &new.to_string()])
- .output()
- .map_err(|e| format!("could not run git archive: {e}"))?;
- if !archive.status.success() {
- return Err(format!("git archive failed for {new}"));
- }
- let work = sandbox.work_dir();
- let mut child = Command::new("tar")
- .args(["-x", "-C"])
- .arg(&work)
- .stdin(Stdio::piped())
- .spawn()
- .map_err(|e| format!("could not run tar: {e}"))?;
- child
- .stdin
- .take()
- .ok_or("tar did not accept stdin")?
- .write_all(&archive.stdout)
- .map_err(|e| format!("could not extract the tree: {e}"))?;
- let status = child
- .wait()
- .map_err(|e| format!("tar did not complete: {e}"))?;
- if status.success() {
- Ok(())
- } else {
- Err(format!("could not unpack the tree at {new}"))
- }
+ .args(["archive", "--format=tar", &new.to_string()]);
+ let mut extract = Command::new("tar");
+ extract.args(["-x", "-C"]).arg(sandbox.work_dir());
+ crate::stream::pipe(archive, extract, &format!("syncing the tree at {new}"))
}
/// Resolve and extract every distinct toolchain named across `runnable` into
crates/git-effect/src/stream.rs
@@ -1,0 +1,41 @@
+//! Kernel-side piping between two child processes.
+
+use std::process::{Command, Stdio};
+
+/// Run `producer | consumer` with the pipe handed to the kernel: the
+/// producer's stdout *is* the consumer's stdin, so the stream — a cache or
+/// tree archive that can outgrow the machine's memory — never lands in this
+/// process. Both exit statuses gate success: a producer that dies mid-stream
+/// fails the pipe even when the consumer accepted the truncated input.
+pub(crate) fn pipe(mut producer: Command, mut consumer: Command, what: &str) -> Result<(), String> {
+ let producer_name = producer.get_program().to_string_lossy().into_owned();
+ let consumer_name = consumer.get_program().to_string_lossy().into_owned();
+ let mut producing = producer
+ .stdin(Stdio::null())
+ .stdout(Stdio::piped())
+ .spawn()
+ .map_err(|e| format!("could not run {producer_name} for {what}: {e}"))?;
+ let Some(stdout) = producing.stdout.take() else {
+ let _killed = producing.kill();
+ let _reaped = producing.wait();
+ return Err(format!("{producer_name} gave no stdout for {what}"));
+ };
+ let consumed = consumer.stdin(Stdio::from(stdout)).status();
+ // Release this process's copy of the pipe's read end before waiting on
+ // the producer: with the consumer dead mid-stream, the producer only
+ // sees EPIPE — and stops blocking on a full pipe — once no read end is
+ // left open.
+ drop(consumer);
+ let produced = producing
+ .wait()
+ .map_err(|e| format!("{producer_name} did not complete for {what}: {e}"))?;
+ let consumed =
+ consumed.map_err(|e| format!("could not run {consumer_name} for {what}: {e}"))?;
+ if !produced.success() {
+ return Err(format!("{producer_name} failed for {what}: {produced}"));
+ }
+ if !consumed.success() {
+ return Err(format!("{consumer_name} failed for {what}: {consumed}"));
+ }
+ Ok(())
+}