gunnar-server 3.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 gunnar_server::repair::{self, FleetConfig, RepairConfig, ServerEstate};
use gunnar_server::stores::ServerStores;
use gunnar_server::twin;
use gunnar_server::CoreControlPlane;
use monetize_embed::fleet::{fleet_call_message, order_message};

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