gunnar-server 2.0.0

The gunnar git server: SSH and smart-HTTP git transports and the gRPC control plane. The appliance half, extracted from gunnar-cli so the server code lives in its own crate.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174

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