gunnar-server 2.0.0

//! **The guardian fires by queuing a signed order; the estate pulls it** (5.4,
//! owner 2026-10-10 08:25 UTC).
//!
//! Monetize never calls into an estate. Here a fake guardian — a real
//! `monetize.v1.Fleet` server on loopback, holding the monetize key — queues
//! "recharge from twin, now" for the `main` of estate `acme`. The re-laid main's
//! [`gunnar_server::repair::pull_loop`] reports to it, pulls the order, verifies it
//! under the monetize key, runs the ONE `recharge_from_twin` (gunnar-repair) over
//! the real wire against the twin, and acks it in its next Report.
//!
//! Before that, on the same estate:
//!
//! * the **boot check** with auto-fix OFF (the default) says the data is lost and
//!   that a recharge is available — and touches nothing;
//! * an order signed by a key that is NOT monetize's is never run and never acked.
//!
//! After it, a second re-laid main whose estate opted into auto-fix recharges
//! itself at boot, with monetize down (no fleet at all).
#![cfg(feature = "fleet-pull")]

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

use ed25519_dalek::{Signer, SigningKey, Verifier};
use gunnar_core::ssh_key::PrivateKey;
use gunnar_core::{NewRepo, RepoRef};
use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
use gunnar_repair::exec::BootStep;
use gunnar_repair::fleet::{self, pb};
use gunnar_repair::fleet::{WireAck as OrderAck, WireOrder as SignedOrder};
use gunnar_repair::Finding;
use monetize_embed::fleet::{fleet_call_message, order_message};
use gunnar_server::repair::{self, FleetConfig, RepairConfig, ServerEstate};
use gunnar_server::stores::ServerStores;
use gunnar_server::twin;
use gunnar_server::CoreControlPlane;

const MONETIZE_SEED: [u8; 32] = [7u8; 32];

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 control plane for `stores` at `addr`, until the sender fires. `restore_keys`:
/// the keys a re-laid main was fed to take a restore from.
struct Served {
    rt: tokio::runtime::Runtime,
    stop: Option<tokio::sync::oneshot::Sender<()>>,
    addr: SocketAddr,
}

fn serve(stores: &Arc<ServerStores>, addr: SocketAddr, restore_keys: Vec<String>) -> Served {
    let plane: Arc<dyn gunnar_grpc::ControlPlane> = Arc::new(CoreControlPlane::new(Arc::clone(stores)).with_restore_keys(restore_keys));
    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, 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 status(&self) -> gunnar_proto::BootstrapStatusResponse {
        let url = format!("http://{}", self.addr);
        self.rt.block_on(async move {
            let mut c = gunnar_proto::bootstrap_client::BootstrapClient::connect(url).await.expect("connect");
            c.status(gunnar_proto::BootstrapStatusRequest {}).await.expect("Bootstrap.Status").into_inner()
        })
    }
}

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

/// The estate's repair key: the private half on disk (what `--repair-key` names),
/// the public line the re-laid main is fed as a restore key.
fn repair_key(dir: &std::path::Path) -> (std::path::PathBuf, PrivateKey, String) {
    let pem = gunnar_core::mint_ed25519().expect("mint");
    let key = PrivateKey::from_openssh(std::str::from_utf8(&pem).unwrap()).expect("parse");
    let line = key.public_key().to_openssh().expect("public line");
    let path = dir.join("repair.key");
    std::fs::write(&path, &pem).expect("write the key");
    (path, key, line)
}

fn now_ms() -> u64 {
    twin::now_ms().max(0) as u64
}

fn sign(seed: [u8; 32], seq: u64, role: &str, kind: &str, body: serde_json::Value) -> pb::FleetOrder {
    let now = now_ms();
    let mut o = SignedOrder {
        seq,
        product: "gunnar".into(),
        instance: "acme".into(),
        role: role.into(),
        kind: kind.into(),
        body: serde_json::to_string(&body).unwrap(),
        issued_unix_ms: now,
        not_after_unix_ms: now + 15 * 60_000,
        signature: Vec::new(),
    };
    o.signature = SigningKey::from_bytes(&seed).sign(&order_message(&o)).to_bytes().to_vec();
    pb::FleetOrder { seq: o.seq, product: o.product, instance: o.instance, role: o.role, kind: o.kind, body: o.body, issued_unix_ms: o.issued_unix_ms, not_after_unix_ms: o.not_after_unix_ms, signature: o.signature }
}

/// **The fake guardian**: monetize's `Fleet`, holding a queue of signed orders and
/// every report a box sent it. It checks each call's signature under the key the
/// box presented (TOFU, as monetize does inside a pin window).
#[derive(Default)]
struct Guardian {
    queue: Mutex<VecDeque<pb::FleetOrder>>,
    reports: Mutex<Vec<fleet::Report>>,
    pinned: Mutex<Option<[u8; 32]>>,
    orders_calls: Mutex<u64>,
}

impl Guardian {
    fn heard(&self, req: &pb::FleetSigned) -> Result<(), tonic::Status> {
        let mut pinned = self.pinned.lock().unwrap();
        if pinned.is_none() {
            let k: [u8; 32] = req.box_key.as_slice().try_into().map_err(|_| tonic::Status::unauthenticated("no box key"))?;
            *pinned = Some(k);
        }
        let key = ed25519_dalek::VerifyingKey::from_bytes(pinned.as_ref().unwrap()).map_err(|_| tonic::Status::unauthenticated("bad key"))?;
        let call = fleet::FleetCall { product: req.product.clone(), instance: req.instance.clone(), role: req.role.clone(), call: req.call.clone(), body: req.body.clone(), at_unix_ms: req.at_unix_ms };
        let sig = ed25519_dalek::Signature::from_slice(&req.signature).map_err(|_| tonic::Status::unauthenticated("bad signature"))?;
        key.verify(&fleet_call_message(&call), &sig).map_err(|_| tonic::Status::unauthenticated("not the box's signature"))
    }

    fn acks(&self) -> Vec<OrderAck> {
        self.reports.lock().unwrap().iter().flat_map(|r| r.acked.clone()).collect()
    }
}

struct Shared(Arc<Guardian>);

#[tonic::async_trait]
impl pb::fleet_server::Fleet for Shared {
    async fn report(&self, req: tonic::Request<pb::FleetSigned>) -> Result<tonic::Response<pb::FleetReportAnswer>, tonic::Status> {
        let req = req.into_inner();
        self.0.heard(&req)?;
        let report: fleet::Report = serde_json::from_str(&req.body).map_err(|e| tonic::Status::invalid_argument(e.to_string()))?;
        let acked = report.acked.iter().map(|a| a.seq).collect();
        self.0.reports.lock().unwrap().push(report);
        Ok(tonic::Response::new(pb::FleetReportAnswer { acked, next_report_ms: 200 }))
    }

    async fn orders(&self, req: tonic::Request<pb::FleetSigned>) -> Result<tonic::Response<pb::FleetOrdersAnswer>, tonic::Status> {
        let req = req.into_inner();
        self.0.heard(&req)?;
        *self.0.orders_calls.lock().unwrap() += 1;
        let after: fleet::OrdersAfter = serde_json::from_str(&req.body).map_err(|e| tonic::Status::invalid_argument(e.to_string()))?;
        let orders = self.0.queue.lock().unwrap().iter().filter(|o| o.seq > after.after && o.role == req.role).cloned().collect();
        Ok(tonic::Response::new(pb::FleetOrdersAnswer { orders }))
    }

    async fn ask(&self, req: tonic::Request<pb::FleetSigned>) -> Result<tonic::Response<pb::FleetAskAnswer>, tonic::Status> {
        self.0.heard(&req.into_inner())?;
        Ok(tonic::Response::new(pb::FleetAskAnswer::default()))
    }
}

fn guardian(rt: &tokio::runtime::Runtime) -> (Arc<Guardian>, SocketAddr) {
    let g = Arc::new(Guardian::default());
    let addr = free_addr();
    let svc = pb::fleet_server::FleetServer::new(Shared(Arc::clone(&g)));
    rt.spawn(async move {
        let _ = tonic::transport::Server::builder().add_service(svc).serve(addr).await;
    });
    wait("the guardian listens", Duration::from_secs(10), || std::net::TcpStream::connect(addr).ok());
    (g, addr)
}

#[test]
fn the_guardian_queues_a_signed_recharge_and_the_relaid_main_pulls_and_runs_it() {
    let (old_addr, twin_addr, main_addr, auto_addr) = (free_addr(), free_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, Vec::new());

    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));
    // Level with the primary's LIVE head, as the twin itself says it: a twin a step
    // behind is a stale source, and the review refuses it (rightly).
    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's follower finds it gone); the provider lays a fresh box ──
    served_primary.kill();
    drop(primary);
    // The twin finds it gone within one poll; until then it still "hears" it, and a
    // recharge beside a primary the twin hears is refused `not_fenced` (R-13).
    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, Vec::new());

    let kd = tempfile::tempdir().unwrap();
    let (key_path, key, key_line) = repair_key(kd.path());
    let md = tempfile::tempdir().unwrap();
    let main = Arc::new(ServerStores::open(md.path()).expect("the re-laid main"));
    let served_main = serve(&main, main_addr, vec![key_line.clone()]);
    assert!(repair::data_dead(&main), "a re-laid main's data set is empty");

    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.clone()), twin: Some(twin_addr.to_string()), fleet: None };

    // ── 1. the boot check, auto-fix OFF: it SAYS, and touches nothing ──
    let report = rt.block_on(repair::boot(Arc::clone(&main), cfg.clone(), main_addr.to_string())).expect("a dead main with a twin reports");
    assert!(matches!(report.finding, Finding::DataLost { .. }), "{report:?}");
    match &report.step {
        BootStep::Say { line } => assert!(line.contains("auto-fix is off") && line.contains(gunnar_repair::DATA_LOST_LINE), "{line}"),
        other => panic!("auto-fix is off by default: {other:?}"),
    }
    assert!(report.recharge.is_none());
    assert_eq!(served_main.status().log_head_seq, main.txlog().head_seq());
    assert!(repair::data_dead(&main), "still empty: nobody recharged it");
    // Held (R-7b): the box says its data is LOST, which is what the guardian reads.
    assert_eq!(repair::health(&main, &main_addr.to_string()).get("data").map(String::as_str), Some("lost"));
    assert_eq!(repair::held(&main), Some(twin_addr.to_string()));

    // ── 2. the guardian queues: a forgery first, then monetize's own order ──
    let (g, fleet_addr) = guardian(&rt);
    let body = serde_json::json!({"main": main_addr.to_string(), "twin": twin_addr.to_string(), "accept_lost": false, "epoch": 1, "case": "case-r5-4", "fenced": "the old main's machine was killed and its server read back gone (case-r5-4)"});
    g.queue.lock().unwrap().push_back(sign([9u8; 32], 1, "main", "recharge_from_twin", body.clone()));

    let fleet_cfg = FleetConfig {
        url: format!("http://{fleet_addr}"),
        ca: None,
        ca_pem: None,
        at: None,
        instance: "acme".into(),
        role: "main".into(),
        monetize_pubkey: Some(SigningKey::from_bytes(&MONETIZE_SEED).verifying_key().to_bytes()),
        every: Duration::from_secs(1),
    };
    let estate = ServerEstate { key: key.clone(), own: main_addr.to_string(), entitlements: Default::default() };
    let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);
    let pulling = rt.spawn(repair::pull_loop(Arc::clone(&main), fleet_cfg, Some(estate), repair::Entitlements::of(Arc::clone(&main)), main_addr.to_string(), stop_rx));

    wait("the box pulls the forged order", Duration::from_secs(20), || (*g.orders_calls.lock().unwrap() >= 2).then_some(()));
    assert!(g.acks().is_empty(), "a forged order is never run and never acked: {:?}", g.acks());
    assert!(repair::data_dead(&main), "the forgery did nothing");
    let first = g.reports.lock().unwrap().first().cloned().expect("the box reported");
    assert_eq!(first.health.get("data").map(String::as_str), Some("lost"), "{first:?}");
    assert_eq!(first.health.get("serving").map(String::as_str), Some("no"), "{first:?}");
    assert_eq!(first.health.get("serves_at"), Some(&main_addr.to_string()), "{first:?}");

    // monetize's own order, at the SAME seq: a forgery never advanced it.
    g.queue.lock().unwrap().clear();
    g.queue.lock().unwrap().push_back(sign(MONETIZE_SEED, 1, "main", "recharge_from_twin", body));

    let ack = wait("the box runs monetize's order and acks it in its next report", Duration::from_secs(60), || g.acks().into_iter().find(|a| a.seq == 1));
    assert!(ack.ok, "{ack:?}");
    assert!(ack.detail.contains("recharged"), "{ack:?}");
    let receipt: gunnar_repair::Receipt = serde_json::from_str(&ack.receipt_json).expect("the receipt rides the ack");
    assert_eq!(receipt.head_seq, head, "{receipt:?}");
    assert_eq!(served_main.status().log_head_seq, head, "the main holds what the twin held");
    assert!(!twin::is_restoring(md.path()));
    assert!(!repair::data_dead(&main), "the main has its data set back");
    assert_eq!(repair::held(&main), None, "and nothing holds it any more");
    assert!(std::fs::metadata(md.path().join(fleet::BOX_KEY_REL)).is_ok(), "the box's own fleet key, minted once");
    let state = std::fs::read_to_string(md.path().join(fleet::STATE_REL)).expect("pulled state");
    assert!(state.contains("\"last_seq\":1"), "{state}");

    let _ = stop_tx.send(true);
    rt.block_on(async { tokio::time::timeout(Duration::from_secs(30), pulling).await }).expect("the pull loop stops").expect("cleanly");
    let after = served_main.status().log_head_seq;
    println!("ROW fleet-order: forged order ignored; monetize's seq 1 pulled → recharged {main_addr} from {twin_addr} to {after}");
    served_main.kill();

    // ── 3. a second re-laid main whose estate opted into auto-fix: it recharges at boot, monetize down ──
    let ad = tempfile::tempdir().unwrap();
    let auto_main = Arc::new(ServerStores::open(ad.path()).expect("another re-laid main"));
    let served_auto = serve(&auto_main, auto_addr, vec![key_line]);
    let auto_cfg = RepairConfig { auto: true, ..cfg };
    let report = rt.block_on(repair::boot(Arc::clone(&auto_main), auto_cfg, auto_addr.to_string())).expect("a dead main with a twin reports");
    assert!(matches!(report.step, BootStep::Recharge { .. }), "{report:?}");
    let r = report.recharge.clone().expect("auto-fix ran it").expect("and it was not refused");
    assert_eq!(r.head_seq, head, "{r:?}");
    assert_eq!(served_auto.status().log_head_seq, head);
    println!("ROW boot-auto: auto-fix on → recharged {auto_addr} from {twin_addr} to {head} with no monetize at all");
    served_auto.kill();
    served_twin.kill();
    twin::stop_follower();
}