server/lib/src/pull_intake/crates.rs
//! **A cargo sparse index as a remote source** (`kind: crates`, with `url`):
//! crates.io (`https://index.crates.io/`), any registry that speaks the sparse
//! protocol (RFC 2789).
//!
//! A `.crate` is taken only when its bytes hash to the `cksum` (sha256) of its
//! index line. crates.io publishes no signature or provenance for a crate, so a
//! floor that requires one (`rules: (require_signed: true)`) refuses every crate
//! by name; that is the floor doing its job, not a gap of the pull.
//!
//! Holger's own index for the source lists only the AVAILABLE versions, each with
//! the upstream's line as it was when the crate was pulled (its dependencies and
//! features are the registry's, its `cksum` is the sealed bytes' sha256).
use super::http::Fetch;
use crate::lock::LockOrRecover as _;
use serde_json::Value;
/// The cap on one index file.
const MAX_INDEX: u64 = 64 * 1024 * 1024;
// The names, index paths and coordinates are the cargo door's (lane HOL-CARGO-PUBLISH-DOOR):
// a pulled crate stands at `<source>/<name>/<name>-<version>.crate`, a published one at
// `<door>/<name>/<name>-<version>.crate`.
pub use crate::promote::cargo::{coords_of, id, index_path, name_of_index_path, valid_name};
/// One line of an index file.
#[derive(Debug, Clone, PartialEq)]
pub struct IndexLine {
pub name: String,
pub vers: String,
pub cksum: String,
pub yanked: bool,
/// The line as the registry wrote it.
pub raw: Value,
}
/// One crate's index file as every walk shares it: one copy, behind an `Arc`.
pub type Lines = std::sync::Arc<[IndexLine]>;
/// How long a read index file is shared before the upstream is asked again.
pub const SHARED_INDEX_TTL: std::time::Duration = std::time::Duration::from_secs(60);
/// The most index files kept at once; past it the oldest are dropped.
pub const SHARED_INDEX_CAP: usize = 16_384;
enum Slot {
/// Read at `at`; `None` = the registry does not publish the crate.
Ready {
lines: Option<Lines>,
at: std::time::Instant,
},
/// One caller reads it now; the others wait on the flag.
Reading(std::sync::Arc<(std::sync::Mutex<bool>, std::sync::Condvar)>),
}
/// **The ONE index of the crates sources, shared by every pull and closure walk**
/// (HOL-DOGFOOD-8). Before it, each walk read every dependency's sparse-index file from the
/// upstream itself and held its own copy (all versions, raw JSON, cloned per edge): about a
/// thousand walks at once when `cargo fetch` of a big workspace came through, and the
/// dogfood Holger reached 40 GB in 30 s. Now a file is read once per
/// [`SHARED_INDEX_TTL`] (single flight: concurrent askers wait for the one read), held once,
/// and the map is bounded by [`SHARED_INDEX_CAP`] files.
pub struct SharedIndex {
slots: std::sync::Mutex<std::collections::HashMap<(String, String), Slot>>,
ttl: std::time::Duration,
cap: usize,
reads: std::sync::atomic::AtomicU64,
}
impl Default for SharedIndex {
fn default() -> Self {
Self::new(SHARED_INDEX_TTL, SHARED_INDEX_CAP)
}
}
impl SharedIndex {
pub fn new(ttl: std::time::Duration, cap: usize) -> Self {
Self {
slots: std::sync::Mutex::new(std::collections::HashMap::new()),
ttl,
cap: cap.max(1),
reads: std::sync::atomic::AtomicU64::new(0),
}
}
/// How many upstream reads this index made.
pub fn reads(&self) -> u64 {
self.reads.load(std::sync::atomic::Ordering::Relaxed)
}
/// How many index files it holds.
pub fn len(&self) -> usize {
self.slots().len()
}
/// Whether it holds none.
pub fn is_empty(&self) -> bool {
self.len() == 0
}
fn slots(
&self,
) -> std::sync::MutexGuard<'_, std::collections::HashMap<(String, String), Slot>> {
self.slots.lock_or_recover()
}
/// **`name`'s index file from `source`**: the shared copy while it is younger than the
/// time to live, else read by `read` (once, however many ask at the same time).
/// `(lines, true)` when this call read it from the upstream: the caller records the
/// listing then, once per read. A failed read is not kept.
pub fn get(
&self,
source: &str,
name: &str,
read: impl FnOnce() -> anyhow::Result<Option<Vec<IndexLine>>>,
) -> anyhow::Result<(Option<Lines>, bool)> {
let key = (source.to_string(), name.to_ascii_lowercase());
let mut read = Some(read);
loop {
let wait = {
let mut m = self.slots();
match m.get(&key) {
Some(Slot::Ready { lines, at }) if at.elapsed() < self.ttl => {
return Ok((lines.clone(), false));
}
Some(Slot::Reading(w)) => std::sync::Arc::clone(w),
_ => {
let w = std::sync::Arc::new((
std::sync::Mutex::new(false),
std::sync::Condvar::new(),
));
m.insert(key.clone(), Slot::Reading(std::sync::Arc::clone(&w)));
drop(m);
let Some(read) = read.take() else {
anyhow::bail!("the shared index read '{name}' twice in one call");
};
// A read that panics must not leave the others waiting for good.
let got = std::panic::catch_unwind(std::panic::AssertUnwindSafe(read));
self.reads
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let mut m = self.slots();
let out = match got {
Err(panic) => {
m.remove(&key);
drop(m);
let (done, cv) = &*w;
*done.lock_or_recover() = true;
cv.notify_all();
std::panic::resume_unwind(panic);
}
Ok(got) => got,
};
let out = match out {
Ok(l) => {
let lines: Option<Lines> = l.map(Lines::from);
m.insert(
key.clone(),
Slot::Ready {
lines: lines.clone(),
at: std::time::Instant::now(),
},
);
self.bound(&mut m);
Ok((lines, true))
}
Err(e) => {
m.remove(&key);
Err(e)
}
};
drop(m);
let (done, cv) = &*w;
*done.lock_or_recover() = true;
cv.notify_all();
return out;
}
}
};
let (done, cv) = &*wait;
let mut d = done.lock_or_recover();
while !*d {
d = crate::lock::recover(cv.wait(d));
}
}
}
/// Past the cap: drop what expired, then the oldest.
fn bound(&self, m: &mut std::collections::HashMap<(String, String), Slot>) {
if m.len() <= self.cap {
return;
}
let ttl = self.ttl;
m.retain(|_, s| !matches!(s, Slot::Ready { at, .. } if at.elapsed() >= ttl));
while m.len() > self.cap {
let oldest = m
.iter()
.filter_map(|(k, s)| match s {
Slot::Ready { at, .. } => Some((*at, k.clone())),
Slot::Reading(_) => None,
})
.min_by_key(|(at, _)| *at)
.map(|(_, k)| k);
match oldest {
Some(k) => {
m.remove(&k);
}
None => break,
}
}
}
}
fn index_base(url: &str) -> String {
let u = url.strip_prefix("sparse+").unwrap_or(url);
if u.ends_with('/') {
u.to_string()
} else {
format!("{u}/")
}
}
/// **Read one crate's index file.** `None` when the registry does not know it.
pub fn index(
http: &dyn Fetch,
url: &str,
name: &str,
auth: Option<(&str, &str)>,
) -> anyhow::Result<Option<Vec<IndexLine>>> {
if !valid_name(name) {
anyhow::bail!("'{name}' is not a crate name");
}
let url = format!("{}{}", index_base(url), index_path(name));
let r = http.get(&url, &[], auth, MAX_INDEX)?;
match r.status {
200 => {}
404 | 410 | 451 => return Ok(None),
s => anyhow::bail!("the index {url} answered HTTP {s}"),
}
let mut out = Vec::new();
for l in String::from_utf8_lossy(&r.body)
.lines()
.filter(|l| !l.trim().is_empty())
{
let Ok(raw) = serde_json::from_str::<Value>(l) else {
continue;
};
let s = |k: &str| raw.get(k).and_then(Value::as_str).unwrap_or("").to_string();
let line = IndexLine {
name: s("name"),
vers: s("vers"),
cksum: s("cksum").to_ascii_lowercase(),
yanked: raw.get("yanked").and_then(Value::as_bool).unwrap_or(false),
raw,
};
if line.name.eq_ignore_ascii_case(name) && !line.vers.is_empty() && line.cksum.len() == 64 {
out.push(line);
}
}
Ok(Some(out))
}
/// The newest release that is not yanked (a pre-release only when nothing else is).
pub fn newest(lines: &[IndexLine]) -> Option<&IndexLine> {
let cmp =
|a: &&IndexLine, b: &&IndexLine| crate::promote::npm::compare_versions(&a.vers, &b.vers);
let live = lines.iter().filter(|l| !l.yanked);
live.clone()
.filter(|l| !l.vers.contains('-'))
.max_by(cmp)
.or_else(|| live.max_by(cmp))
}
/// The download URL of one version: the registry's `config.json` `dl` template
/// (`{crate}`, `{version}`, `{prefix}`, `{lowerprefix}`, `{sha256-checksum}`), or
/// `<dl>/<crate>/<version>/download` when it names none.
pub fn download_url(dl: &str, line: &IndexLine) -> String {
let markers = [
"{crate}",
"{version}",
"{prefix}",
"{lowerprefix}",
"{sha256-checksum}",
];
if !markers.iter().any(|m| dl.contains(m)) {
return format!(
"{}/{}/{}/download",
dl.trim_end_matches('/'),
line.name,
line.vers
);
}
let prefix = {
let p = index_path(&line.name);
p.rsplit_once('/')
.map(|(d, _)| d.to_string())
.unwrap_or_default()
};
let raw_prefix = match line.name.len() {
1 | 2 => prefix.clone(),
3 => format!("3/{}", &line.name[..1]),
_ => format!("{}/{}", &line.name[..2], &line.name[2..4]),
};
dl.replace("{crate}", &line.name)
.replace("{version}", &line.vers)
.replace("{lowerprefix}", &prefix)
.replace("{prefix}", &raw_prefix)
.replace("{sha256-checksum}", &line.cksum)
}
/// **Fetch one crate**, held against its index line's `cksum`.
pub fn fetch_crate(
http: &dyn Fetch,
url: &str,
line: &IndexLine,
auth: Option<(&str, &str)>,
cap: u64,
) -> anyhow::Result<Vec<u8>> {
let cfg_url = format!("{}config.json", index_base(url));
let c = http.get(&cfg_url, &[], auth, MAX_INDEX)?;
if c.status != 200 {
anyhow::bail!("the index config {cfg_url} answered HTTP {}", c.status);
}
let cfg: Value = serde_json::from_slice(&c.body)
.map_err(|e| anyhow::anyhow!("the index config {cfg_url} is not JSON: {e}"))?;
let dl = cfg
.get("dl")
.and_then(Value::as_str)
.ok_or_else(|| anyhow::anyhow!("the index config {cfg_url} names no `dl`"))?;
let url = download_url(dl, line);
let r = http.get(&url, &[], auth, cap)?;
if r.status != 200 {
anyhow::bail!(
"{}-{}.crate: {url} answered HTTP {}",
line.name,
line.vers,
r.status
);
}
let got = crate::attest::ArtifactDigest::of_sha256(&r.body)
.hex()
.to_string();
if got != line.cksum {
anyhow::bail!(
"{}-{}.crate: the bytes hash to sha256 {got}, the index publishes {}; not taken",
line.name,
line.vers,
line.cksum
);
}
Ok(r.body)
}
/// **A crate's index line from its own sealed `.crate`** (a downstream's view, H3): the
/// normalised `Cargo.toml` that `cargo package` wrote at `<name>-<version>/Cargo.toml`, as the
/// sparse index states it, and `cksum` the sealed bytes' sha256. The read is the scan worker's
/// row read ([`holger_index_rows::crates`]), bounded, from a verified artifact.
pub fn line_from_crate(bytes: &[u8], name: &str, version: &str) -> anyhow::Result<String> {
holger_index_rows::crates::line_from_crate(bytes, name, version).map_err(anyhow::Error::msg)
}
/// **The `config.json` Holger serves** for a crates source: downloads come from the
/// source's own path below the production repository. `auth_required` (reads are gated):
/// `"auth-required": true`, which is what makes cargo send its token for the index files and
/// the `.crate`s, not only for `config.json` (HOL-DOCFIND b).
pub fn config_json(base: &str, source: &str, auth_required: bool) -> Vec<u8> {
let mut doc = serde_json::json!({ "dl": format!("{}/{source}/{{crate}}/{{crate}}-{{version}}.crate", base.trim_end_matches('/')) });
if auth_required {
doc["auth-required"] = serde_json::json!(true);
}
doc.to_string().into_bytes()
}
#[cfg(test)]
mod shared_index_tests {
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
fn lines(name: &str) -> Vec<IndexLine> {
(0..200)
.map(|v| IndexLine {
name: name.into(),
vers: format!("1.0.{v}"),
cksum: "0".repeat(64),
yanked: false,
raw: serde_json::json!({ "name": name, "vers": format!("1.0.{v}"), "deps": [] }),
})
.collect()
}
/// ★ **HOL-DOGFOOD-8: 64 closure walks at once read each index file ONCE and share one
/// copy.** Before, every walk fetched every dependency's index itself and held its own
/// copy (cloned again per edge): 64 walks over the same 12 crates were 768 upstream reads
/// and 768 copies; now 12 reads, and every walk holds the same `Arc`.
#[test]
fn concurrent_walks_share_one_read_and_one_copy_per_index_file() {
let index = SharedIndex::new(Duration::from_secs(60), 1024);
let fetches = AtomicUsize::new(0);
let names: Vec<String> = (0..12).map(|i| format!("crate-{i}")).collect();
let held: Vec<Vec<Lines>> = std::thread::scope(|s| {
let walks: Vec<_> = (0..64)
.map(|_| {
s.spawn(|| {
names
.iter()
.map(|n| {
let (l, _) = index
.get("crates-io", n, || {
fetches.fetch_add(1, Ordering::SeqCst);
// A slow upstream: the other walks wait for this read.
std::thread::sleep(Duration::from_millis(20));
Ok(Some(lines(n)))
})
.unwrap();
l.expect("published")
})
.collect()
})
})
.collect();
walks.into_iter().map(|h| h.join().unwrap()).collect()
});
assert_eq!(
fetches.load(Ordering::SeqCst),
names.len(),
"one upstream read per index file, whatever the number of walks"
);
for (i, _) in names.iter().enumerate() {
let first = &held[0][i];
assert!(
held.iter().all(|w| Arc::ptr_eq(&w[i], first)),
"every walk holds the same copy"
);
}
assert_eq!(index.reads(), names.len() as u64);
}
/// The first read says it is fresh (the caller records the listing once); a read past
/// the time to live goes to the upstream again; a failed read is not kept.
#[test]
fn a_read_is_fresh_once_expires_and_a_failure_is_not_kept() {
let index = SharedIndex::new(Duration::from_secs(60), 1024);
let (_, fresh) = index
.get("crates-io", "Serde", || Ok(Some(lines("serde"))))
.unwrap();
assert!(fresh);
let (_, fresh) = index
.get("crates-io", "serde", || panic!("cached"))
.unwrap();
assert!(!fresh, "names are case-insensitive, as cargo's are");
let (gone, fresh) = index.get("crates-io", "nope", || Ok(None)).unwrap();
assert!(
gone.is_none() && fresh,
"an unknown crate is remembered as unknown"
);
let short = SharedIndex::new(Duration::ZERO, 1024);
short.get("s", "a", || Ok(Some(lines("a")))).unwrap();
let (_, fresh) = short.get("s", "a", || Ok(Some(lines("a")))).unwrap();
assert!(fresh, "past its time to live the index is read again");
assert!(index
.get("crates-io", "flaky", || anyhow::bail!("HTTP 503"))
.is_err());
let (l, fresh) = index
.get("crates-io", "flaky", || Ok(Some(lines("flaky"))))
.unwrap();
assert!(l.is_some() && fresh, "a failed read is read again at once");
}
/// The cache is bounded: past its cap the oldest index files are dropped.
#[test]
fn the_shared_index_is_bounded() {
let index = SharedIndex::new(Duration::from_secs(60), 8);
for i in 0..100 {
index
.get("crates-io", &format!("c{i}"), || Ok(Some(lines("c"))))
.unwrap();
}
assert!(index.len() <= 8, "{} index files kept", index.len());
}
}
#[cfg(test)]
mod tests {
use super::*;
/// A `.crate` as `cargo package` writes one: a gzipped tar of `<name>-<version>/…`.
pub(crate) fn dot_crate(name: &str, version: &str, manifest: &str) -> Vec<u8> {
let mut tar = tar::Builder::new(Vec::new());
for (path, body) in [
("Cargo.toml", manifest.as_bytes()),
("src/lib.rs", b"pub fn hello() {}\n".as_slice()),
] {
let mut h = tar::Header::new_gnu();
h.set_size(body.len() as u64);
h.set_mode(0o644);
h.set_cksum();
tar.append_data(&mut h, format!("{name}-{version}/{path}"), body)
.unwrap();
}
let raw = tar.into_inner().unwrap();
let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
std::io::Write::write_all(&mut gz, &raw).unwrap();
gz.finish().unwrap()
}
/// ★ **A serve-only source's index line comes from the sealed `.crate`** (H3): the
/// dependencies of every kind and target, a renamed one with `package`, the
/// features (`v: 2` for `dep:`), and `cksum` the bytes' sha256.
#[test]
fn a_crates_index_line_is_read_from_its_own_manifest() {
let manifest = r#"
[package]
name = "demo-crate"
version = "0.1.0"
rust-version = "1.70"
[dependencies]
serde = { version = "1.0", features = ["derive"], optional = true }
log = "0.4"
[dependencies.myrand]
version = "0.8"
package = "rand"
default-features = false
[dev-dependencies]
tempfile = "3"
[target.'cfg(unix)'.dependencies]
libc = "0.2"
[features]
default = ["log-only"]
log-only = []
json = ["dep:serde"]
"#;
let bytes = dot_crate("demo-crate", "0.1.0", manifest);
let line: Value =
serde_json::from_str(&line_from_crate(&bytes, "demo-crate", "0.1.0").unwrap()).unwrap();
assert_eq!(line["name"], "demo-crate");
assert_eq!(line["vers"], "0.1.0");
assert_eq!(
line["cksum"],
crate::attest::ArtifactDigest::of_sha256(&bytes).hex()
);
assert_eq!(line["v"], 2);
assert_eq!(line["rust_version"], "1.70");
assert_eq!(line["features"]["json"], serde_json::json!(["dep:serde"]));
let deps = line["deps"].as_array().unwrap();
let dep = |n: &str| {
deps.iter()
.find(|d| d["name"] == n)
.unwrap_or_else(|| panic!("no dep {n}: {deps:?}"))
};
assert_eq!(dep("serde")["optional"], true);
assert_eq!(dep("serde")["features"], serde_json::json!(["derive"]));
assert_eq!(dep("log")["req"], "0.4");
assert_eq!(dep("myrand")["package"], "rand");
assert_eq!(dep("myrand")["default_features"], false);
assert_eq!(dep("tempfile")["kind"], "dev");
assert_eq!(dep("libc")["target"], "cfg(unix)");
assert!(
line_from_crate(&bytes, "demo-crate", "0.2.0").is_err(),
"another version's member is not this one"
);
assert!(
line_from_crate(&dot_crate("other", "0.1.0", manifest), "other", "0.1.0").is_err(),
"the manifest must name the crate"
);
}
/// **A dependency on another registry never reaches the index** (SEC-REVIEW 2026-11-01
/// P11). A dependency's `registry-index` went into the line's `registry`, so a scanned and
/// approved crate made `cargo build` fetch that dependency from any registry it named,
/// past Holger. Such a crate has no index line: the door refuses it.
#[test]
fn a_dependency_on_another_registry_has_no_index_line() {
for dep in [
r#"evil = { version = "1", registry-index = "https://evil.example/index" }"#,
r#"evil = { version = "1", registry-index = "https://github.com/rust-lang/crates.io-index.evil.example" }"#,
r#"evil = { version = "1", registry = "evil" }"#,
] {
let manifest = format!(
"[package]\nname = \"demo-crate\"\nversion = \"0.1.0\"\n\n[dependencies]\n{dep}\n"
);
let bytes = dot_crate("demo-crate", "0.1.0", &manifest);
let got = line_from_crate(&bytes, "demo-crate", "0.1.0");
assert!(
got.as_ref().is_err_and(
|e| e.to_string().contains("evil") && e.to_string().contains("registry")
),
"{dep}: {got:?}"
);
}
}
/// A crates.io dependency keeps its index (cargo writes it for every crates.io dependency
/// of a crate published to an alternative registry).
#[test]
fn a_crates_io_dependency_keeps_its_index() {
let manifest = "[package]\nname = \"demo-crate\"\nversion = \"0.1.0\"\n\n[dependencies]\nlog = { version = \"0.4\", registry-index = \"https://github.com/rust-lang/crates.io-index\" }\n";
let bytes = dot_crate("demo-crate", "0.1.0", manifest);
let line: Value =
serde_json::from_str(&line_from_crate(&bytes, "demo-crate", "0.1.0").unwrap()).unwrap();
assert_eq!(
line["deps"][0]["registry"],
"https://github.com/rust-lang/crates.io-index"
);
}
#[test]
fn index_paths_are_cargos() {
assert_eq!(index_path("a"), "1/a");
assert_eq!(index_path("ab"), "2/ab");
assert_eq!(index_path("abc"), "3/a/abc");
assert_eq!(index_path("cfg-if"), "cf/g-/cfg-if");
assert_eq!(index_path("Serde"), "se/rd/serde");
assert_eq!(
name_of_index_path("cf/g-/cfg-if").as_deref(),
Some("cfg-if")
);
assert_eq!(name_of_index_path("cf/xx/cfg-if"), None);
assert_eq!(name_of_index_path("config.json"), None);
}
#[test]
fn a_coordinate_reads_back() {
let i = id("crates-io", "cfg-if", "1.0.5");
assert_eq!(i.name, "crates-io/cfg-if/cfg-if-1.0.5.crate");
assert_eq!(
coords_of("crates-io", &i),
Some(("cfg-if".into(), "1.0.5".into()))
);
}
#[test]
fn the_dl_template_is_filled_as_cargo_fills_it() {
let l = IndexLine {
name: "cfg-if".into(),
vers: "1.0.5".into(),
cksum: "ab".repeat(32),
yanked: false,
raw: Value::Null,
};
assert_eq!(
download_url("https://static.crates.io/crates", &l),
"https://static.crates.io/crates/cfg-if/1.0.5/download"
);
assert_eq!(
download_url(
"https://h/{lowerprefix}/{crate}/{crate}-{version}.crate",
&l
),
"https://h/cf/g-/cfg-if/cfg-if-1.0.5.crate"
);
}
#[test]
fn the_newest_live_release_is_chosen() {
let l = |v: &str, y: bool| IndexLine {
name: "x".into(),
vers: v.into(),
cksum: "0".repeat(64),
yanked: y,
raw: Value::Null,
};
let lines = vec![
l("1.0.0", false),
l("1.2.0", true),
l("1.1.0", false),
l("2.0.0-rc.1", false),
];
assert_eq!(newest(&lines).unwrap().vers, "1.1.0");
}
}