holperf/src/chain.rs
//! The estate-box path: the intake chain's production door (pip), its npm door and the
//! chain views (apt, dnf, Helm, Maven, Go), over one chain holding every format's packages.
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};
use anyhow::Result;
use server_lib::audit::MemoryAuditLog;
use server_lib::chain_view::{ChainSlot, ChainView};
use server_lib::promote::intake::{
ArtifactState::*, IntakeChain, IntakeSettings, TransitionRequest,
};
use server_lib::promote::{PromotionEngine, Stage, StageLadder};
use server_lib::raw::RawBackend;
use traits::{ArtifactFormat, ArtifactId, Bytes, Caps, HttpReply, RepositoryBackendTrait};
use crate::gen;
use crate::harness::Scenario;
use crate::seed::{want, Open};
fn name(i: usize) -> String {
format!("pkg{i:05}")
}
fn picks(n: usize) -> Vec<usize> {
(0..64).map(|k| (k * 7919) % n).collect()
}
pub fn make_chain(dir: &Path) -> Arc<IntakeChain> {
let incoming: Arc<dyn RepositoryBackendTrait> =
Arc::new(RawBackend::new("incoming".into(), dir.join("incoming")));
let settings = IntakeSettings {
incoming: "incoming".into(),
quarantine: "quarantine".into(),
approved: "approved".into(),
production: "production".into(),
transitions: vec![],
inspect: Default::default(),
secrets: None,
rules: server_lib::sandbox::IntakeRules {
sandbox: server_lib::sandbox::SandboxRule::Off,
},
sandbox: None,
scan: None,
approvals: 1,
trust_policy: None,
npm: Some("npm".into()),
oci_push: Vec::new(),
unscanned: Vec::new(),
cargo: None,
bundles: None,
bundles_trust_policy: None,
};
let prod = dir.join("prod");
let ladder = StageLadder::new(
settings
.repos()
.iter()
.map(|r| Stage::new(*r, *r))
.collect(),
);
let engine = PromotionEngine::new(&prod)
.with_served_repositories(settings.repos())
.with_ladder("intake", ladder);
Arc::new(IntakeChain::new(
settings,
engine,
&prod,
Arc::new(MemoryAuditLog::new()),
incoming,
))
}
fn land(chain: &IntakeChain, dir: &Path, coord: &str, bytes: &[u8]) -> Result<()> {
let id = ArtifactId {
namespace: None,
name: coord.into(),
version: String::new(),
};
let inc = RawBackend::new("incoming".into(), dir.join("incoming"));
chain
.ingest("importer", "127.0.0.1:1", "incoming", &id, bytes, || {
inc.put(&id, bytes)
})
.map_err(|e| anyhow::anyhow!("{coord}: {}", e.tagged()))?;
for to in [
Scanned,
PolicyChecked,
Approved,
Sealed,
Promoted,
Available,
] {
let who = if to == Approved {
"approver"
} else {
"officer"
};
chain
.advance(TransitionRequest::new(id.clone(), to, who))
.map_err(|e| anyhow::anyhow!("{coord} -> {to:?}: {}", e.tagged()))?;
}
Ok(())
}
/// Every file the chain gets, per format; `m` packages x 2 versions + one large.
fn files(m: usize, small: usize, large: usize) -> Vec<(String, Vec<u8>)> {
let mut f: Vec<(String, Vec<u8>)> = Vec::new();
let vers = ["1.0.0", "1.1.0"];
for i in 0..m {
for (j, v) in vers.iter().enumerate() {
let s = (i * 2 + j) as u64;
let n = name(i);
f.push((
format!("{n}-{v}-py3-none-any.whl"),
gen::wheel(&n, v, small, s),
));
f.push((format!("{n}/-/{n}-{v}.tgz"), gen::npm_tgz(&n, v, small, s)));
let mb = format!("maven/org/bench/{n}/{v}/{n}-{v}");
f.push((format!("{mb}.jar"), gen::jar("org.bench", &n, v, small, s)));
f.push((
format!("{mb}.pom"),
gen::pom("org.bench", &n, v).into_bytes(),
));
f.push((format!("charts/{n}-{v}.tgz"), gen::chart(&n, v, small, s)));
let m_ = format!("example.com/m/{n}");
f.push((
format!("go/{m_}/@v/v{v}.zip"),
gen::go_zip(&m_, &format!("v{v}"), small, s),
));
f.push((
format!("go/{m_}/@v/v{v}.mod"),
format!("module {m_}\n").into_bytes(),
));
f.push((
format!("go/{m_}/@v/v{v}.info"),
format!("{{\"Version\":\"v{v}\",\"Time\":\"2026-01-01T00:00:00Z\"}}").into_bytes(),
));
f.push((
format!("apt/pool/main/p/{n}/{n}_{v}_amd64.deb"),
gen::deb(&n, v, "amd64", small, s),
));
f.push((
format!("yum/Packages/{n}-{v}-1.x86_64.rpm"),
gen::rpm(&format!("{n}-{v}-1"), small, s),
));
}
}
f.push((
"bigpkg-1.0.0-py3-none-any.whl".into(),
gen::wheel("bigpkg", "1.0.0", large, 7),
));
f.push((
"bigpkg/-/bigpkg-1.0.0.tgz".into(),
gen::npm_tgz("bigpkg", "1.0.0", large, 7),
));
f.push((
"maven/org/bench/bigpkg/1.0.0/bigpkg-1.0.0.jar".into(),
gen::jar("org.bench", "bigpkg", "1.0.0", large, 7),
));
f.push((
"maven/org/bench/bigpkg/1.0.0/bigpkg-1.0.0.pom".into(),
gen::pom("org.bench", "bigpkg", "1.0.0").into_bytes(),
));
f.push((
"charts/bigpkg-1.0.0.tgz".into(),
gen::chart("bigpkg", "1.0.0", large, 7),
));
f.push((
"go/example.com/m/bigpkg/@v/v1.0.0.zip".into(),
gen::go_zip("example.com/m/bigpkg", "v1.0.0", large, 7),
));
f.push((
"apt/pool/main/b/bigpkg/bigpkg_1.0.0_amd64.deb".into(),
gen::deb("bigpkg", "1.0.0", "amd64", large, 7),
));
f.push((
"yum/Packages/bigpkg-1.0.0-1.x86_64.rpm".into(),
gen::rpm("bigpkg-1.0.0-1", large, 7),
));
f
}
/// The production door (pip) as the HTTP door calls it, in process.
pub struct PipDoor(pub Arc<IntakeChain>);
/// The npm door as the HTTP door calls it, in process.
pub struct NpmDoor(pub Arc<IntakeChain>);
macro_rules! door_common {
() => {
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Raw
}
fn is_writable(&self) -> bool {
false
}
fn fetch(&self, _id: &ArtifactId) -> anyhow::Result<Option<Bytes>> {
Ok(None)
}
fn put(&self, _id: &ArtifactId, _d: &[u8]) -> anyhow::Result<()> {
anyhow::bail!("read-only")
}
fn caps(&self) -> Caps {
Caps::NONE
}
};
}
fn reply(r: Option<(&'static str, Vec<u8>)>) -> HttpReply {
match r {
Some((ct, b)) => (200, vec![("Content-Type".into(), ct.into())], b),
None => (404, vec![], b"Not found".to_vec()),
}
}
impl RepositoryBackendTrait for PipDoor {
fn name(&self) -> &str {
"production"
}
door_common!();
fn handle_http2_request(&self, _m: &str, suburl: &str, _b: &[u8]) -> anyhow::Result<HttpReply> {
let sub = suburl.strip_prefix("/production/").unwrap_or("");
if sub == "simple" || sub.starts_with("simple/") {
return Ok(reply(
self.0
.simple_index_as(sub, true)
.map(|(c, b)| (c, Vec::from(b))),
));
}
let id = ArtifactId {
namespace: None,
name: sub.into(),
version: String::new(),
};
Ok(reply(
self.0
.serve(&id)
.map_err(|e| anyhow::anyhow!(e.tagged()))?
.map(|b| ("application/octet-stream", b)),
))
}
}
impl RepositoryBackendTrait for NpmDoor {
fn name(&self) -> &str {
"npm"
}
door_common!();
fn handle_http2_request(&self, _m: &str, suburl: &str, _b: &[u8]) -> anyhow::Result<HttpReply> {
let route = server_lib::promote::npm::route(suburl.strip_prefix("/npm").unwrap_or(""));
Ok(reply(
self.0
.npm_read(&route, "http://127.0.0.1/npm")
.map_err(|e| anyhow::anyhow!(e.tagged()))?
.map(|(c, b)| (c, Vec::from(b))),
))
}
}
/// Land every `(coordinate, bytes)` into a fresh chain under `dir` (16 at a time), once:
/// `dir/seeded` holds the count and the refused coordinates, so a second run reuses it.
/// Returns the refused coordinates.
pub fn land_set(dir: &Path, all: &[(String, Vec<u8>)]) -> Result<Vec<String>> {
let marker = dir.join("seeded");
if let Ok(s) = std::fs::read_to_string(&marker) {
return Ok(s.lines().skip(1).map(str::to_string).collect());
}
let _ = std::fs::remove_dir_all(dir);
std::fs::create_dir_all(dir)?;
let chain = make_chain(dir);
let t = std::time::Instant::now();
let next = std::sync::atomic::AtomicUsize::new(0);
let errs = std::sync::Mutex::new(Vec::new());
std::thread::scope(|s| {
for _ in 0..16 {
s.spawn(|| loop {
let i = next.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let Some((c, b)) = all.get(i) else { break };
if let Err(e) = land(&chain, dir, c, b) {
errs.lock().unwrap().push((c.clone(), e.to_string()));
}
});
}
});
let errs = errs.into_inner().unwrap();
eprintln!(
"chain: landed {} files in {:.1}s, {} refused (first: {:?})",
all.len(),
t.elapsed().as_secs_f64(),
errs.len(),
errs.first()
);
let mut m = format!("{}\n", all.len());
for (c, _) in &errs {
m.push_str(c);
m.push('\n');
}
std::fs::write(&marker, m)?;
Ok(errs.into_iter().map(|(c, _)| c).collect())
}
/// The estate shape for the HTTP door: the real routes, ONE chain shared by all.
pub fn estate_routes(
dir: &Path,
chain: &Arc<IntakeChain>,
) -> Vec<(String, Arc<dyn RepositoryBackendTrait>)> {
let slot: ChainSlot = Arc::new(OnceLock::new());
let _ = slot.set(Arc::clone(chain));
let mut routes: Vec<(String, Arc<dyn RepositoryBackendTrait>)> = vec![
(
"production".into(),
Arc::new(RawBackend::new(
"production".into(),
dir.join("production-raw"),
)),
),
(
"npm".into(),
Arc::new(r_npm::NpmRepoZnippy::new("npm".into())),
),
];
for (n, t) in [
("maven", "maven3"),
("charts", "helm"),
("go", "go"),
("apt", "deb"),
("yum", "rpm"),
] {
routes.push((
n.into(),
Arc::new(ChainView::new(n, t, slot.clone()).unwrap()),
));
}
routes
}
pub struct ChainSet {
pub scenarios: Vec<Scenario>,
/// The gateway routes of the estate shape (production, npm, the views) and its chain.
pub chain: Arc<IntakeChain>,
pub routes: Vec<(String, Arc<dyn RepositoryBackendTrait>)>,
pub available: usize,
/// What the intake policy refused to land (coordinates).
pub refused: Vec<String>,
}
pub fn build(
root: &Path,
m: usize,
small: usize,
large: usize,
only: Option<&[String]>,
) -> Result<Option<ChainSet>> {
let kinds = [
"chain-pip",
"chain-npm",
"chain-maven",
"chain-helm",
"chain-go",
"chain-apt",
"chain-dnf",
];
if !kinds.iter().any(|k| want(only, k)) {
return Ok(None);
}
let dir: PathBuf = root.join(format!("chain-{m}"));
if !dir.join("seeded").exists() {
let all = files(m, small, large);
let refused = land_set(&dir, &all)?;
// HOL-RACE-1 proof: `HOLPERF_LAND_ONLY=1` lands, reports and stops; any refusal fails the run.
if std::env::var_os("HOLPERF_LAND_ONLY").is_some() {
let _ = std::fs::remove_file(dir.join("seeded"));
anyhow::ensure!(
refused.is_empty(),
"{} of {} landings refused",
refused.len(),
all.len()
);
let available = make_chain(&dir).available().len();
eprintln!("chain: {available} AVAILABLE after landing");
anyhow::ensure!(
available == all.len(),
"{available} AVAILABLE of {} landed",
all.len()
);
return Ok(None);
}
}
let chain = make_chain(&dir);
let available = chain.available().len();
eprintln!("chain: {available} AVAILABLE");
let mut v = Vec::new();
let mk = |format: &'static str,
route: &str,
open: Open,
index: Vec<String>,
meta2: Vec<String>,
sm: Vec<String>,
lg: Vec<String>| Scenario {
format,
route: route.into(),
open,
dir: dir.clone(),
index,
meta2,
small: sm,
large: lg,
small_size: small as u64,
large_size: large as u64,
packages: m,
};
let d = dir.clone();
let view = move |n: &'static str, t: &'static str| -> Open {
let d = d.clone();
Box::new(move || {
let slot: ChainSlot = Arc::new(OnceLock::new());
let _ = slot.set(make_chain(&d));
Arc::new(ChainView::new(n, t, slot).unwrap())
})
};
let p = picks(m);
if want(only, "chain-pip") {
let d = dir.clone();
v.push(mk(
"chain-pip",
"production",
Box::new(move || Arc::new(PipDoor(make_chain(&d)))),
p.iter()
.map(|&i| format!("/production/simple/{}/", name(i)))
.collect(),
vec!["/production/simple/".into()],
p.iter()
.map(|&i| format!("/production/{}-1.0.0-py3-none-any.whl", name(i)))
.collect(),
vec!["/production/bigpkg-1.0.0-py3-none-any.whl".into()],
));
}
if want(only, "chain-npm") {
let d = dir.clone();
v.push(mk(
"chain-npm",
"npm",
Box::new(move || Arc::new(NpmDoor(make_chain(&d)))),
p.iter().map(|&i| format!("/npm/{}", name(i))).collect(),
vec!["/npm/bigpkg".into()],
p.iter()
.map(|&i| format!("/npm/{0}/-/{0}-1.0.0.tgz", name(i)))
.collect(),
vec!["/npm/bigpkg/-/bigpkg-1.0.0.tgz".into()],
));
}
if want(only, "chain-maven") {
v.push(mk(
"chain-maven",
"maven",
view("maven", "maven3"),
p.iter()
.map(|&i| format!("/maven/org/bench/{}/maven-metadata.xml", name(i)))
.collect(),
p.iter()
.map(|&i| format!("/maven/org/bench/{0}/1.0.0/{0}-1.0.0.jar.sha1", name(i)))
.collect(),
p.iter()
.map(|&i| format!("/maven/org/bench/{0}/1.0.0/{0}-1.0.0.jar", name(i)))
.collect(),
vec!["/maven/org/bench/bigpkg/1.0.0/bigpkg-1.0.0.jar".into()],
));
}
if want(only, "chain-helm") {
v.push(mk(
"chain-helm",
"charts",
view("charts", "helm"),
vec!["/charts/index.yaml".into()],
vec![],
p.iter()
.map(|&i| format!("/charts/{}-1.0.0.tgz", name(i)))
.collect(),
vec!["/charts/bigpkg-1.0.0.tgz".into()],
));
}
if want(only, "chain-go") {
v.push(mk(
"chain-go",
"go",
view("go", "go"),
p.iter()
.map(|&i| format!("/go/example.com/m/{}/@v/list", name(i)))
.collect(),
p.iter()
.map(|&i| format!("/go/example.com/m/{}/@v/v1.0.0.info", name(i)))
.collect(),
p.iter()
.map(|&i| format!("/go/example.com/m/{}/@v/v1.0.0.zip", name(i)))
.collect(),
vec!["/go/example.com/m/bigpkg/@v/v1.0.0.zip".into()],
));
}
if want(only, "chain-apt") {
v.push(mk(
"chain-apt",
"apt",
view("apt", "deb"),
vec!["/apt/dists/stable/main/binary-amd64/Packages".into()],
vec!["/apt/dists/stable/Release".into()],
p.iter()
.map(|&i| format!("/apt/pool/main/p/{0}/{0}_1.0.0_amd64.deb", name(i)))
.collect(),
vec!["/apt/pool/main/b/bigpkg/bigpkg_1.0.0_amd64.deb".into()],
));
}
if want(only, "chain-dnf") {
v.push(mk(
"chain-dnf",
"yum",
view("yum", "rpm"),
vec!["/yum/repodata/primary.xml".into()],
vec!["/yum/repodata/repomd.xml".into()],
p.iter()
.map(|&i| format!("/yum/Packages/{}-1.0.0-1.x86_64.rpm", name(i)))
.collect(),
vec!["/yum/Packages/bigpkg-1.0.0-1.x86_64.rpm".into()],
));
}
let routes = estate_routes(&dir, &chain);
Ok(Some(ChainSet {
scenarios: v,
chain,
routes,
available,
refused: Vec::new(),
}))
}