server/lib/tests/a_znippy_download_streams_from_the_archive.rs
//! **HOL-RSS: a download from a znippy archive streams from the archive.**
//!
//! A znippy backend answered a download with `extract_file`: the whole file read into one
//! buffer and handed to hyper as one body, held until the client had the last byte. The
//! README bench (holperf on the real sets, oden, 2026-10-31) measured Holger at 3.6 to
//! 8.4 GiB of peak RSS, and the live heap of the pip set alone peaked at 570 MiB with 32
//! downloads in flight, each holding its wheel; the lean appliance has 1 GB.
//!
//! This binary counts every live heap byte (its own global allocator), seals a 48 MiB
//! wheel (stored, it does not compress) and an 8 MiB file that compresses into one pip
//! archive and downloads both over the real HTTP door: every byte arrives, and the extra
//! memory while a file streams stays within the door's pieces in flight and one chunk.
#![cfg(all(feature = "intake", feature = "fmt-python"))]
use std::alloc::{GlobalAlloc, Layout, System};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use holger_server_lib::audit::NoopAuditLog;
use holger_server_lib::auth::AuthConfig;
use holger_server_lib::exposed::fast_routes::FastRoutes;
use holger_server_lib::exposed::http::{start_http_gateway, GatewaySettings};
use traits::RepositoryBackendTrait;
struct Counting;
static LIVE: AtomicUsize = AtomicUsize::new(0);
static PEAK: AtomicUsize = AtomicUsize::new(0);
unsafe impl GlobalAlloc for Counting {
unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
let p = unsafe { System.alloc(layout) };
if !p.is_null() {
let now = LIVE.fetch_add(layout.size(), Ordering::SeqCst) + layout.size();
PEAK.fetch_max(now, Ordering::SeqCst);
}
p
}
unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
unsafe { System.dealloc(ptr, layout) };
LIVE.fetch_sub(layout.size(), Ordering::SeqCst);
}
}
#[global_allocator]
static A: Counting = Counting;
const MIB: usize = 1024 * 1024;
const WHEEL: &str = "packages/bigpkg/bigpkg-1.0-py3-none-any.whl";
const TEXT: &str = "packages/bigpkg/bigpkg-1.0.tar";
/// `len` bytes that do not compress.
fn noise(len: usize) -> Vec<u8> {
let mut x: u64 = 0x9E37_79B9_7F4A_7C15;
let mut v = Vec::with_capacity(len + 8);
while v.len() < len {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
v.extend_from_slice(&x.to_le_bytes());
}
v.truncate(len);
v
}
/// `len` bytes that do.
fn text(len: usize) -> Vec<u8> {
let line = b"holger keeps one piece of a download in memory, never the artifact\n";
line.iter().copied().cycle().take(len).collect()
}
/// GET `path` on a fresh connection: the head, the body's length and its sha256, read
/// through one buffer allocated before the request.
fn download(addr: std::net::SocketAddr, path: &str, buf: &mut [u8]) -> (String, usize, String) {
use sha2::Digest as _;
use std::io::{Read as _, Write as _};
let mut s = std::net::TcpStream::connect(addr).unwrap();
write!(
s,
"GET {path} HTTP/1.1\r\nHost: h\r\nConnection: close\r\n\r\n"
)
.unwrap();
let (mut head, mut in_body, mut n, mut h) =
(Vec::with_capacity(4096), false, 0usize, sha2::Sha256::new());
loop {
let got = s.read(buf).unwrap();
if got == 0 {
break;
}
let chunk = &buf[..got];
if !in_body {
head.extend_from_slice(chunk);
if let Some(at) = head.windows(4).position(|w| w == b"\r\n\r\n") {
in_body = true;
n += head.len() - (at + 4);
h.update(&head[at + 4..]);
head.truncate(at);
}
continue;
}
n += chunk.len();
h.update(chunk);
}
(
String::from_utf8_lossy(&head).to_string(),
n,
hex(&h.finalize()),
)
}
fn hex(b: &[u8]) -> String {
b.iter().map(|x| format!("{x:02x}")).collect()
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_znippy_download_streams_from_the_archive_in_bounded_memory() {
use sha2::Digest as _;
let dir = tempfile::tempdir().unwrap();
let archive = dir.path().join("store.znippy");
let files = [(WHEEL, noise(48 * MIB)), (TEXT, text(8 * MIB))];
let want: Vec<(String, usize, String)> = files
.iter()
.map(|(p, b)| (p.to_string(), b.len(), hex(&sha2::Sha256::digest(b))))
.collect();
{
let src = dir.path().join("src");
for (p, b) in &files {
let f = src.join(p);
std::fs::create_dir_all(f.parent().unwrap()).unwrap();
std::fs::write(f, b).unwrap();
}
znippy_compress::compress_dir(&src, &archive, false, None, None, None).unwrap();
std::fs::remove_dir_all(&src).unwrap();
}
drop(files);
// the two kinds of chunk a stream reads: the wheel stored, the text compressed
let compressed = |p: &str| {
let locs = traits::znippy_common::index::locate_file(&archive, p).unwrap();
assert!(!locs.is_empty(), "{p} is in the archive");
locs.iter().any(|c| c.compressed)
};
assert!(
!compressed(WHEEL),
"the wheel does not compress, so it is stored"
);
assert!(compressed(TEXT), "the text compresses");
let pypi: Arc<dyn RepositoryBackendTrait> = Arc::new(
repository_znippy_python::PipRepoZnippy::with_archive("pypi".into(), archive).unwrap(),
);
let settings = Arc::new(GatewaySettings {
routes: FastRoutes::new(vec![("pypi".to_string(), pypi)]),
auth_config: Arc::new(AuthConfig::default()),
read_gate: Arc::new(holger_server_lib::auth::ReadGate::GLOBAL_ONLY),
tls: None,
max_body_bytes: 64 * MIB,
audit: Arc::new(NoopAuditLog),
operation_mode: Default::default(),
twin: Default::default(),
grow: Default::default(),
security_profile: Default::default(),
sboms: None,
attest: None,
evidence_key: None,
reserved_properties: Default::default(),
promotion_required_property: None,
imports: None,
properties: None,
admit_upload: None,
trusted_proxies: None,
door: "default".to_string(),
visitors: Arc::new(holger_server_lib::visitors::Visitors::empty()),
console: Default::default(),
peers: Default::default(),
board: None,
flow: None,
scan_probe: None,
ready: None,
#[cfg(feature = "admission")]
admission: holger_server_lib::admission::DoorAdmission::pass_through(),
intake: None,
repo_key: None,
#[cfg(feature = "auth-passkey")]
passkeys: None,
});
let addr = std::net::SocketAddr::from(([127, 0, 0, 1], traits::test_support::hold_port()));
start_http_gateway(addr, settings).expect("the gateway binds");
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
let measured = tokio::task::spawn_blocking(move || {
let mut buf = vec![0u8; 256 * 1024];
let mut out = Vec::new();
for (path, _, _) in [&want[0], &want[1], &want[0]] {
let held = LIVE.load(Ordering::SeqCst);
PEAK.store(held, Ordering::SeqCst);
let (head, n, sha) = download(addr, &format!("/pypi/{path}"), &mut buf);
let extra = PEAK.load(Ordering::SeqCst).saturating_sub(held);
out.push((path.clone(), head, n, sha, extra));
}
(out, want)
})
.await
.unwrap();
let (measured, want) = measured;
for (path, head, n, sha, extra) in measured {
eprintln!("{path}: {n} bytes, {} KiB extra live", extra / 1024);
let (_, len, digest) = want.iter().find(|w| w.0 == path).unwrap();
assert!(head.starts_with("HTTP/1.1 200"), "{path}: {head}");
assert_eq!(
(n, sha.as_str()),
(*len, digest.as_str()),
"{path}: every byte arrives"
);
// the door's pieces in flight (8 of 1 MiB) and the one compressed chunk decoded
assert!(
extra <= 12 * MIB,
"{path}: streaming {} MiB out of the archive took {} MiB more memory: it must stay within the door's pieces in flight and one chunk",
len / MIB,
extra / MIB
);
}
}