gunnar-server 3.0.0

//! **An entry larger than one call is restored — resumed INSIDE it** (restore-large-entry,
//! 2026-10-07).
//!
//! MEASURED by the r5 integration test (row R1-recharge-main-lost): a new main recharging from
//! its twin through monetize's restore lane restored up to a 200 MiB pack and then, every call,
//! "Bootstrap.Restore refused: the stream ended inside an entry; resend from the box's head + 1"
//! — 41 times, never progressing. Every stream the lane opens is cut a fixed time after it opens,
//! and the lane resumed per ENTRY: one entry that needs longer than a call to stream could never
//! be restored, and the main never came back.
//!
//! Here two real boxes — a twin that followed a primary, and a fresh main — are served on the
//! real wire, and a LANE drives the recharge the way monetize's gunnar guest does: per call, read
//! `Bootstrap.Status`, fetch the twin's log from where the box is (`Mirror.LogFetch`, from the
//! byte of the entry it holds part of when it says one), push it into `Bootstrap.Restore`, and
//! stop when the call's budget is spent — a byte budget here, so the test is deterministic: the
//! guest's is time. The big entry needs several calls; the lane must get through it.

use std::net::SocketAddr;
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, Instant};

use gix_object::Kind as ObjKind;
use gunnar_core::{HashKind, NewRepo, Principal, PrincipalKind, RepoRef};
use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
use gunnar_proto::bootstrap_client::BootstrapClient;
use gunnar_proto::mirror_client::MirrorClient;
use gunnar_server::stores::ServerStores;
use gunnar_server::twin::{self, Follower, Progress};
use gunnar_server::CoreControlPlane;
use gunnar_store::conformance::pack::PackBuilder;

/// The big entry: a pack holding one blob of noise this large (noise, so the pack is as large).
const BIG: usize = 24 << 20;
/// What one call of the lane carries: a third of the big entry and a bit — it is cut inside it
/// at least three times.
const BUDGET: u64 = 7 << 20;
/// Where a restoring box stages the entry it holds part of.
const PARTIAL_DIR: &str = "txlog-restore-partial";

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

/// A control plane for `stores`, served at `addr` until killed.
struct Served {
    _rt: tokio::runtime::Runtime,
    stop: Option<tokio::sync::oneshot::Sender<()>>,
    addr: SocketAddr,
}

fn serve(stores: &Arc<ServerStores>) -> Served {
    let addr = free_addr();
    let plane: Arc<dyn gunnar_grpc::ControlPlane> =
        Arc::new(CoreControlPlane::new(Arc::clone(stores)));
    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 (tx, rx) = tokio::sync::oneshot::channel::<()>();
    rt.spawn(async move {
        let _ = gunnar_grpc::router(ctx, || 0)
            .serve_with_shutdown(addr, async {
                let _ = rx.await;
            })
            .await;
    });
    wait("the box listens", Duration::from_secs(10), || {
        std::net::TcpStream::connect(addr).ok()
    });
    Served {
        _rt: rt,
        stop: Some(tx),
        addr,
    }
}

impl Served {
    fn kill(mut self) {
        if let Some(tx) = self.stop.take() {
            let _ = tx.send(());
        }
        let addr = self.addr;
        wait("the box stops listening", Duration::from_secs(10), || {
            std::net::TcpStream::connect(addr).is_err().then_some(())
        });
    }
}

fn wait<T>(what: &str, within: Duration, mut probe: impl FnMut() -> Option<T>) -> T {
    let t0 = Instant::now();
    loop {
        if let Some(v) = probe() {
            return v;
        }
        assert!(t0.elapsed() < within, "{what}: not within {within:?}");
        std::thread::sleep(Duration::from_millis(25));
    }
}

fn noise(seed: u64, n: usize) -> Vec<u8> {
    let mut x = seed
        .wrapping_mul(6364136223846793005)
        .wrapping_add(1442695040888963407);
    (0..n)
        .map(|_| {
            x = x
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            (x >> 33) as u8
        })
        .collect()
}

fn one_blob_pack(hash: gix_hash::Kind, content: &[u8]) -> (Vec<u8>, Vec<u8>) {
    let id = gix_object::compute_hash(hash, ObjKind::Blob, content).expect("hash");
    let mut pb = PackBuilder::new(hash);
    pb.object(ObjKind::Blob, content.to_vec());
    let fixture = pb.build().expect("pack");
    (fixture.bytes, id.as_bytes().to_vec())
}

/// What the primary holds: small entries on both sides of ONE big pack.
struct Primary {
    stores: Arc<ServerStores>,
    oids: Vec<Vec<u8>>,
}

fn primary(dir: &Path) -> Primary {
    let stores = Arc::new(ServerStores::open(dir).expect("primary stores"));
    stores.catalog().create_namespace("ns").expect("namespace");
    let repo = stores
        .create_repo(
            NewRepo::new(RepoRef::new("ns", "app").unwrap()).hash_kind(HashKind::Sha256),
            false,
        )
        .expect("repo");
    let store = stores.repo_store(&repo).expect("store");
    let git = store.serve();
    let hash = store.hash_kind();
    let (p1, o1) = one_blob_pack(hash, b"before the big one\n");
    let (p2, o2) = one_blob_pack(hash, &noise(7, BIG));
    let (p3, o3) = one_blob_pack(hash, b"after the big one\n");
    git.put_pack(&p1).expect("pack 1");
    git.update_ref("refs/heads/main", None, Some(&o1))
        .expect("main");
    git.put_pack(&p2).expect("the big pack");
    git.update_ref("refs/heads/big", None, Some(&o2))
        .expect("big");
    git.put_pack(&p3).expect("pack 3");
    git.update_ref("refs/heads/after", None, Some(&o3))
        .expect("after");
    stores
        .auth()
        .add_principal(Principal::new("alice", "Alice", PrincipalKind::User).unwrap())
        .expect("principal");
    assert!(
        p2.len() >= BIG,
        "the big pack is as big as its noise: {}",
        p2.len()
    );
    Primary {
        stores,
        oids: vec![o1, o2, o3],
    }
}

/// A twin of `primary` — a fresh box made a twin, following it until level — in `dir`.
fn twin_of(primary: &Primary, dir: &Path) -> Arc<ServerStores> {
    let twin = Arc::new(ServerStores::open(dir).expect("twin stores"));
    twin::become_twin(&twin, "primary.test:50051").expect("a fresh box becomes a twin");
    drop(twin);
    let twin = Arc::new(ServerStores::open(dir).expect("the twin, reopened as one"));
    // What a twin over the wire is told on its first poll (an in-process source carries no
    // identity): a box restored from it is restored from a NAMED history, as a network
    // restore requires.
    let id = twin::log_id(primary.stores.root()).expect("a primary mints its log's identity");
    twin::pin_log_id(twin.root(), &id).expect("the twin pins its primary's log identity");
    let f = Follower::new(
        Arc::clone(&twin),
        Box::new(Arc::clone(primary.stores.txlog())),
    );
    let head = primary.stores.txlog().head_seq();
    for _ in 0..64 {
        match f.step().expect("the twin follows") {
            Progress::CaughtUp { .. } => break,
            Progress::Applied { .. } => continue,
            other => panic!("the twin cannot follow: {other:?}"),
        }
    }
    assert_eq!(
        twin.txlog().head_seq(),
        head,
        "the twin is level with the primary"
    );
    twin
}

/// The sequence of the big pack's frame in a log.
fn big_seq(log: &gunnar_core::txlog::Log) -> u64 {
    (log.first_seq()..=log.head_seq())
        .find(|s| log.read(*s).map(|e| e.blob.len() >= BIG).unwrap_or(false))
        .expect("the big frame")
}

/// How a call of the lane ends when its budget is spent.
#[derive(Clone, Copy, Debug)]
enum Cut {
    /// The guest's own budget: it closes the request half at a piece boundary.
    Close,
    /// The host's deadline: the call is CANCELLED (tonic reads it as a clean end, or an error).
    Abort,
    /// An old twin: no `from_offset` asked or honoured, no `offset` on the chunks; closed at
    /// the budget.
    OldTwinClose,
}

/// What one call did.
#[derive(Debug)]
enum Call {
    /// The box answered the close: restored to this head.
    Done(u64),
    /// The budget ended the call; what the box said, if it said anything.
    Stopped(String),
}

/// **The lane**: one call = what the gunnar guest does in one call, over the real wire.
struct Lane {
    rt: tokio::runtime::Runtime,
    twin: SocketAddr,
    budget: u64,
}

impl Lane {
    fn new(twin: SocketAddr, budget: u64) -> Lane {
        Lane {
            rt: tokio::runtime::Builder::new_multi_thread()
                .worker_threads(2)
                .enable_all()
                .build()
                .unwrap(),
            twin,
            budget,
        }
    }

    fn status(&self, main: SocketAddr) -> gunnar_proto::BootstrapStatusResponse {
        self.rt.block_on(async {
            let mut c = BootstrapClient::connect(format!("http://{main}"))
                .await
                .expect("connect");
            c.status(gunnar_proto::BootstrapStatusRequest {})
                .await
                .expect("Bootstrap.Status")
                .into_inner()
        })
    }

    fn call(&self, main: SocketAddr, cut: Cut) -> Call {
        let budget = self.budget;
        let twin = self.twin;
        let st = self.status(main);
        self.rt.block_on(async move {
            let mut mirror = MirrorClient::connect(format!("http://{twin}"))
                .await
                .expect("the twin");
            let head = mirror
                .log_head(gunnar_proto::LogHeadRequest::default())
                .await
                .expect("LogHead")
                .into_inner();
            let from = if st.restoring {
                head.first_seq.max(st.log_head_seq + 1)
            } else {
                head.first_seq.max(1)
            };
            let old_twin = matches!(cut, Cut::OldTwinClose);
            let offset = if st.restoring && st.restore_partial_seq == from && !old_twin {
                st.restore_partial_bytes
            } else {
                0
            };
            let (tx, rx) = tokio::sync::mpsc::channel::<gunnar_proto::LogChunk>(4);
            let restore = tokio::spawn(async move {
                let mut c = BootstrapClient::connect(format!("http://{main}"))
                    .await
                    .expect("connect main");
                c.restore(tokio_stream::wrappers::ReceiverStream::new(rx))
                    .await
                    .map(|r| r.into_inner())
            });
            let (mut cursor, mut sent, mut highest) = (from, 0u64, 0u64);
            while cursor <= head.head_seq {
                let req = gunnar_proto::LogFetchRequest {
                    from: cursor,
                    to: head.head_seq,
                    max_bytes: 0,
                    from_offset: if cursor == from { offset } else { 0 },
                };
                let mut s = mirror.log_fetch(req).await.expect("LogFetch").into_inner();
                while let Some(mut chunk) = s.message().await.expect("the twin's stream") {
                    chunk.restore_head = head.head_seq;
                    chunk.log_id.clone_from(&head.log_id);
                    if old_twin {
                        chunk.offset = None;
                        chunk.entry_crc = None;
                    }
                    let (seq, last) = (chunk.seq, chunk.last);
                    sent += chunk.blob.len() as u64;
                    highest = highest.max(seq);
                    if tx.send(chunk).await.is_err() {
                        break;
                    }
                    if sent >= budget && !(last && seq >= head.head_seq) {
                        return match cut {
                            Cut::Close | Cut::OldTwinClose => {
                                drop(tx);
                                match restore.await.expect("the restore task") {
                                    Ok(r) => {
                                        Call::Stopped(format!("answered OK at head {}", r.head_seq))
                                    }
                                    Err(status) => Call::Stopped(status.message().to_owned()),
                                }
                            }
                            Cut::Abort => {
                                restore.abort();
                                drop(tx);
                                Call::Stopped("cancelled".into())
                            }
                        };
                    }
                }
                assert!(highest >= cursor, "the twin answered nothing at {cursor}");
                cursor = highest + 1;
            }
            drop(tx);
            match restore.await.expect("the restore task") {
                Ok(r) => Call::Done(r.head_seq),
                Err(status) => Call::Stopped(status.message().to_owned()),
            }
        })
    }
}

/// Every frame of `restored` is `twin`'s, byte for byte.
fn assert_frames_equal(twin: &ServerStores, restored: &ServerStores) {
    let (t, r) = (twin.txlog(), restored.txlog());
    assert_eq!(
        r.head_seq(),
        t.head_seq(),
        "the restored main is at the twin's head"
    );
    for seq in t.first_seq()..=t.head_seq() {
        assert_eq!(
            r.read(seq).unwrap(),
            t.read(seq).unwrap(),
            "seq {seq} differs between the twin's log and the restored main's"
        );
    }
}

/// The restored main, restarted as the primary it was, holds the primary's objects and refs.
fn assert_same_objects(primary: &Primary, restored: &ServerStores) {
    let at = RepoRef::new("ns", "app").unwrap();
    let p = primary
        .stores
        .repo_store(&primary.stores.catalog().get(&at).unwrap().unwrap())
        .unwrap();
    let r = restored
        .repo_store(
            &restored
                .catalog()
                .get(&at)
                .unwrap()
                .expect("the repository came back"),
        )
        .unwrap();
    for oid in &primary.oids {
        assert!(
            r.serve().has(oid).unwrap(),
            "object {} missing on the restored main",
            hex::encode(oid)
        );
        assert_eq!(
            p.serve().read(oid).unwrap(),
            r.serve().read(oid).unwrap(),
            "object {} differs",
            hex::encode(oid)
        );
    }
    let mut refs_p: Vec<_> = p
        .serve()
        .refs()
        .unwrap()
        .into_iter()
        .map(|r| (r.name, r.oid))
        .collect();
    let mut refs_r: Vec<_> = r
        .serve()
        .refs()
        .unwrap()
        .into_iter()
        .map(|r| (r.name, r.oid))
        .collect();
    refs_p.sort();
    refs_r.sort();
    assert_eq!(refs_p, refs_r, "refs");
}

/// ★ **The defect, and its cure.** A 24 MiB entry, a lane that carries 7 MiB per call, calls that
/// END cleanly at their budget (the guest's own deadline): the entry is cut inside at least three
/// times, each call resumes it at the byte the box holds durably, and the restore finishes. The
/// staging goes the moment the entry is applied — nothing is left on the volume — and the
/// restored main is the twin's log frame for frame and the primary's objects and refs.
///
/// RED before the fix: the box dropped the half entry every call and the lane asked for it again
/// from byte 0 — 12 calls, never past the big entry.
#[test]
fn an_entry_larger_than_one_call_is_cut_three_times_and_resumed_inside_it() {
    let (pd, td, md) = (
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
    );
    let primary = primary(pd.path());
    let twin = twin_of(&primary, td.path());
    let big = big_seq(twin.txlog());
    let twin_served = serve(&twin);
    let main = Arc::new(ServerStores::open(md.path()).expect("a fresh main"));
    let main_served = serve(&main);
    let lane = Lane::new(twin_served.addr, BUDGET);

    let mut inside = Vec::new();
    let mut done = None;
    for call in 0..12 {
        let said = lane.call(main_served.addr, Cut::Close);
        let st = lane.status(main_served.addr);
        println!(
            "call {call}: {said:?} — head {} restoring {} partial {}:{}",
            st.log_head_seq, st.restoring, st.restore_partial_seq, st.restore_partial_bytes
        );
        if st.log_head_seq >= big {
            assert!(
                !md.path().join(PARTIAL_DIR).exists(),
                "call {call}: the big entry is applied and its staging is still on the volume"
            );
        } else if st.log_head_seq + 1 == big {
            inside.push(st.restore_partial_bytes);
        }
        if let Call::Done(head) = said {
            done = Some(head);
            break;
        }
    }
    assert_eq!(
        done,
        Some(twin.txlog().head_seq()),
        "the restore never finished; the box held, call after call, {inside:?} bytes of seq {big}"
    );
    assert!(
        inside.len() >= 3,
        "the big entry was cut inside {} time(s), not three: {inside:?}",
        inside.len()
    );
    assert!(
        inside.windows(2).all(|w| w[1] > w[0]),
        "every call went further INTO the entry: {inside:?}"
    );
    assert!(
        !md.path().join(PARTIAL_DIR).exists(),
        "nothing staged is left on the volume"
    );
    assert_frames_equal(&twin, &main);
    main_served.kill();
    drop(main);
    let restored = ServerStores::open(md.path()).expect("the restored main, restarted");
    assert_same_objects(&primary, &restored);
    twin_served.kill();
}

/// ★ **A main restarted in the middle of the entry resumes from what it made durable.** The
/// calls are CANCELLED at their budget (the host's deadline, not the guest's close); after two
/// of them the main goes down — and, as a crash would, its staged file carries bytes past what
/// its record counts. It comes back, says the same durable count it said before, and the lane
/// finishes the entry from there; the result is the twin's log and the primary's objects.
#[test]
fn a_main_restarted_inside_the_entry_resumes_from_what_it_made_durable() {
    let (pd, td, md) = (
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
    );
    let primary = primary(pd.path());
    let twin = twin_of(&primary, td.path());
    let big = big_seq(twin.txlog());
    let twin_served = serve(&twin);
    let lane = Lane::new(twin_served.addr, BUDGET);

    let mut main = Arc::new(ServerStores::open(md.path()).expect("a fresh main"));
    let mut main_served = serve(&main);
    let mut held = Vec::new();
    for _ in 0..2 {
        lane.call(main_served.addr, Cut::Abort);
        // The cancelled stream's handler commits as it ends; give it the moment it needs.
        let st = wait(
            "the cancelled stream's handler let go",
            Duration::from_secs(20),
            || {
                let st = lane.status(main_served.addr);
                (st.restore_partial_seq == 0
                    || st.restore_partial_bytes > held.last().copied().unwrap_or(0))
                .then_some(st)
            },
        );
        held.push(st.restore_partial_bytes);
    }
    let before = lane.status(main_served.addr);
    // The main goes down mid-entry...
    main_served.kill();
    drop(main);
    // ...and, as after a power cut, its staged file holds a tail nobody counted.
    let data = md.path().join(PARTIAL_DIR).join("entry.blob");
    if data.exists() {
        use std::io::Write as _;
        std::fs::OpenOptions::new()
            .append(true)
            .open(&data)
            .unwrap()
            .write_all(&noise(99, 12_345))
            .unwrap();
    }
    main = Arc::new(ServerStores::open(md.path()).expect("the main, back mid-restore"));
    main_served = serve(&main);
    let after = lane.status(main_served.addr);
    assert!(after.restoring, "the main comes back still restoring");
    assert_eq!(
        (after.restore_partial_seq, after.restore_partial_bytes),
        (before.restore_partial_seq, before.restore_partial_bytes),
        "it holds what it made durable — no more, no less"
    );

    let mut done = None;
    for _ in 0..12 {
        if let Call::Done(head) = lane.call(main_served.addr, Cut::Close) {
            done = Some(head);
            break;
        }
    }
    assert_eq!(
        done,
        Some(twin.txlog().head_seq()),
        "the restore never finished after the restart (held {held:?} of seq {big} before it)"
    );
    assert!(
        !md.path().join(PARTIAL_DIR).exists(),
        "nothing staged is left on the volume"
    );
    assert_frames_equal(&twin, &main);
    main_served.kill();
    drop(main);
    let restored = ServerStores::open(md.path()).expect("the restored main, restarted");
    assert_same_objects(&primary, &restored);
    twin_served.kill();
}

/// **An older twin cannot carry an entry that needs more than one call, and the box says so by
/// name** instead of asking for the same resend for ever. The old twin ignores `from_offset` and
/// sends the entry from byte 0 with no `offset` on its chunks; the box skips what it holds, which
/// makes the stream no shorter: once a stream of such a sender ends without one byte past what was
/// held, the refusal is `restore-twin-too-old` — "upgrade the twin" — and nothing held is lost.
#[test]
fn an_old_twin_that_cannot_resume_inside_an_entry_is_named_not_looped_on() {
    let (pd, td, md) = (
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
    );
    let primary = primary(pd.path());
    let twin = twin_of(&primary, td.path());
    let twin_served = serve(&twin);
    let main = Arc::new(ServerStores::open(md.path()).expect("a fresh main"));
    let main_served = serve(&main);
    let lane = Lane::new(twin_served.addr, BUDGET);

    let first = lane.call(main_served.addr, Cut::OldTwinClose);
    let mut held = lane.status(main_served.addr).restore_partial_bytes;
    assert!(
        held > 0,
        "the first call staged part of the big entry: {first:?}"
    );
    // Each resend from byte 0 ends where the last did (the budget is the same): at the latest
    // the second one gains nothing, and is named.
    let mut named = None;
    for _ in 0..3 {
        let Call::Stopped(said) = lane.call(main_served.addr, Cut::OldTwinClose) else {
            panic!("an old twin cannot finish the big entry")
        };
        let now = lane.status(main_served.addr).restore_partial_bytes;
        assert!(now >= held, "nothing held is lost: {now} < {held}");
        held = now;
        if said.contains("restore-twin-too-old") {
            named = Some(said);
            break;
        }
    }
    let said = named.expect("the box never named the old twin");
    assert!(
        said.contains("upgrade the twin"),
        "the box says what to do: {said}"
    );
    // The twin upgraded: the same box finishes from what it held.
    let mut done = None;
    for _ in 0..12 {
        if let Call::Done(head) = lane.call(main_served.addr, Cut::Close) {
            done = Some(head);
            break;
        }
    }
    assert_eq!(done, Some(twin.txlog().head_seq()));
    assert_frames_equal(&twin, &main);
    main_served.kill();
    twin_served.kill();
}

/// **Bytes put together across calls are applied only if they ARE the entry.** One staged byte
/// flips on the main's disk between two calls: the last piece's crc (the twin's own payload crc,
/// checked before it left) does not match what the box assembled, nothing is applied, the
/// staging is dropped, and the entry starts again from byte 0 — and then finishes, the twin's
/// log frame for frame.
#[test]
fn a_staged_entry_whose_bytes_rotted_between_calls_is_refused_by_its_crc_and_taken_again() {
    let (pd, td, md) = (
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
        tempfile::tempdir().unwrap(),
    );
    let primary = primary(pd.path());
    let twin = twin_of(&primary, td.path());
    let big = big_seq(twin.txlog());
    let twin_served = serve(&twin);
    let main = Arc::new(ServerStores::open(md.path()).expect("a fresh main"));
    let main_served = serve(&main);
    let lane = Lane::new(twin_served.addr, BUDGET);

    lane.call(main_served.addr, Cut::Close);
    let st = lane.status(main_served.addr);
    assert_eq!(
        st.restore_partial_seq, big,
        "the first call staged part of the big entry"
    );
    {
        use std::io::{Read as _, Seek as _, SeekFrom, Write as _};
        let mut f = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(md.path().join(PARTIAL_DIR).join("entry.blob"))
            .unwrap();
        f.seek(SeekFrom::Start(1000)).unwrap();
        let mut b = [0u8; 1];
        f.read_exact(&mut b).unwrap();
        f.seek(SeekFrom::Start(1000)).unwrap();
        f.write_all(&[b[0] ^ 0x5A]).unwrap();
    }
    let mut refused = None;
    let mut done = None;
    for _ in 0..16 {
        match lane.call(main_served.addr, Cut::Close) {
            Call::Done(head) => {
                done = Some(head);
                break;
            }
            Call::Stopped(said) if said.contains("crc") => refused = Some(said),
            Call::Stopped(_) => {}
        }
    }
    let refused = refused.expect("the rotted entry was refused by its crc");
    assert!(refused.contains("starts again from byte 0"), "{refused}");
    assert_eq!(
        done,
        Some(twin.txlog().head_seq()),
        "taken again from byte 0, the entry finished"
    );
    assert_frames_equal(&twin, &main);
    main_served.kill();
    twin_served.kill();
}