nordisk/gunnar

crates/gunnar-grpc/tests/keepalive_under_load.rs

//! **A client's keepalive PING is answered while every handler is busy**
//! (R5.1-G, 2026-10-07).
//!
//! The control-plane handlers make synchronous calls into the store — a disk
//! walk, a txlog fsync, a snapshot — and each one holds the worker it runs on
//! until it returns. gunnar served them on the same runtime that drives the
//! HTTP/2 connections, with `cores / 4` (at least 2) workers. Two slow calls at
//! once held both workers, so nothing polled the connection: the client's
//! keepalive PING (the console's every 10 s, timeout 5 s; monetize's door the
//! same) went unanswered, the client declared the connection dead, and every
//! call in flight on it failed — the slow ones AND the quick ones behind them.
//! On test2 that was the console's `KeepAliveTimedOut` stalls and a paid
//! `Entitlement.Set` that came back as an `Internal` stream reset.
//!
//! The test: a real router on a 2-worker runtime, a handler runtime of 2
//! workers, `Bootstrap.Status` stalled 3 s on its thread, and a client that
//! PINGs every 500 ms and gives up after 1 s without an answer. Four calls at
//! once, on ONE connection, must all answer — slowly (they queue behind each
//! other on the handler workers), but every one of them.

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

use gunnar_grpc::{
    router, ControlPlane, EventBus, MemoryControlPlane, ServerConfig, ServiceContext,
};
use gunnar_proto::bootstrap_client::BootstrapClient;
use gunnar_proto::BootstrapStatusRequest;
use tonic::transport::Channel;
use tonic::Request;

/// Each `Bootstrap.Status` holds its thread this long.
const STALL: Duration = Duration::from_secs(3);
/// The client's keepalive: a PING every `PING_EVERY`, the connection declared
/// dead `PING_TIMEOUT` after one goes unanswered — both far shorter than
/// `STALL`, so a server that cannot answer a PING during a stall loses the
/// connection inside the first one.
const PING_EVERY: Duration = Duration::from_millis(500);
const PING_TIMEOUT: Duration = Duration::from_secs(1);
/// Calls at once: twice the handler workers, so every worker is held and
/// calls wait for one.
const CALLS: usize = 4;
const WORKERS: usize = 2;

fn free_addr() -> SocketAddr {
    TcpListener::bind("127.0.0.1:0")
        .expect("a loopback port")
        .local_addr()
        .expect("its address")
}

fn await_listening(addr: SocketAddr) {
    let deadline = Instant::now() + Duration::from_secs(10);
    while Instant::now() < deadline {
        if std::net::TcpStream::connect_timeout(&addr, Duration::from_millis(200)).is_ok() {
            return;
        }
        std::thread::sleep(Duration::from_millis(20));
    }
    panic!("the router never bound {addr}");
}

fn runtime(workers: usize, name: &str) -> tokio::runtime::Runtime {
    tokio::runtime::Builder::new_multi_thread()
        .worker_threads(workers)
        .thread_name(name)
        .enable_all()
        .build()
        .expect("runtime")
}

#[test]
fn a_ping_is_answered_and_every_call_survives_while_all_handler_workers_are_held() {
    let addr = free_addr();
    let plane = Arc::new(MemoryControlPlane::with_repo("nordisk", "gunnar"));
    plane.stall_bootstrap_state(STALL);
    let control: Arc<dyn ControlPlane> = plane;

    // The production shape: the connections on one runtime, the handlers on
    // another — both as small as `gunnar-server` would ever make them.
    let server = runtime(WORKERS, "control-conn");
    let handlers = runtime(WORKERS, "control-handler");
    let ctx = ServiceContext::new(
        control,
        Arc::new(EventBus::new(64)),
        ServerConfig::default(),
    )
    .with_handler_runtime(handlers.handle().clone());
    server.spawn(async move {
        let _ = router(ctx, || 0).serve(addr).await;
    });
    await_listening(addr);

    let client = runtime(2, "client");
    let started = Instant::now();
    let answers: Vec<Result<(), String>> = client.block_on(async move {
        let channel: Channel = Channel::from_shared(format!("http://{addr}"))
            .expect("a valid endpoint")
            .http2_keep_alive_interval(PING_EVERY)
            .keep_alive_timeout(PING_TIMEOUT)
            .keep_alive_while_idle(true)
            .connect()
            .await
            .expect("connect to the router");
        let calls = (0..CALLS).map(|_| {
            let mut c = BootstrapClient::new(channel.clone());
            tokio::spawn(async move {
                c.status(Request::new(BootstrapStatusRequest::default()))
                    .await
                    .map(|_| ())
                    .map_err(|s| format!("{:?}: {}", s.code(), s.message()))
            })
        });
        let mut out = Vec::new();
        for call in calls.collect::<Vec<_>>() {
            out.push(call.await.expect("the call task"));
        }
        out
    });
    let took = started.elapsed();

    let failed: Vec<&String> = answers.iter().filter_map(|a| a.as_ref().err()).collect();
    assert!(
        failed.is_empty(),
        "{} of {CALLS} calls failed while the handlers were busy — the server did not answer \
         the client's keepalive PING (every {PING_EVERY:?}, dead after {PING_TIMEOUT:?}) during a \
         {STALL:?} synchronous call, so the client killed the connection and every call on it: \
         {failed:?}",
        failed.len()
    );
    // Not vacuous: the stall really held the workers (two rounds of STALL).
    assert!(
        took >= STALL * 2,
        "the calls answered in {took:?}: the stall ({STALL:?} each, {CALLS} calls on {WORKERS} \
         workers) never held the workers, so this proved nothing"
    );

    drop(client);
    server.shutdown_timeout(Duration::from_secs(1));
    handlers.shutdown_timeout(Duration::from_secs(1));
}