Skip to content

fix(proof-harvest): no proxy tar leak on missing holdout shard #237

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cursor merged 2 commits into from
Sep 8, 2026
Merged

fix(proof-harvest): no proxy tar leak on missing holdout shard #237

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d8eae35
fix(proof-harvest): no proxy tar leak on missing holdout shard
cursoragent Sep 8, 2026
4873539
fix(proof-harvest): case-insensitive leftover tar extension check
cursoragent Sep 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 74 additions & 20 deletions crates/proof-harvest/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
use std::io::Read;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use async_trait::async_trait;
Expand Down Expand Up @@ -263,13 +264,24 @@ pub fn pack_holdout_tar(store: &Path, holdout: &[HoldoutRecord]) -> Result<Vec<u
/// The archive is written to disk so live score can stream it to SSH
/// instead of buffering the model in control-plane memory.
pub fn pack_proxy_tar(dir: &Path) -> Result<PathBuf, EvalError> {
pack_proxy_tar_to(dir, &std::env::temp_dir())
}

fn next_proxy_tar_path(dest_dir: &Path) -> PathBuf {
static SEQ: AtomicU64 = AtomicU64::new(0);
let seq = SEQ.fetch_add(1, Ordering::Relaxed);
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |d| d.as_nanos());
dest_dir.join(format!(
"proof-proxy-{}-{nanos}-{seq}.tar",
std::process::id()
))
}

fn pack_proxy_tar_to(dir: &Path, dest_dir: &Path) -> Result<PathBuf, EvalError> {
proxy_archive_usable(dir)?;
let dest = {
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |d| d.as_nanos());
std::env::temp_dir().join(format!("proof-proxy-{}-{nanos}.tar", std::process::id()))
};
let dest = next_proxy_tar_path(dest_dir);
let status = Command::new("tar")
.arg("-C")
.arg(dir)
Expand Down Expand Up @@ -336,6 +348,11 @@ pub struct LiumProofHarvest {
proxy_model_dir: Option<PathBuf>,
/// Host directory of holdout shard files named `<content_sha256>`.
holdout_store: Option<PathBuf>,
/// Host directory that receives the staged `proof-proxy-*.tar`.
///
/// Defaults to the process temp dir. Tests override this so leftover
/// scans do not race sibling unit tests that share `/tmp`.
proxy_tar_dir: PathBuf,
}

impl LiumProofHarvest {
Expand All @@ -349,6 +366,7 @@ impl LiumProofHarvest {
judge_api_key: None,
proxy_model_dir: None,
holdout_store: None,
proxy_tar_dir: std::env::temp_dir(),
}
}

Expand Down Expand Up @@ -378,6 +396,16 @@ impl LiumProofHarvest {
self
}

/// Host directory for the staged proxy tar (`proof-proxy-*.tar`).
#[cfg(test)]
#[must_use]
fn with_proxy_tar_dir(mut self, dir: PathBuf) -> Self {
if !dir.as_os_str().is_empty() {
self.proxy_tar_dir = dir;
}
self
}

fn live_extras(
&self,
holdout: &[HoldoutRecord],
Expand All @@ -393,7 +421,7 @@ impl LiumProofHarvest {
// Holdout first: a missing/mismatched shard must not leave a
// model-sized proxy tar on the control plane.
let holdout_tar = pack_holdout_tar(store, holdout)?;
let proxy_path = pack_proxy_tar(proxy)?;
let proxy_path = pack_proxy_tar_to(proxy, &self.proxy_tar_dir)?;
let guard = ProxyTarGuard(Some(proxy_path.clone()));
Ok((
RunExtras {
Expand Down Expand Up @@ -919,9 +947,8 @@ mod tests {
assert!(!*pod.booted.lock().expect("booted"));
}

fn leftover_proxy_tars() -> Vec<PathBuf> {
let prefix = format!("proof-proxy-{}-", std::process::id());
let Ok(entries) = std::fs::read_dir(std::env::temp_dir()) else {
fn leftover_proxy_tars(dir: &Path) -> Vec<PathBuf> {
let Ok(entries) = std::fs::read_dir(dir) else {
return Vec::new();
};
let mut paths: Vec<PathBuf> = entries
Expand All @@ -930,7 +957,7 @@ mod tests {
.filter(|p| {
p.file_name()
.and_then(|n| n.to_str())
.is_some_and(|n| n.starts_with(&prefix))
.is_some_and(|n| n.starts_with("proof-proxy-"))
&& p.extension()
.is_some_and(|ext| ext.eq_ignore_ascii_case("tar"))
})
Expand All @@ -939,23 +966,31 @@ mod tests {
paths
}

fn proxy_tar_staging(proxy: &Path) -> PathBuf {
let dir = proxy.parent().expect("root").join("proxy-tars");
std::fs::create_dir_all(&dir).expect("staging");
dir
}

fn assert_live_extras_holdout_refuse_leaves_no_proxy_tar(
harvest: &LiumProofHarvest,
holdout: &[HoldoutRecord],
staging: &Path,
) {
let before = leftover_proxy_tars();
let before = leftover_proxy_tars(staging);
match harvest.live_extras(holdout) {
Ok(_) => panic!("holdout refuse"),
Err(err) => assert!(matches!(err, EvalError::HoldoutStoreMissing), "{err}"),
}
let after = leftover_proxy_tars();
let after = leftover_proxy_tars(staging);
assert_eq!(after, before, "proxy tar leaked: {after:?}");
}

#[tokio::test]
async fn missing_holdout_shard_does_not_leak_a_proxy_tar() {
let recs = synthetic_holdout(STRATUM_SIZE, 1);
let (proxy, store) = live_asset_dirs(&recs);
let staging = proxy_tar_staging(&proxy);
let pod = CapturePod::new();
let harvest = LiumProofHarvest::new(
pod.clone(),
Expand All @@ -964,11 +999,12 @@ mod tests {
)
.with_judge_api_key(Some("sk-live-not-a-real-secret".into()))
.with_proxy_model_dir(Some(proxy))
.with_holdout_store(Some(store));
.with_holdout_store(Some(store))
.with_proxy_tar_dir(staging.clone());
harvest.ready().expect("store is usable");
let mut missing = recs.clone();
missing[0].content_sha256 = "ab".repeat(32);
let before = leftover_proxy_tars();
let before = leftover_proxy_tars(&staging);
let err = harvest
.score(
&harvest_pin(),
Expand All @@ -983,26 +1019,44 @@ mod tests {
.expect_err("missing shard");
assert!(matches!(err, EvalError::HoldoutStoreMissing), "{err}");
assert!(!*pod.booted.lock().expect("booted"));
let after = leftover_proxy_tars();
let after = leftover_proxy_tars(&staging);
assert_eq!(after, before, "proxy tar leaked: {after:?}");
}

#[test]
fn live_extras_holdout_refuse_does_not_leak_a_proxy_tar() {
let recs = synthetic_holdout(STRATUM_SIZE, 1);
let (proxy, store) = live_asset_dirs(&recs);
let harvest = harvest_for_ready(proxy.clone(), store.clone());
let staging = proxy_tar_staging(&proxy);
let harvest =
harvest_for_ready(proxy.clone(), store.clone()).with_proxy_tar_dir(staging.clone());
let mut missing = recs.clone();
missing[0].content_sha256 = "ab".repeat(32);
assert_live_extras_holdout_refuse_leaves_no_proxy_tar(&harvest, &missing);
assert_live_extras_holdout_refuse_leaves_no_proxy_tar(&harvest, &missing, &staging);

std::fs::write(
store.join(recs[0].content_sha256.to_ascii_lowercase()),
b"not-the-catalogued-shard\n",
)
.expect("tamper");
let harvest = harvest_for_ready(proxy, store);
assert_live_extras_holdout_refuse_leaves_no_proxy_tar(&harvest, &recs);
let harvest = harvest_for_ready(proxy, store).with_proxy_tar_dir(staging.clone());
assert_live_extras_holdout_refuse_leaves_no_proxy_tar(&harvest, &recs, &staging);
}

#[test]
fn live_extras_guard_deletes_the_proxy_tar_in_the_configured_dir() {
let recs = synthetic_holdout(STRATUM_SIZE, 1);
let (proxy, store) = live_asset_dirs(&recs);
let staging = proxy_tar_staging(&proxy);
let harvest = harvest_for_ready(proxy, store).with_proxy_tar_dir(staging.clone());
{
let (extras, _guard) = harvest.live_extras(&recs).expect("extras");
let path = extras.proxy_tar_path.expect("path");
assert!(path.starts_with(&staging), "{path:?}");
assert!(path.is_file(), "{path:?}");
assert_eq!(leftover_proxy_tars(&staging).len(), 1);
}
assert_eq!(leftover_proxy_tars(&staging), Vec::<PathBuf>::new());
}

#[test]
Expand Down
Loading