gunnar-server 3.0.0

//! **R-14 — the boot check, end to end, through `Server::start`** (5.4, report §4.1).
//!
//! A main is re-laid empty while its twin holds the estate. Started as the real
//! server starts it (`ServeConfig.repair`), with monetize down (no fleet at all):
//!
//! * auto-fix OFF (the default): the box SAYS the data is lost and HOLDS (R-7b) —
//!   `Bootstrap.Status` answers the twin it awaits, `Bootstrap.Enroll` takes no first
//!   admin, nothing is streamed. Removing the hold marker is the admin's way out;
//! * auto-fix ON (the estate's opt-in): it recharges itself from the twin at boot,
//!   and the hold is gone once the data set is back.
#![cfg(feature = "replication")]

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

use gunnar_core::ssh_key::PrivateKey;
use gunnar_core::{NewRepo, RepoRef};
use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
use gunnar_server::repair::{self, RepairConfig};
use gunnar_server::stores::ServerStores;
use gunnar_server::twin;
use gunnar_server::CoreControlPlane;

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

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));
    }
}

/// A bare control plane for `stores` at `addr` (the old primary, the twin).
struct Served {
    _rt: tokio::runtime::Runtime,
    stop: Option<tokio::sync::oneshot::Sender<()>>,
    addr: SocketAddr,
}

fn serve(stores: &Arc<ServerStores>, addr: SocketAddr) -> Served {
    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 write_repo(stores: &ServerStores, ns: &str, name: &str) {
    if stores
        .catalog()
        .namespaces()
        .unwrap()
        .iter()
        .all(|n| n.name != ns)
    {
        stores.catalog().create_namespace(ns).expect("namespace");
    }
    stores
        .create_repo(NewRepo::new(RepoRef::new(ns, name).expect("legal")), false)
        .expect("repo");
}

fn mint() -> (Vec<u8>, String) {
    let pem = gunnar_core::mint_ed25519().expect("mint");
    let line = PrivateKey::from_openssh(std::str::from_utf8(&pem).unwrap())
        .expect("parse")
        .public_key()
        .to_openssh()
        .expect("line");
    (pem, line)
}

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

async fn enroll(at: SocketAddr, line: &str) -> Result<gunnar_proto::EnrollResponse, tonic::Status> {
    let mut c = gunnar_proto::bootstrap_client::BootstrapClient::connect(format!("http://{at}"))
        .await
        .expect("connect");
    c.enroll(gunnar_proto::EnrollRequest {
        name: "alice".into(),
        openssh_public_key: line.to_owned(),
    })
    .await
    .map(|r| r.into_inner())
}

/// A re-laid main, started as `gunnar serve` starts it, with the repair wired in.
fn start_main(
    rt: &tokio::runtime::Runtime,
    stores: &Arc<ServerStores>,
    restore_key: &str,
    cfg: RepairConfig,
) -> (gunnar_server::Server, SocketAddr) {
    let config = gunnar_server::ServeConfig {
        no_grpc: false,
        restore_pubkeys: vec![restore_key.to_owned()],
        repair: Some(cfg),
        ..gunnar_server::ServeConfig::ephemeral()
    };
    let server = rt
        .block_on(gunnar_server::Server::start(Arc::clone(stores), config))
        .expect("the main starts");
    let at = server.grpc_addr().expect("the control plane listens");
    (server, at)
}

#[test]
fn the_boot_check_holds_a_relaid_main_and_recharges_it_only_when_the_estate_opted_in() {
    let (old_addr, twin_addr) = (free_addr(), free_addr());

    // ── an estate: a primary with data, its twin following it ──
    let pd = tempfile::tempdir().unwrap();
    let primary = Arc::new(ServerStores::open(pd.path()).expect("primary"));
    write_repo(&primary, "alice", "one");
    write_repo(&primary, "bob", "two");
    let served_primary = serve(&primary, old_addr);
    let td = tempfile::tempdir().unwrap();
    let twin_stores = Arc::new(ServerStores::open(td.path()).expect("twin stores"));
    twin::become_twin(&twin_stores, &old_addr.to_string()).expect("a fresh box becomes a twin");
    twin::start_follower_every(Arc::clone(&twin_stores), Duration::from_millis(200));
    wait(
        "the twin levels with the primary",
        Duration::from_secs(30),
        || {
            let live = primary.txlog().head_seq();
            twin::status(td.path())
                .filter(|s| {
                    s.state == twin::State::Following
                        && s.applied_seq == live
                        && s.primary_head_seq == live
                        && twin_stores.txlog().head_seq() == live
                })
                .map(|_| ())
        },
    );

    // ── the main dies; the twin finds it gone ──
    served_primary.kill();
    drop(primary);
    wait(
        "the twin finds its primary gone",
        Duration::from_secs(10),
        || {
            twin::status(td.path())
                .filter(|s| s.state == twin::State::Unreachable)
                .map(|_| ())
        },
    );
    let head = twin_stores.txlog().head_seq();
    let served_twin = serve(&twin_stores, twin_addr);

    let kd = tempfile::tempdir().unwrap();
    let (pem, key_line) = mint();
    let key_path = kd.path().join("repair.key");
    std::fs::write(&key_path, &pem).expect("the repair key");
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .expect("rt");
    let cfg = RepairConfig {
        auto: false,
        key: Some(key_path),
        twin: Some(twin_addr.to_string()),
        fleet: None,
    };

    // ── 1. auto-fix OFF: the box holds, says its twin, takes no first admin ──
    let md = tempfile::tempdir().unwrap();
    let main = Arc::new(ServerStores::open(md.path()).expect("the re-laid main"));
    let (server, at) = start_main(&rt, &main, &key_line, cfg.clone());
    let st = wait(
        "the boot check holds the box",
        Duration::from_secs(30),
        || {
            let st = rt.block_on(status(at));
            (st.recharge_from == twin_addr.to_string()).then_some(st)
        },
    );
    assert!(st.window_open && !st.restoring, "{st:?}");
    assert_eq!(
        st.log_head_seq,
        main.txlog().head_seq(),
        "nothing was streamed: auto-fix is off"
    );
    assert!(repair::data_dead(&main));
    let (_, admin) = mint();
    let refused = rt
        .block_on(enroll(at, &admin))
        .expect_err("a held box takes no first admin");
    assert_eq!(
        refused.code(),
        tonic::Code::FailedPrecondition,
        "{refused:?}"
    );
    assert!(
        refused.message().contains(gunnar_repair::DATA_LOST_LINE)
            && refused.message().contains(repair::HOLD_REL),
        "{}",
        refused.message()
    );
    // The admin's way out: remove the marker; then the box is an ordinary fresh one.
    std::fs::remove_file(md.path().join(repair::HOLD_REL)).expect("the marker");
    assert_eq!(rt.block_on(status(at)).recharge_from, "");
    rt.block_on(enroll(at, &admin))
        .expect("enrols once the hold is lifted by hand");
    rt.block_on(server.shutdown()).expect("shutdown");
    println!("ROW boot-hold: auto-fix off → {at} held for {twin_addr} (Enroll FailedPrecondition, Status.recharge_from) until the marker went");

    // ── 2. auto-fix ON, monetize down: it recharges itself at boot, and nothing holds it after ──
    let ad = tempfile::tempdir().unwrap();
    let auto_main = Arc::new(ServerStores::open(ad.path()).expect("another re-laid main"));
    let (server, at) = start_main(
        &rt,
        &auto_main,
        &key_line,
        RepairConfig { auto: true, ..cfg },
    );
    wait(
        "the boot check recharges the box",
        Duration::from_secs(60),
        || (rt.block_on(status(at)).log_head_seq == head).then_some(()),
    );
    wait("the restore settles", Duration::from_secs(30), || {
        (!twin::is_restoring(ad.path())).then_some(())
    });
    assert!(
        !repair::data_dead(&auto_main),
        "the main has its data set back"
    );
    assert_eq!(
        repair::held(&auto_main),
        None,
        "a recharged box is held by nothing"
    );
    assert_eq!(rt.block_on(status(at)).recharge_from, "");
    rt.block_on(server.shutdown()).expect("shutdown");
    println!("ROW boot-auto-e2e: auto-fix on, Server::start, no monetize → {at} recharged from {twin_addr} to {head}");

    served_twin.kill();
    twin::stop_follower();
}