nordisk/holger

server/lib/tests/one_gib_put_never_holds_the_body.rs

//! W12's memory guard, in a process of its own.
//!
//! A 1 GiB `PUT` through the raw repository's streaming door must not make
//! the server hold the body. `VmHWM` — the process's lifetime peak resident
//! set, from `/proc/self/status` — is the judge, and it is only a judge when
//! nothing else shares the process, which is why this is an integration test
//! and not one of the seven hundred in the lib binary. The body is generated
//! as it is sent on the client side too, so the only place a gigabyte COULD
//! sit is the server.

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 holger_server_lib::raw::RawBackend;

const LEN: u64 = 1024 * 1024 * 1024;

/// A deterministic byte stream produced 8 bytes at a time — never whole in
/// memory on the client side.
struct Pattern {
    remaining: u64,
    x: u64,
}

impl Pattern {
    fn fill(&mut self, buf: &mut [u8]) -> usize {
        let n = (buf.len() as u64).min(self.remaining) as usize;
        let mut x = self.x;
        for chunk in buf[..n].chunks_mut(8) {
            x ^= x << 13;
            x ^= x >> 7;
            x ^= x << 17;
            let le = x.to_le_bytes();
            chunk.copy_from_slice(&le[..chunk.len()]);
        }
        self.x = x;
        self.remaining -= n as u64;
        n
    }
}

/// This process's lifetime peak resident set, from `/proc/self/status`.
fn vm_hwm_bytes() -> u64 {
    let s = std::fs::read_to_string("/proc/self/status").unwrap_or_default();
    s.lines()
        .find_map(|l| l.strip_prefix("VmHWM:"))
        .and_then(|v| v.split_whitespace().next())
        .and_then(|kb| kb.parse::<u64>().ok())
        .map(|kb| kb * 1024)
        .unwrap_or(0)
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_one_gib_put_never_holds_the_whole_body() {
    let dir = tempfile::tempdir().unwrap();
    let raw: Arc<dyn traits::RepositoryBackendTrait> =
        Arc::new(RawBackend::new("raw".into(), dir.path().to_path_buf()));
    let settings = Arc::new(GatewaySettings {
        routes: FastRoutes::new(vec![("raw".to_string(), raw)]),
        auth_config: Arc::new(AuthConfig::default()),
        read_gate: Arc::new(holger_server_lib::auth::ReadGate::GLOBAL_ONLY),
        tls: None,
        max_body_bytes: 2 * 1024 * 1024 * 1024,
        audit: Arc::new(NoopAuditLog),
        operation_mode: Default::default(),
        twin: Default::default(),
        grow: Default::default(),
        security_profile: Default::default(),
        #[cfg(feature = "intake")]
        sboms: None,
        #[cfg(feature = "intake")]
        attest: None,
        #[cfg(feature = "intake")]
        evidence_key: None,
        repo_key: None,
        #[cfg(feature = "intake")]
        reserved_properties: Default::default(),
        #[cfg(feature = "intake")]
        promotion_required_property: None,
        #[cfg(feature = "intake")]
        imports: None,
        #[cfg(feature = "intake")]
        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(),
        #[cfg(feature = "intake")]
        intake: None,
        #[cfg(feature = "auth-passkey")]
        passkeys: None,
    });
    let addr = {
        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let a = l.local_addr().unwrap();
        drop(l);
        a
    };
    start_http_gateway(addr, settings).expect("gateway binds");
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    let status = tokio::task::spawn_blocking(move || {
        use std::io::{BufRead as _, BufReader, Write as _};
        let mut s = std::net::TcpStream::connect(addr).unwrap();
        write!(
            s,
            "PUT /raw/big/blob.bin HTTP/1.1\r\nHost: {addr}\r\nContent-Length: {LEN}\r\nConnection: close\r\n\r\n"
        )
        .unwrap();
        let mut gen = Pattern { remaining: LEN, x: 0x9E37_79B9_7F4A_7C15 };
        let mut buf = vec![0u8; 1 << 20];
        loop {
            let n = gen.fill(&mut buf);
            if n == 0 {
                break;
            }
            s.write_all(&buf[..n]).unwrap();
        }
        s.flush().unwrap();
        let mut r = BufReader::new(s);
        let mut line = String::new();
        r.read_line(&mut line).unwrap();
        line.split_whitespace().nth(1).and_then(|c| c.parse::<u16>().ok()).unwrap_or(0)
    })
    .await
    .unwrap();
    assert_eq!(status, 201, "the streamed 1 GiB PUT must land");
    let on_disk = std::fs::metadata(dir.path().join("big/blob.bin"))
        .unwrap()
        .len();
    assert_eq!(on_disk, LEN, "the whole blob must be on disk");

    let hwm = vm_hwm_bytes();
    assert!(
        hwm > 0,
        "VmHWM must be readable for this guard to mean anything"
    );
    assert!(
        hwm < 256 * 1024 * 1024,
        "peak resident set was {} MiB during a 1 GiB PUT: the door is holding the body",
        hwm / (1024 * 1024)
    );
}