crates/gunnar-server/tests/twin_promotion_rpc.rs
//! **A twin is promoted and fenced over the real wire** (lane OX-WIRED2): `Mirror.SetRole`
//! driven through the real gRPC router (the signed-request gate, no insecure mode) over the
//! real `CoreControlPlane` with its replication slot, on a real store:
//!
//! * the server's admin makes the box a STANDBY (it then refuses client writes, on disk) and
//! promotes it back to PRIMARY; `ReplicationStatus` reads each role back;
//! * a person who is not the server's admin is refused both, and the role does not move;
//! * an unsigned call is refused, and an unknown role is refused by name rather than read
//! as PRIMARY.
//!
//! Before this, the promotion was driven only below the router
//! (`gunnar-cli/tests/replication.rs` calls the plane's `replication_set_role`).
#![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::{Principal, PrincipalKind};
use gunnar_grpc::client_signer::{authed, plain_channel};
use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
use gunnar_proto::mirror_client::MirrorClient;
use gunnar_proto::{ReplicationRole, ReplicationStatusRequest, SetRoleRequest};
use gunnar_server::stores::ServerStores;
use gunnar_server::CoreControlPlane;
use tonic::Code;
fn free_addr() -> SocketAddr {
let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
l.local_addr().expect("addr")
}
fn mint() -> (PrivateKey, String) {
let pem = gunnar_core::mint_ed25519().expect("mint");
let k = PrivateKey::from_openssh(std::str::from_utf8(&pem).unwrap()).expect("parse");
let line = k.public_key().to_openssh().expect("line");
(k, line)
}
/// What the box's replication config on disk says now: does it refuse client writes?
fn frozen_on_disk(root: &std::path::Path) -> bool {
gunnar_server::mirror::ReplicationConfig::load(root)
.expect("the replication config reads")
.role
.refuses_client_writes()
}
#[test]
fn the_admin_fences_and_promotes_the_box_over_the_real_router_and_nobody_else_does() {
let d = tempfile::tempdir().unwrap();
let stores = Arc::new(ServerStores::open(d.path()).expect("stores"));
let (admin, admin_line) = mint();
stores
.auth()
.bootstrap_admin(&admin_line)
.expect("the first admin");
stores
.auth()
.add_principal(Principal::new("bob", "bob", PrincipalKind::User).expect("legal id"))
.expect("principal");
let (bob, bob_line) = mint();
stores
.auth()
.register_key("bob", &bob_line, None)
.expect("bob's SSH key");
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.expect("rt");
let addr = free_addr();
let bus = Arc::new(EventBus::new(16));
let slot = gunnar_server::mirror::ReplicationSlot::new(
Arc::clone(&stores),
Arc::clone(&bus),
rt.handle().clone(),
);
let plane: Arc<dyn gunnar_grpc::ControlPlane> =
Arc::new(CoreControlPlane::new(Arc::clone(&stores)).with_replication(slot));
let ctx = ServiceContext::new(plane, bus, ServerConfig::default());
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;
});
let t0 = Instant::now();
while std::net::TcpStream::connect(addr).is_err() {
assert!(t0.elapsed() < Duration::from_secs(10), "the box listens");
std::thread::sleep(Duration::from_millis(25));
}
let at = addr.to_string();
let root = d.path().to_path_buf();
rt.block_on(async {
let mut mirror = MirrorClient::new(plain_channel(&at).expect("a channel"));
let role = |r: &gunnar_proto::ReplicationStatusResponse| {
ReplicationRole::try_from(r.role).expect("a known role")
};
macro_rules! role_now {
() => {
role(
&mirror
.replication_status(authed(&admin, ReplicationStatusRequest::default()))
.await
.expect("Mirror.ReplicationStatus")
.into_inner(),
)
};
}
assert_eq!(
role_now!(),
ReplicationRole::Primary,
"a fresh box is its own primary"
);
assert!(!frozen_on_disk(&root));
// Nobody else moves it: unsigned, and a person who is not the server's admin.
let anon = mirror
.set_role(tonic::Request::new(SetRoleRequest {
role: ReplicationRole::Standby as i32,
}))
.await
.unwrap_err();
assert_eq!(anon.code(), Code::Unauthenticated, "{anon:?}");
let refused = mirror
.set_role(authed(
&bob,
SetRoleRequest {
role: ReplicationRole::Standby as i32,
},
))
.await
.unwrap_err();
assert_eq!(refused.code(), Code::PermissionDenied, "{refused:?}");
assert_eq!(
role_now!(),
ReplicationRole::Primary,
"the refusal moved nothing"
);
// The admin fences it: a standby, which refuses client writes, on disk.
let fenced = mirror
.set_role(authed(
&admin,
SetRoleRequest {
role: ReplicationRole::Standby as i32,
},
))
.await
.expect("Mirror.SetRole(STANDBY)")
.into_inner();
assert_eq!(role(&fenced), ReplicationRole::Standby, "{fenced:?}");
assert!(frozen_on_disk(&root), "a standby refuses client writes");
// bob cannot promote it either; an unknown role is refused by name, not read as PRIMARY.
let refused = mirror
.set_role(authed(
&bob,
SetRoleRequest {
role: ReplicationRole::Primary as i32,
},
))
.await
.unwrap_err();
assert_eq!(refused.code(), Code::PermissionDenied, "{refused:?}");
let unknown = mirror
.set_role(authed(&admin, SetRoleRequest { role: 99 }))
.await
.unwrap_err();
assert_eq!(unknown.code(), Code::InvalidArgument, "{unknown:?}");
assert!(
unknown.message().contains("unknown replication role 99"),
"{unknown:?}"
);
assert_eq!(role_now!(), ReplicationRole::Standby, "still the standby");
// The admin promotes it: the primary again, taking writes.
let promoted = mirror
.set_role(authed(
&admin,
SetRoleRequest {
role: ReplicationRole::Primary as i32,
},
))
.await
.expect("Mirror.SetRole(PRIMARY)")
.into_inner();
assert_eq!(role(&promoted), ReplicationRole::Primary, "{promoted:?}");
assert!(
!frozen_on_disk(&root),
"the promoted box takes client writes"
);
});
let _ = tx.send(());
}