gunnar-server 3.0.0

//! **One entry bigger than the batch budget is streamed, never held** (twin-streaming,
//! 2026-10-06) — measured over the real wire, `Mirror.LogFetch` into the appliance's own
//! `GrpcSource` and `Follower`.
//!
//! The batch budget (`DEFAULT_FETCH_BYTES`, 64 MiB) bounds how many entries one fetch carries;
//! it never bounded ONE entry. Until this file, a 256 MiB pack cost its whole size several
//! times over on both boxes: the primary's `fetch_range` read the entry whole, `chunks_of` cut
//! a second whole copy into 1 MiB messages before the first left, and the twin's
//! `GrpcSource::fetch` glued the pieces back into one `Vec` before it appended it. That is the
//! shape of test2's old "1 GB twin silent under 256 MiB packs".
//!
//! A primary's log gets ONE 512 MiB frame (written in 1 MiB pieces, so the fixture never holds
//! it either), the process's peak resident set is reset (`/proc/self/clear_refs` 5), and a twin
//! — a fresh data root made a twin by `become_twin` — follows it over a loopback gRPC listener.
//! Both boxes are in this ONE process, so the judge is both halves at once: `VmHWM` above the
//! resident set at the reset may not pass [`CEILING`]. This file is its own test binary so the
//! peak is this transfer's and no other test's.
//!
//! The copy is then proven, not assumed: the twin's frame is byte-identical to the primary's
//! (hashed in pieces from both logs), and the twin's head is the primary's.

use std::io::{Read, Seek, SeekFrom};
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::{Duration, Instant};

use gunnar_core::txlog::{Kind, Log, Meta};
use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
use gunnar_server::stores::ServerStores;
use gunnar_server::twin::{self, Follower, GrpcSource, Progress};
use gunnar_server::CoreControlPlane;

/// The one entry: eight times the batch budget.
const ENTRY_BYTES: u64 = 512 << 20;

/// Growth of the peak resident set allowed while the twin takes it, both boxes together. Holding
/// the entry ONCE on either end is 512 MiB.
const CEILING: u64 = 64 << 20;

fn free_addr() -> SocketAddr {
    let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
    l.local_addr().expect("addr")
}

/// `(VmRSS, VmHWM)` of this process, bytes.
fn rss_hwm() -> (u64, u64) {
    let s = std::fs::read_to_string("/proc/self/status").unwrap_or_default();
    let kib = |key: &str| {
        s.lines()
            .find_map(|l| l.strip_prefix(key))
            .and_then(|v| v.split_whitespace().next())
            .and_then(|v| v.parse::<u64>().ok())
            .unwrap_or(0)
            << 10
    };
    (kib("VmRSS:"), kib("VmHWM:"))
}

/// Reset this process's peak resident set to what it holds now.
fn reset_hwm() {
    std::fs::write("/proc/self/clear_refs", "5").expect("clear_refs 5 resets VmHWM (Linux >= 4.0)");
}

/// The fixture's bytes, eight at a time (an LCG: incompressible enough, and cheap in a debug
/// build), written into the frame piece by piece — the primary never holds the entry either.
fn append_big_frame(log: &Log, len: u64) -> u64 {
    let meta = Meta::Pack {
        repo: "alice/big".into(),
        hash: "sha1".into(),
    }
    .to_bytes();
    let mut pending = log
        .begin(None, Kind::Pack as u16, &meta, len)
        .expect("begin the big frame");
    let mut state: u64 = 0x2545_F491_4F6C_DD1D;
    let mut buf = vec![0u8; 1 << 20];
    let mut left = len;
    while left > 0 {
        let n = (buf.len() as u64).min(left) as usize;
        for c in buf[..n].chunks_mut(8) {
            state = state
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            c.copy_from_slice(&(state >> 1).to_le_bytes()[..c.len()]);
        }
        pending.write(&buf[..n]).expect("write a piece");
        left -= n as u64;
    }
    pending.commit().expect("commit the big frame")
}

/// A frame's blob hashed in 1 MiB pieces off the segment file, never whole.
fn blob_digest(log: &Log, seq: u64) -> (u64, u64) {
    let at = log.locate_blob(seq).expect("locate");
    let mut f = std::fs::File::open(&at.path).expect("segment");
    f.seek(SeekFrom::Start(at.blob_offset)).expect("seek");
    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
    let mut buf = vec![0u8; 1 << 20];
    let mut left = at.blob_len;
    while left > 0 {
        let n = (buf.len() as u64).min(left) as usize;
        f.read_exact(&mut buf[..n]).expect("read");
        for b in &buf[..n] {
            h ^= u64::from(*b);
            h = h.wrapping_mul(0x0100_0000_01b3);
        }
        left -= n as u64;
    }
    (at.blob_len, h)
}

#[test]
fn one_entry_eight_times_the_batch_budget_streams_to_the_twin_with_bounded_memory() {
    // ── the primary: a real control plane on loopback, one 512 MiB frame in its log ──
    let pd = tempfile::tempdir().expect("tempdir");
    let primary = Arc::new(ServerStores::open(pd.path()).expect("primary stores"));
    let big = append_big_frame(primary.txlog(), ENTRY_BYTES);
    let plane: Arc<dyn gunnar_grpc::ControlPlane> =
        Arc::new(CoreControlPlane::new(Arc::clone(&primary)));
    let ctx = ServiceContext::new(
        plane,
        Arc::new(EventBus::new(16)),
        ServerConfig::default().insecure_allow_all_for_local_development(),
    );
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .expect("rt");
    let addr = free_addr();
    let (stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>();
    rt.spawn(async move {
        let _ = gunnar_grpc::router(ctx, || 0)
            .serve_with_shutdown(addr, async {
                let _ = stop_rx.await;
            })
            .await;
    });
    let t0 = Instant::now();
    while std::net::TcpStream::connect(addr).is_err() {
        assert!(
            t0.elapsed() < Duration::from_secs(10),
            "the primary never listened"
        );
        std::thread::sleep(Duration::from_millis(20));
    }
    let head = primary.txlog().head_seq();
    assert_eq!(head, big, "the big frame is the primary's head");

    // ── the twin ──
    let td = tempfile::tempdir().expect("tempdir");
    let twin_stores = Arc::new(ServerStores::open(td.path()).expect("twin stores"));
    twin::become_twin(&twin_stores, &addr.to_string()).expect("a fresh box becomes a twin");
    let key = twin::twin_private_key(twin_stores.root()).expect("the twin's key");
    let source = GrpcSource::connect(&addr.to_string(), key, twin_stores.root()).expect("source");
    let follower = Follower::new(Arc::clone(&twin_stores), Box::new(source));

    // ── measured: the twin takes everything up to the head ──
    std::thread::sleep(Duration::from_millis(200));
    let (rss0, _) = rss_hwm();
    reset_hwm();
    let t = Instant::now();
    let mut progress = follower.step();
    while matches!(progress, Ok(Progress::Applied { .. })) && twin_stores.txlog().head_seq() < head
    {
        progress = follower.step();
    }
    let took = t.elapsed();
    let (rss1, hwm) = rss_hwm();
    let growth = hwm.saturating_sub(rss0);
    let line = format!(
        "one {} MiB entry (batch budget {} MiB) followed over loopback gRPC in {:.1} s: peak resident set +{} MiB over {} MiB at the reset (ceiling +{} MiB), RSS now {} MiB",
        ENTRY_BYTES >> 20,
        twin::DEFAULT_FETCH_BYTES >> 20,
        took.as_secs_f64(),
        growth >> 20,
        rss0 >> 20,
        CEILING >> 20,
        rss1 >> 20
    );
    eprintln!("MEASURED {line}");
    let _ = stop_tx.send(());

    let progress = progress.expect("the twin follows the primary");
    assert!(
        matches!(
            progress,
            Progress::Applied { .. } | Progress::CaughtUp { .. }
        ),
        "{progress:?}"
    );
    assert_eq!(
        twin_stores.txlog().head_seq(),
        head,
        "the twin holds every entry up to the primary's head"
    );
    assert_eq!(
        blob_digest(twin_stores.txlog(), big),
        blob_digest(primary.txlog(), big),
        "the twin's copy of the big frame is the primary's, byte for byte"
    );
    let e = twin_stores
        .txlog()
        .locate_blob(big)
        .expect("the twin's frame");
    assert_eq!((e.kind, e.blob_len), (Kind::Pack as u16, ENTRY_BYTES));
    assert!(
        growth < CEILING,
        "the entry was held whole, not streamed: {line}"
    );
    gunnar_server::functional_status(
        "twin",
        "one entry 8x the batch budget streams to the twin under +64 MiB peak",
        true,
        &line,
    );
}