crates/monetize-ui-server/tests/stuck_fleet.rs
//! ★ **PROD-4 (prod ONE box 2026-10-26)**: the console's OVERVIEW and GUARDIAN pages loaded
//! forever and never showed a word. Both wait on `Guardian.Cases`; on prod each one listed the
//! UpCloud account once per silent main (four estates the box cannot reach: seconds each, one
//! after another). The box now answers Cases from what its slow look job last saw; and the
//! console bounds every fleet read: past `estates::FLEET_WITHIN` it answers from the last
//! fleet it read, with that state's age, or with an error that names the bound — never an
//! endless spinner.
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use axum::body::Body;
use axum::Router;
use http::{Request, StatusCode};
use http_body_util::BodyExt;
use monetize_proto as pb;
use monetize_proto::estate_admin_server::{EstateAdmin, EstateAdminServer};
use monetize_proto::estates_server::{Estates, EstatesServer};
use monetize_proto::guardian_server::{Guardian, GuardianServer};
use monetize_ui_server::app::App;
use monetize_ui_server::grpc::Monetize;
use monetize_ui_server::serve;
use monetize_ui_server::sse::EventHub;
use serde_json::Value;
use tonic::{Request as TReq, Response, Status};
use tower::ServiceExt;
const BEARER: &str = "the-stuck-fleet-bearer";
const ADMIN: &str = "break-glass-stuck-fleet";
/// What a page may wait, all told: the fleet's bound and a little.
const PAGE_WITHIN: Duration = Duration::from_millis(2_500);
/// The ONE monetize's fleet: four estates nothing of which answers, and a Cases that can stick.
#[derive(Default)]
struct Fleet {
stuck: AtomicBool,
asked: AtomicUsize,
}
#[derive(Clone)]
struct Svc(Arc<Fleet>);
fn estate(instance: &str, product: &str) -> pb::GuardianEstate {
let b = |role: &str, heard: u64| pb::GuardianBox { role: role.into(), heard_unix_ms: heard, silent: true, laid_ipv4: "10.6.0.9".into(), ..Default::default() };
pb::GuardianEstate { instance: instance.into(), product: product.into(), boxes: vec![b("front", 0), b("main", 5), b("twin", 5)], ..Default::default() }
}
#[tonic::async_trait]
impl Guardian for Svc {
async fn cases(&self, r: TReq<pb::GuardianCasesRequest>) -> Result<Response<pb::GuardianCasesResponse>, Status> {
self.0.asked.fetch_add(1, Ordering::SeqCst);
if self.0.stuck.load(Ordering::SeqCst) {
// A refresh that is stuck: the console's own 120 s bound is all that ends it.
tokio::time::sleep(Duration::from_secs(600)).await;
}
let want = r.into_inner().instance;
let all = vec![estate("acme2", "holger"), estate("ignalina", "gunnar"), estate("polka", "gunnar"), estate("test3-holger-rs", "holger")];
Ok(Response::new(pb::GuardianCasesResponse { estates: all.into_iter().filter(|e| want.is_empty() || e.instance == want).collect() }))
}
async fn press(&self, _: TReq<pb::GuardianPressRequest>) -> Result<Response<pb::GuardianCase>, Status> {
Err(Status::unimplemented("not in this mock"))
}
async fn set_auto_fix(&self, _: TReq<pb::GuardianAutoFixRequest>) -> Result<Response<pb::GuardianAutoFixResponse>, Status> {
Err(Status::unimplemented("not in this mock"))
}
async fn set_hold(&self, _: TReq<pb::GuardianHoldRequest>) -> Result<Response<pb::GuardianHoldResponse>, Status> {
Err(Status::unimplemented("not in this mock"))
}
}
/// The ONE monetize's classified estates (OVW-CLEAN): read beside the fleet, sticking with it.
#[tonic::async_trait]
impl Estates for Svc {
async fn overview(&self, _: TReq<pb::EstatesOverviewRequest>) -> Result<Response<pb::EstatesOverviewResponse>, Status> {
if self.0.stuck.load(Ordering::SeqCst) {
tokio::time::sleep(Duration::from_secs(600)).await;
}
let row = |name: &str, product: &str| pb::EstateStanding { name: name.into(), product: product.into(), standing: "installed".into(), in_fleet: true, ..Default::default() };
Ok(Response::new(pb::EstatesOverviewResponse { estates: vec![row("acme2", "holger"), row("ignalina", "gunnar"), row("polka", "gunnar"), row("test3-holger-rs", "holger")], ..Default::default() }))
}
async fn new_estate(&self, _: TReq<pb::NewEstateRequest>) -> Result<Response<pb::NewEstateResponse>, Status> {
Err(Status::unimplemented("not in this mock"))
}
}
#[tonic::async_trait]
impl EstateAdmin for Svc {
async fn draw(&self, r: TReq<pb::DrawEstateRequest>) -> Result<Response<pb::DrawEstateResponse>, Status> {
if self.0.stuck.load(Ordering::SeqCst) {
tokio::time::sleep(Duration::from_secs(600)).await;
}
let r = r.into_inner();
Ok(Response::new(pb::DrawEstateResponse { svg: format!("<svg data-estate=\"{}\"/>", r.estate), machines: Some(3) }))
}
}
fn bearer_gate(req: TReq<()>) -> Result<TReq<()>, Status> {
match req.metadata().get("authorization").and_then(|v| v.to_str().ok()) {
Some(v) if v == format!("Bearer {BEARER}") => Ok(req),
_ => Err(Status::unauthenticated("bearer missing or wrong")),
}
}
async fn console(stuck: bool) -> (Router, Arc<Fleet>) {
let (router, fleet, _) = console_with(stuck).await;
(router, fleet)
}
async fn console_with(stuck: bool) -> (Router, Arc<Fleet>, Monetize) {
let fleet = Arc::new(Fleet { stuck: AtomicBool::new(stuck), ..Default::default() });
let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let at = l.local_addr().unwrap().to_string();
tokio::spawn(tonic::transport::Server::builder().add_service(GuardianServer::with_interceptor(Svc(fleet.clone()), bearer_gate)).add_service(EstateAdminServer::with_interceptor(Svc(fleet.clone()), bearer_gate)).add_service(EstatesServer::with_interceptor(Svc(fleet.clone()), bearer_gate)).serve_with_incoming(tonic::transport::server::TcpIncoming::from(l)));
let monetize = Monetize::connect_lazy(&at, BEARER).unwrap();
let app = Arc::new(App { webauthn: None, admin_token: Some(ADMIN.into()), events: EventHub::spawn(monetize.clone(), Duration::from_secs(3600)), monetize: monetize.clone(), operator: None, ui_dir: None });
(serve::router(app), fleet, monetize)
}
/// One page's call, bounded here too so a red is a red, not a hang.
async fn get(router: &Router, path: &str) -> (StatusCode, Value, Duration) {
let (st, _, v, took) = get_h(router, path).await;
(st, v, took)
}
async fn get_h(router: &Router, path: &str) -> (StatusCode, http::HeaderMap, Value, Duration) {
let t0 = Instant::now();
let req = Request::builder().uri(path).header("authorization", format!("Bearer {ADMIN}")).body(Body::empty()).unwrap();
let res = tokio::time::timeout(Duration::from_secs(20), router.clone().oneshot(req)).await.unwrap_or_else(|_| panic!("{path} did not answer within 20 s: an endless spinner")).unwrap();
let (status, headers) = (res.status(), res.headers().clone());
let bytes = res.into_body().collect().await.unwrap().to_bytes();
let v = serde_json::from_slice(&bytes).unwrap_or_else(|_| Value::String(String::from_utf8_lossy(&bytes).into_owned()));
(status, headers, v, t0.elapsed())
}
/// Every console read that waits on the estates or the cloud (the map included: 20 s on prod).
const PAGES: [&str; 6] = ["/api/overview", "/api/estates", "/api/guardian", "/api/guardian/acme2", "/api/estates/acme2/summary", "/api/estates/acme2/map"];
/// ★ RED: a fleet that never answers is an ERROR within the bound, naming it — on every page.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_stuck_fleet_read_is_an_error_within_a_bound_never_an_endless_spinner() {
let (router, _) = console(true).await;
for path in PAGES {
let (st, v, took) = get(&router, path).await;
assert!(took < PAGE_WITHIN, "{path} took {took:?}");
assert!(!st.is_success(), "{path}: {st} {v}");
assert!(v.to_string().contains("did not answer within"), "{path}: the page is told why: {v}");
}
}
/// ★ RED: once the fleet was read, a refresh that sticks answers every page from that state,
/// with its age, within the bound.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_stuck_refresh_answers_from_the_last_fleet_with_its_age() {
let (router, fleet) = console(false).await;
for path in PAGES {
let (st, v, _) = get(&router, path).await;
assert_eq!(st, StatusCode::OK, "{path}: {v}");
assert!(v.get("stale_secs").is_none_or(|s| s.is_null()), "a fresh read is not stale: {v}");
}
fleet.stuck.store(true, Ordering::SeqCst);
for path in PAGES {
let (st, h, v, took) = get_h(&router, path).await;
assert!(took < PAGE_WITHIN, "{path} took {took:?}");
assert_eq!(st, StatusCode::OK, "{path}: {v}");
if path.ends_with("/map") {
assert!(h.get("x-monetize-stale-secs").is_some(), "the map says how old it is: {h:?}");
} else if path != "/api/estates" {
assert!(v["stale_secs"].is_u64(), "{path} says how old its state is: {v}");
}
}
let (_, v, _) = get(&router, "/api/overview").await;
assert_eq!(v["estates"].as_array().map(|a| a.len()), Some(4), "every estate is a row, at once: {v}");
}
/// ★ PROD-5: an estate that runs front, main and twin has three machines — a box this box laid
/// counts, reported or not (prod: the front never reports, the console said "Machines 2").
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn front_main_and_twin_are_three_machines() {
let (router, _) = console(false).await;
let (st, v, _) = get(&router, "/api/estates/acme2/summary").await;
assert_eq!(st, StatusCode::OK, "{v}");
assert_eq!(v["machines"], 3, "front + main + twin: {v}");
}
/// ★ RA3-1 RED (re-analysis 3): **twenty looks at a stuck fleet ask it at most once.** Each
/// Overview and each estate row's summary used to start its own whole `Guardian.Cases`, each
/// left running for up to 120 s; now the one kept read is joined.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn twenty_looks_at_a_stuck_fleet_ask_it_at_most_once() {
let (router, fleet) = console(false).await;
let (st, v, _) = get(&router, "/api/overview").await;
assert_eq!(st, StatusCode::OK, "{v}");
// The keeper's start-up ask may still be landing; let it, then the fleet sticks.
tokio::time::sleep(Duration::from_millis(300)).await;
fleet.stuck.store(true, Ordering::SeqCst);
let before = fleet.asked.load(Ordering::SeqCst);
let looks = (0..10).flat_map(|_| ["/api/overview", "/api/estates/acme2/summary"]).map(|path| {
let router = router.clone();
async move { get(&router, path).await }
});
for (st, v, took) in futures_util::future::join_all(looks).await {
assert_eq!(st, StatusCode::OK, "{v}");
assert!(took < PAGE_WITHIN, "took {took:?}");
}
let asked = fleet.asked.load(Ordering::SeqCst) - before;
assert!(asked <= 1, "20 looks at a stuck fleet asked Guardian.Cases {asked} times");
}
/// ★ RA3-1 RED: **the fleet is kept by its jera job, not by page looks**: started with the
/// console, it has read the fleet before any page asks, so the first page is answered at once.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_fleet_is_read_by_its_job_before_any_page_asks() {
let (router, fleet, monetize) = console_with(false).await;
monetize.keep_the_fleet().expect("a runtime");
let t0 = Instant::now();
while fleet.asked.load(Ordering::SeqCst) == 0 && t0.elapsed() < Duration::from_secs(5) {
tokio::time::sleep(Duration::from_millis(20)).await;
}
assert_eq!(fleet.asked.load(Ordering::SeqCst), 1, "the job read the fleet with no page asking");
// Its answer lands; only then does the fleet stick.
tokio::time::sleep(Duration::from_millis(300)).await;
fleet.stuck.store(true, Ordering::SeqCst);
let (st, v, took) = get(&router, "/api/overview").await;
assert_eq!(st, StatusCode::OK, "the job's kept fleet answers the first page: {v}");
assert!(took < PAGE_WITHIN && v["stale_secs"].is_u64(), "{v}");
}