From 0b6714dc37574ada926872d976d6766eb014a76b Mon Sep 17 00:00:00 2001 From: Eric Moore Date: Tue, 8 Sep 2026 09:57:25 -0500 Subject: [PATCH 1/3] The stop is legible from the node's own log: a serve marker names a stop that skipped the door, the drain reports its numbers, and every stop request says who asked (#568) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CIRISServer#568 read as "the node re-composes after announce and drops the response". It was the agent's setup-complete hand-off 130 ms later, and the node log could not say so: a serve that is replaced or killed leaves no line, and the stop that IS clean did not say what it drained. - `serve_marker`: `/serving.json` (pid, instance_id, started_at, listen_addr, key_id) written the instant the read API is bound, cleared only after it has drained. The next boot inspects it first: a present marker is WARN "the previous serve did not stop through the shutdown door — replaced, killed or crashed; any response in flight was lost to its caller", ERROR if that pid is still alive (the bind will fail), recorded as a compose_status mark, then cleared. - lens-core read API: an in-flight request counter over the whole router, host routes included; `shutdown()` logs "draining — in_flight=N" and "stopped — drained N, still_in_flight 0, took_ms" so a dropped connection can be placed on the right side of the door from the log alone. - `node_control::request_shutdown_from(origin)`: every stop request logs its origin; `shutdown_node()` names the embedding host. Gates: the marker brackets the listener (inspect < listener_bound < write < drain < clear); every stop request says who asked; unit tests for the marker (clean, own pid, dead pid, damaged) and the counter. Co-Authored-By: Claude Fable 5.1 --- crates/ciris-lens-core/src/role/node.rs | 104 ++++++++- src/compose.rs | 18 ++ src/lib.rs | 1 + src/node_control.rs | 15 +- src/serve_marker.rs | 287 ++++++++++++++++++++++++ tests/shutdown_signals.rs | 49 ++++ 6 files changed, 471 insertions(+), 3 deletions(-) create mode 100644 src/serve_marker.rs diff --git a/crates/ciris-lens-core/src/role/node.rs b/crates/ciris-lens-core/src/role/node.rs index 86f610d6..6ac8bb3d 100644 --- a/crates/ciris-lens-core/src/role/node.rs +++ b/crates/ciris-lens-core/src/role/node.rs @@ -960,13 +960,64 @@ impl ReadApiHandle { } /// Signal the read-API server to stop and await its task. + /// + /// The stop is a DRAIN, not a cut: accepting stops at once, every request + /// already inside runs to completion and its response is written, then + /// the listener closes. The two lines this logs carry the numbers — how + /// many were inside when the stop was asked, how many were still inside + /// when the task returned (expected 0), and how long the drain took — so + /// a caller's dropped connection can be placed on the right side of this + /// door from the log alone (CIRISServer#568). pub async fn shutdown(self) -> Result<(), NodeError> { + let inside = in_flight(); + let t0 = std::time::Instant::now(); + tracing::info!( + listen_addr = %self.listen_addr, + in_flight = inside, + "lens read API draining — accepting stops now; requests already inside complete \ + and their responses are written before the listener closes" + ); let _ = self.http_shutdown_tx.send(true); let _ = self.http_join.await; + tracing::info!( + listen_addr = %self.listen_addr, + drained = inside, + still_in_flight = in_flight(), + took_ms = t0.elapsed().as_millis() as u64, + "lens read API stopped — listener closed; no response was cut" + ); Ok(()) } } +/// HTTP requests currently inside the read API: accepted, response not yet +/// fully written. Counted by [`track_in_flight`] on every route, including the +/// routes a host merges in. Read at shutdown so the drain is a number. +pub static IN_FLIGHT: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); + +/// Requests inside the read API right now. +#[must_use] +pub fn in_flight() -> usize { + IN_FLIGHT.load(std::sync::atomic::Ordering::SeqCst) +} + +/// The counting layer. A guard, not a pair of calls, so a handler that panics +/// or is cancelled still decrements. +pub async fn track_in_flight( + request: axum::extract::Request, + next: axum::middleware::Next, +) -> Response { + struct Inside; + impl Drop for Inside { + fn drop(&mut self) { + IN_FLIGHT.fetch_sub(1, std::sync::atomic::Ordering::SeqCst); + } + } + IN_FLIGHT.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let _inside = Inside; + next.run(request).await +} + // ─── LensCore::node ──────────────────────────────────────────────── impl LensCore { @@ -1042,7 +1093,12 @@ impl LensCore { api_root: ux.api_root.clone(), fidelity, }; - let router = extra.merge(build_read_router(state)); + // The in-flight counter wraps the WHOLE router — the host's merged + // routes included — so a drain at shutdown counts every response it + // is about to finish writing (CIRISServer#568). + let router = extra + .merge(build_read_router(state)) + .layer(axum::middleware::from_fn(track_in_flight)); let (http_shutdown_tx, mut http_shutdown_rx) = watch::channel(false); // Bind SYNCHRONOUSLY, before spawning the accept loop (CIRISServer#279). // The old shape bound inside the spawned task and swallowed the error @@ -1071,7 +1127,7 @@ impl LensCore { }) .await .ok(); - tracing::info!(%listen_addr, "lens read API stopped"); + tracing::debug!(%listen_addr, "lens read API accept loop returned"); }); Ok(ReadApiHandle { http_shutdown_tx, @@ -1805,4 +1861,48 @@ mod tests { ); assert!(v.get("row_fidelity").is_none(), "{v}"); } + + /// The in-flight counter follows a request from accept to the written + /// response, and a stop that reads it sees the truth (CIRISServer#568). + #[tokio::test] + async fn in_flight_counts_a_request_from_accept_to_response() { + use axum::{routing::get, Router}; + use tower::ServiceExt as _; + static GATE: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(0); + let app = Router::new() + .route( + "/slow", + get(|| async { + // Hold the request inside until the test has looked. + let _permit = GATE.acquire().await.expect("gate"); + "done" + }), + ) + .layer(axum::middleware::from_fn(track_in_flight)); + let before = in_flight(); + let call = tokio::spawn( + app.oneshot( + axum::http::Request::builder() + .uri("/slow") + .body(axum::body::Body::empty()) + .unwrap(), + ), + ); + // Wait until the request is inside the handler. + for _ in 0..200 { + if in_flight() == before + 1 { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + assert_eq!(in_flight(), before + 1, "one request is inside"); + GATE.add_permits(1); + let resp = call.await.unwrap().unwrap(); + assert_eq!(resp.status(), axum::http::StatusCode::OK); + assert_eq!( + in_flight(), + before, + "the response was written and the count fell back" + ); + } } diff --git a/src/compose.rs b/src/compose.rs index b2b1b2bf..df65ae33 100644 --- a/src/compose.rs +++ b/src/compose.rs @@ -98,6 +98,12 @@ pub async fn serve_with_adapter(cfg: ServerConfig, adapter: Arc) -> // (CIRISServer#557). crate::graph_config::invalidate(); + // What did the PREVIOUS serve on this home leave? A marker still present + // means it did not stop through the door — replaced, killed or crashed — + // and this is where that becomes a line in THIS log rather than a guess + // from a caller's dropped connection (CIRISServer#568). + let _previous_serve = crate::serve_marker::inspect_at_boot(&cfg.data_dir); + // ── RNG startup health-check (CIRISServer#283 finding 2) ────────────────── // Arm the SP 800-90B latch ONCE at boot so `ciris_crypto::random::fill`'s // fail-secure gate is live: if the OS entropy source is producing detectably @@ -1591,6 +1597,13 @@ pub async fn serve_with_adapter(cfg: ServerConfig, adapter: Arc) -> }; crate::compose_status::mark("listener_bound"); tracing::info!(read_api = %read.listen_addr(), "read API up — GET /lens/api/v1/* + GET /v1/identity"); + // From this instant a response can be in flight: write the serve + // marker, cleared only by a stop that drains the read API + // (CIRISServer#568). A write failure is logged, not fatal — the marker + // is legibility, not a lock. + if let Err(e) = crate::serve_marker::write(&cfg.data_dir, read.listen_addr(), &cfg.key_id) { + tracing::warn!(error = %e, "could not write the serve marker"); + } // #279: the listener is now guaranteed BOUND here (lens-core binds // synchronously before spawning the accept loop and a bind failure is // the `?` above). Stamp the milestone so compose_status distinguishes @@ -1793,8 +1806,13 @@ pub async fn serve_with_adapter(cfg: ServerConfig, adapter: Arc) -> } if let Some(read) = read { + // Drains: every request already inside completes and its response is + // written before the listener closes; the handle logs the count. read.shutdown().await.context("shutdown lens read API")?; } + // The listener is closed and nothing is in flight: this stop went through + // the door. Clear the marker so the next boot reads a clean home. + crate::serve_marker::clear(&cfg.data_dir); // #276: read.shutdown() joined the accept task, so :4243 is released here. // Clear the recorded addr — shutdown_node() is now a no-op until the next // serve arms it, and its port-free probe will already be succeeding. diff --git a/src/lib.rs b/src/lib.rs index 847af40b..a52b59ee 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -355,6 +355,7 @@ pub mod node_control; /// carries the superset. pub mod node_identity; pub mod node_key; +pub mod serve_marker; /// The node's tokio runtime, with a worker floor (CIRISServer#446 / #501). pub mod node_runtime; diff --git a/src/node_control.rs b/src/node_control.rs index 1f1e1526..1163270e 100644 --- a/src/node_control.rs +++ b/src/node_control.rs @@ -72,6 +72,19 @@ pub async fn shutdown_requested() { /// Signal the running node to stop (does not wait). `shutdown_node()` layers the /// port-free wait on top of this. pub fn request_shutdown() { + request_shutdown_from("request_shutdown()"); +} + +/// [`request_shutdown`], saying WHO asked. The origin is the first thing a +/// reader of the node log needs when a serve ends: the embedding host's +/// `shutdown_node()`, a signal, or a caller inside this process. Without it a +/// stop reads the same as a crash one line later (CIRISServer#568). +pub fn request_shutdown_from(origin: &'static str) { + tracing::info!( + origin, + "node stop requested — the serve's stop-select will drain the read API (every \ + in-flight response is written), release the port, and clear the serve marker" + ); // `send_replace`, not `send`: `watch::Sender::send` drops the value when no // receiver is alive, and a `shutdown_node()` that lands after `arm()` but // before the serve reaches its stop-select had no receiver yet — the @@ -337,7 +350,7 @@ pub fn shutdown_node_blocking(timeout: Duration) -> bool { Some(a) => a, None => return true, // not serving — nothing to free }; - request_shutdown(); + request_shutdown_from("shutdown_node() from the embedding host"); let start = Instant::now(); loop { // Directly test the postcondition: can we bind the port? An ACTIVE diff --git a/src/serve_marker.rs b/src/serve_marker.rs new file mode 100644 index 00000000..3701cbe5 --- /dev/null +++ b/src/serve_marker.rs @@ -0,0 +1,287 @@ +//! # The serve marker — a stop that did not go through the door leaves a trace +//! +//! `/serving.json` is written the moment the read API is bound and +//! removed after a CLEAN stop — one that went through `shutdown_node()`, +//! SIGTERM or SIGINT, drained the read API and closed the listener. A boot +//! that finds the file still there therefore knows, from this home alone, +//! that the previous serve ended some other way: the process was replaced +//! (`exec`), killed, or crashed. It says so, with the previous pid, instance +//! and start time, and whether that pid is still alive. +//! +//! # Why this exists (CIRISServer#568) +//! +//! On macOS an agent-hosted node announced itself, and one second later a +//! fresh boot appeared in the same log; the caller of the announce saw a bare +//! `ReadError`. Reading the node log alone there was no way to tell a node +//! that restarts itself from a host that replaced it — the "re-compose" was +//! the agent's setup-complete hand-off, and the dropped response was the +//! client side of that hand-off. The node's own stop door never drops a +//! response (the read API drains under axum's graceful shutdown), but a stop +//! that skips the door leaves no line saying so. Now it leaves this one. +//! +//! The file is not a lock. A live listener already refuses a second bind +//! (`AddrInUse`, enriched with the holder's identity); the marker only makes +//! the PREVIOUS serve's ending legible to the next one. + +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; + +/// File name under the data dir. +pub const FILE_NAME: &str = "serving.json"; + +/// What a serving node writes about itself. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Marker { + /// The serving process. + pub pid: u32, + /// `node_identity::instance_id()` — the same id `/health.node` answers. + pub instance_id: String, + /// `node_identity::started_at_rfc3339()` — the same instant `/health.node` answers. + pub started_at: String, + /// The read API listener. + pub listen_addr: String, + /// The configured key the node serves as. + pub key_id: String, +} + +/// What the previous serve on this home left behind. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Previous { + /// No marker: the previous serve stopped through the door (or there was none). + Clean, + /// A marker: the previous serve did NOT stop through the door. + Unclean { + marker: Marker, + /// Whether that pid still exists (`None` where the platform cannot say). + pid_alive: Option, + }, + /// A marker that could not be read — still evidence of an unclean stop. + Unreadable { path: PathBuf, error: String }, +} + +/// Where the marker lives for a data dir. +#[must_use] +pub fn path(data_dir: &Path) -> PathBuf { + data_dir.join(FILE_NAME) +} + +/// Read what the previous serve left, without touching it. +#[must_use] +pub fn inspect(data_dir: &Path) -> Previous { + let p = path(data_dir); + let bytes = match std::fs::read(&p) { + Ok(b) => b, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Previous::Clean, + Err(e) => { + return Previous::Unreadable { + path: p, + error: e.to_string(), + } + } + }; + match serde_json::from_slice::(&bytes) { + Ok(marker) => { + let pid_alive = pid_is_alive(marker.pid); + Previous::Unclean { marker, pid_alive } + } + Err(e) => Previous::Unreadable { + path: p, + error: e.to_string(), + }, + } +} + +/// Does `pid` name a live process? `None` where the platform cannot say. +#[must_use] +pub fn pid_is_alive(pid: u32) -> Option { + #[cfg(unix)] + { + // SAFETY: `kill(pid, 0)` sends no signal; it only checks existence + // and permission. EPERM means the process exists but is not ours — + // still alive for this question. + let rc = unsafe { libc::kill(pid as libc::pid_t, 0) }; + if rc == 0 { + return Some(true); + } + Some(std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM)) + } + #[cfg(not(unix))] + { + let _ = pid; + None + } +} + +/// First thing in a serve: say what the previous serve left, record it for the +/// in-process status channel, and clear it so this serve's own marker can be +/// written. Returns what was found. +pub fn inspect_at_boot(data_dir: &Path) -> Previous { + let prev = inspect(data_dir); + match &prev { + Previous::Clean => { + tracing::debug!( + path = %path(data_dir).display(), + "no serve marker — the previous serve on this home stopped through the door" + ); + } + Previous::Unclean { marker, pid_alive } => { + crate::compose_status::mark("previous_serve_unclean"); + match pid_alive { + Some(true) => tracing::error!( + previous_pid = marker.pid, + previous_instance_id = %marker.instance_id, + previous_started_at = %marker.started_at, + previous_listen_addr = %marker.listen_addr, + previous_key_id = %marker.key_id, + "the previous serve on this home is STILL RUNNING (its serve marker is \ + present and its pid is alive) — this boot will fail to bind the read API \ + with AddrInUse. Stop it through shutdown_node() / SIGTERM first \ + (CIRISServer#568)" + ), + _ => tracing::warn!( + previous_pid = marker.pid, + previous_instance_id = %marker.instance_id, + previous_started_at = %marker.started_at, + previous_listen_addr = %marker.listen_addr, + previous_key_id = %marker.key_id, + pid_alive = ?pid_alive, + "the previous serve on this home did NOT stop through the shutdown door: \ + the process was replaced (exec), killed, or crashed before \ + shutdown_node() / SIGTERM drained its read API. Any HTTP response in \ + flight at that instant was lost to its caller, and a client that kept a \ + pooled connection to it will see a reset on its next request. This is the \ + HOST's hand-off, not a node restart (CIRISServer#568)" + ), + } + } + Previous::Unreadable { path, error } => { + crate::compose_status::mark("previous_serve_unclean"); + tracing::warn!( + path = %path.display(), + error = %error, + "a serve marker is present but unreadable — the previous serve did not stop \ + through the door, and what it wrote is damaged (CIRISServer#568)" + ); + } + } + if !matches!(prev, Previous::Clean) { + let _ = std::fs::remove_file(path(data_dir)); + } + prev +} + +/// Write this serve's marker. Called once the read API is BOUND — the instant +/// from which a response can be in flight. +pub fn write(data_dir: &Path, listen_addr: SocketAddr, key_id: &str) -> std::io::Result<()> { + let marker = Marker { + pid: std::process::id(), + instance_id: crate::node_identity::instance_id().to_owned(), + started_at: crate::node_identity::started_at_rfc3339(), + listen_addr: listen_addr.to_string(), + key_id: key_id.to_owned(), + }; + let p = path(data_dir); + let tmp = p.with_extension("json.tmp"); + let bytes = serde_json::to_vec_pretty(&marker).map_err(std::io::Error::other)?; + std::fs::write(&tmp, bytes)?; + std::fs::rename(&tmp, &p)?; + tracing::info!( + path = %p.display(), + pid = marker.pid, + instance_id = %marker.instance_id, + "serve marker written — cleared only by a stop through the door; a boot that \ + finds it here will say this serve was replaced or killed (CIRISServer#568)" + ); + Ok(()) +} + +/// Remove this serve's marker: the read API has drained and the listener is +/// closed, so no response can be in flight any more. +pub fn clear(data_dir: &Path) { + let p = path(data_dir); + match std::fs::remove_file(&p) { + Ok(()) => tracing::info!( + path = %p.display(), + "serve marker cleared — the read API drained and the listener closed through \ + the shutdown door; nothing was in flight" + ), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => { + tracing::warn!(path = %p.display(), error = %e, "could not clear the serve marker") + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn scratch() -> PathBuf { + let d = std::env::temp_dir().join(format!( + "ciris-serve-marker-{}-{}", + std::process::id(), + chrono::Utc::now().timestamp_nanos_opt().unwrap_or_default() + )); + std::fs::create_dir_all(&d).expect("scratch dir"); + d + } + + #[test] + fn a_clean_home_has_no_marker_and_a_bound_serve_writes_one() { + let d = scratch(); + assert_eq!(inspect(&d), Previous::Clean); + write(&d, "127.0.0.1:4243".parse().unwrap(), "ciris-server").expect("write"); + match inspect(&d) { + Previous::Unclean { marker, pid_alive } => { + assert_eq!(marker.pid, std::process::id()); + assert_eq!(marker.listen_addr, "127.0.0.1:4243"); + assert_eq!(marker.key_id, "ciris-server"); + // Our own pid is alive; a platform that cannot say says None. + assert!( + pid_alive != Some(false), + "our own pid read as dead: {pid_alive:?}" + ); + } + other => panic!("expected the marker we just wrote, got {other:?}"), + } + clear(&d); + assert_eq!(inspect(&d), Previous::Clean, "a clean stop leaves nothing"); + let _ = std::fs::remove_dir_all(&d); + } + + #[test] + fn a_marker_from_a_dead_process_reads_as_an_unclean_stop_and_is_cleared_at_boot() { + let d = scratch(); + let stale = Marker { + pid: 4_000_000, // beyond any sane pid space; dead everywhere we run + instance_id: "prev-instance".into(), + started_at: "2026-09-08T13:12:27Z".into(), + listen_addr: "127.0.0.1:4243".into(), + key_id: "ciris-server".into(), + }; + std::fs::write(path(&d), serde_json::to_vec(&stale).unwrap()).unwrap(); + match inspect_at_boot(&d) { + Previous::Unclean { marker, pid_alive } => { + assert_eq!(marker.instance_id, "prev-instance"); + #[cfg(unix)] + assert_eq!(pid_alive, Some(false)); + #[cfg(not(unix))] + assert_eq!(pid_alive, None); + } + other => panic!("expected an unclean previous serve, got {other:?}"), + } + assert_eq!(inspect(&d), Previous::Clean, "boot clears the stale marker"); + let _ = std::fs::remove_dir_all(&d); + } + + #[test] + fn a_damaged_marker_is_still_evidence_and_is_cleared() { + let d = scratch(); + std::fs::write(path(&d), b"{not json").unwrap(); + assert!(matches!(inspect_at_boot(&d), Previous::Unreadable { .. })); + assert_eq!(inspect(&d), Previous::Clean); + let _ = std::fs::remove_dir_all(&d); + } +} diff --git a/tests/shutdown_signals.rs b/tests/shutdown_signals.rs index d1504be1..98d7243d 100644 --- a/tests/shutdown_signals.rs +++ b/tests/shutdown_signals.rs @@ -139,3 +139,52 @@ fn the_broker_owns_the_signal_and_never_blocks_on_a_log() { "propagate_terminate must not touch tracing" ); } + +/// The serve marker brackets the listener: inspected before anything binds, +/// written the instant the read API is bound, cleared only after the read API +/// has DRAINED (CIRISServer#568). Order is the whole contract — a marker +/// written before the bind, or cleared before the drain, would lie. +#[test] +fn the_serve_marker_brackets_the_listener() { + let src = compose_src(); + let inspect = src + .find("crate::serve_marker::inspect_at_boot(") + .expect("compose inspects the previous serve's marker"); + let bound = src + .find("crate::compose_status::mark(\"listener_bound\")") + .expect("the listener_bound mark"); + let write = src + .find("crate::serve_marker::write(") + .expect("compose writes the marker"); + let drain = src + .find("read.shutdown().await") + .expect("the read API drain"); + let clear = src + .find("crate::serve_marker::clear(") + .expect("compose clears the marker"); + assert!( + inspect < bound && bound < write && write < drain && drain < clear, + "order must be inspect < listener_bound < write < drain < clear; got \ + inspect={inspect} bound={bound} write={write} drain={drain} clear={clear}" + ); +} + +/// Every stop request names its origin, and the embedding host's door names +/// itself — a stop must not read like a crash one line later (CIRISServer#568). +#[test] +fn every_stop_request_says_who_asked() { + let code = node_control_code(); + assert!( + code.contains("request_shutdown_from(\"shutdown_node() from the embedding host\")"), + "shutdown_node() must state its origin" + ); + let body = code + .split_once("pub fn request_shutdown_from(") + .expect("request_shutdown_from exists") + .1; + let body = &body[..body.find("\n}\n").unwrap_or(body.len())]; + assert!( + body.contains("tracing::info!") && body.contains("origin"), + "request_shutdown_from must log the origin before latching:\n{body}" + ); +} From 20f5c72b2da1b448ddd3213dc6523d9b6048b1c8 Mon Sep 17 00:00:00 2001 From: Eric Moore Date: Tue, 8 Sep 2026 11:15:45 -0500 Subject: [PATCH 2/3] The marker rides the body, lands after a phase, spares a live serve, knows a re-used pid, and goes down before the bind (Codex on #569) - The in-flight guard now travels with the response BODY (`CountedBody`, an http_body::Body wrapper): `next.run` returns when the handler has built its response, and a streaming or large body is still on the wire after that. The counter test asserts the request is still inside while the body is unread and falls back when it ends. - The marker is inspected after the first compose phase opens, so its compose_status mark is recorded rather than dropped. - A marker owned by a LIVE previous serve is kept, not cleared; this serve withholds its own marker (the bind is about to fail against the survivor), so a later kill of the survivor still leaves its marker for the next boot. - Process identity: the marker records the Linux /proc starttime; a marker naming our own pid (a container's PID 1 after restart) or a pid whose start identity differs is a REPLACED serve, never a survivor. - The marker is written just BEFORE lens-core binds and exposes the accept loop, and cleared again if the bind fails, so no window exists with a live listener and no marker. The gate encodes the order: first_phase < inspect < write < listener_bound < drain < clear. Co-Authored-By: Claude Fable 5.1 --- Cargo.lock | 1 + crates/ciris-lens-core/Cargo.toml | 3 + crates/ciris-lens-core/src/role/node.rs | 77 +++++++++-- src/compose.rs | 45 +++++-- src/serve_marker.rs | 163 ++++++++++++++++++++++-- tests/shutdown_signals.rs | 35 +++-- 6 files changed, 281 insertions(+), 43 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3480b857..84ab8420 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1069,6 +1069,7 @@ dependencies = [ "criterion", "ed25519-dalek", "hex", + "http-body", "proptest", "pyo3", "serde", diff --git a/crates/ciris-lens-core/Cargo.toml b/crates/ciris-lens-core/Cargo.toml index 0514eee7..1f68a298 100644 --- a/crates/ciris-lens-core/Cargo.toml +++ b/crates/ciris-lens-core/Cargo.toml @@ -93,6 +93,9 @@ hex = "0.4" # path (which could change with an edge bump). Pin matches the version # edge's transport-http pulls in (0.8.x). axum = "0.8" +# The in-flight body wrapper (`track_in_flight`) implements `http_body::Body`; +# 1.x is axum 0.8's own body trait (already in the lock transitively). +http-body = "1" # tower::ServiceExt is used in node mode tests for `oneshot` requests. tower = { version = "0.5", features = ["util"] } diff --git a/crates/ciris-lens-core/src/role/node.rs b/crates/ciris-lens-core/src/role/node.rs index 6ac8bb3d..8cf44e16 100644 --- a/crates/ciris-lens-core/src/role/node.rs +++ b/crates/ciris-lens-core/src/role/node.rs @@ -1001,21 +1001,65 @@ pub fn in_flight() -> usize { IN_FLIGHT.load(std::sync::atomic::Ordering::SeqCst) } -/// The counting layer. A guard, not a pair of calls, so a handler that panics -/// or is cancelled still decrements. +/// One request inside the read API, from accept to the LAST BYTE of its +/// response. A guard, not a pair of calls, so a cancelled handler or a body +/// the client stopped reading still decrements. +struct Inside; +impl Inside { + fn enter() -> Self { + IN_FLIGHT.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Self + } +} +impl Drop for Inside { + fn drop(&mut self) { + IN_FLIGHT.fetch_sub(1, std::sync::atomic::Ordering::SeqCst); + } +} + +/// A response body that carries the request's [`Inside`] guard until the body +/// ends or is dropped. `next.run(..)` returns as soon as the handler has +/// BUILT its response; a streaming body (the SSE route a host merges in) or a +/// large one is still being written after that, and a drain that read the +/// counter then would say `in_flight=0` while bytes were on the wire (Codex +/// on CIRISServer#569). The guard rides the body instead. +struct CountedBody { + inner: B, + _inside: Inside, +} +impl http_body::Body for CountedBody +where + B: http_body::Body + Unpin, +{ + type Data = B::Data; + type Error = B::Error; + fn poll_frame( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll, Self::Error>>> { + std::pin::Pin::new(&mut self.get_mut().inner).poll_frame(cx) + } + fn is_end_stream(&self) -> bool { + self.inner.is_end_stream() + } + fn size_hint(&self) -> http_body::SizeHint { + self.inner.size_hint() + } +} + +/// The counting layer: enter on accept, leave when the response BODY ends. pub async fn track_in_flight( request: axum::extract::Request, next: axum::middleware::Next, ) -> Response { - struct Inside; - impl Drop for Inside { - fn drop(&mut self) { - IN_FLIGHT.fetch_sub(1, std::sync::atomic::Ordering::SeqCst); - } - } - IN_FLIGHT.fetch_add(1, std::sync::atomic::Ordering::SeqCst); - let _inside = Inside; - next.run(request).await + let inside = Inside::enter(); + let response = next.run(request).await; + response.map(|body| { + axum::body::Body::new(CountedBody { + inner: body, + _inside: inside, + }) + }) } // ─── LensCore::node ──────────────────────────────────────────────── @@ -1899,10 +1943,19 @@ mod tests { GATE.add_permits(1); let resp = call.await.unwrap().unwrap(); assert_eq!(resp.status(), axum::http::StatusCode::OK); + // The handler has returned, but the BODY has not been read: the + // request is still inside (the SSE / large-body case, Codex on #569). + assert_eq!( + in_flight(), + before + 1, + "a response whose body is unread is still inside" + ); + let bytes = axum::body::to_bytes(resp.into_body(), 64).await.unwrap(); + assert_eq!(&bytes[..], b"done"); assert_eq!( in_flight(), before, - "the response was written and the count fell back" + "the body ended and the count fell back" ); } } diff --git a/src/compose.rs b/src/compose.rs index df65ae33..27cc5edd 100644 --- a/src/compose.rs +++ b/src/compose.rs @@ -98,12 +98,6 @@ pub async fn serve_with_adapter(cfg: ServerConfig, adapter: Arc) -> // (CIRISServer#557). crate::graph_config::invalidate(); - // What did the PREVIOUS serve on this home leave? A marker still present - // means it did not stop through the door — replaced, killed or crashed — - // and this is where that becomes a line in THIS log rather than a guess - // from a caller's dropped connection (CIRISServer#568). - let _previous_serve = crate::serve_marker::inspect_at_boot(&cfg.data_dir); - // ── RNG startup health-check (CIRISServer#283 finding 2) ────────────────── // Arm the SP 800-90B latch ONCE at boot so `ciris_crypto::random::fill`'s // fail-secure gate is live: if the OS entropy source is producing detectably @@ -117,6 +111,12 @@ pub async fn serve_with_adapter(cfg: ServerConfig, adapter: Arc) -> // exists. "Not a recoverable pause" — only a manual removal of the latch (the // human act a valid accord:lifecycle:active re-activation authorizes) clears it. crate::compose_status::phase("halt_gate"); + // What did the PREVIOUS serve on this home leave? A marker still present + // means it did not stop through the door — replaced, killed or crashed — + // and this is where that becomes a line in THIS log rather than a guess + // from a caller's dropped connection (CIRISServer#568). After the first + // phase has opened, so its compose_status mark has somewhere to land. + let previous_serve = crate::serve_marker::inspect_at_boot(&cfg.data_dir); crate::accord_halt::check_halt_gate(&cfg.home)?; crate::compose_status::phase("capabilities"); @@ -1004,6 +1004,27 @@ pub async fn serve_with_adapter(cfg: ServerConfig, adapter: Arc) -> still serve on the read-API port [#279]" ); } + // The serve marker goes down BEFORE the bind: lens-core binds and exposes + // the accept loop inside the call below, so a marker written after it + // returns would leave a window with a live listener and no marker + // (CIRISServer#568, Codex on #569). Withheld when a previous serve is + // still running — its marker is not ours to overwrite, and the bind is + // about to fail against it anyway. A bind failure clears ours again. + let marker_written = if crate::serve_marker::previous_still_running(&previous_serve) { + tracing::warn!( + "serve marker withheld — the previous serve on this home is still running; \ + its marker stands until it stops (CIRISServer#568)" + ); + false + } else { + match crate::serve_marker::write(&cfg.data_dir, cfg.read_api_addr(), &cfg.key_id) { + Ok(()) => true, + Err(e) => { + tracing::warn!(error = %e, "could not write the serve marker"); + false + } + } + }; let read = { let read = LensCore::read_api_with_extra_at_fidelity( Arc::clone(&engine), @@ -1592,18 +1613,16 @@ pub async fn serve_with_adapter(cfg: ServerConfig, adapter: Arc) -> let read = match read { Ok(read) => read, Err(err) => { + // No listener ever existed: the marker written above must not + // read as an unclean stop at the next boot. + if marker_written { + crate::serve_marker::clear(&cfg.data_dir); + } return Err(enrich_read_api_bind_error(err, cfg.read_api_addr().port()).await); } }; crate::compose_status::mark("listener_bound"); tracing::info!(read_api = %read.listen_addr(), "read API up — GET /lens/api/v1/* + GET /v1/identity"); - // From this instant a response can be in flight: write the serve - // marker, cleared only by a stop that drains the read API - // (CIRISServer#568). A write failure is logged, not fatal — the marker - // is legibility, not a lock. - if let Err(e) = crate::serve_marker::write(&cfg.data_dir, read.listen_addr(), &cfg.key_id) { - tracing::warn!(error = %e, "could not write the serve marker"); - } // #279: the listener is now guaranteed BOUND here (lens-core binds // synchronously before spawning the accept loop and a bind failure is // the `?` above). Stamp the milestone so compose_status distinguishes diff --git a/src/serve_marker.rs b/src/serve_marker.rs index 3701cbe5..236334e4 100644 --- a/src/serve_marker.rs +++ b/src/serve_marker.rs @@ -44,6 +44,12 @@ pub struct Marker { pub listen_addr: String, /// The configured key the node serves as. pub key_id: String, + /// The process's start identity where the platform has one (Linux: + /// `/proc//stat` starttime, in clock ticks since boot). A container + /// restart commonly re-uses the pid — often PID 1 — so pid alone cannot + /// tell "still running" from "replaced"; this can (Codex on #569). + #[serde(default)] + pub proc_start: Option, } /// What the previous serve on this home left behind. @@ -83,7 +89,7 @@ pub fn inspect(data_dir: &Path) -> Previous { }; match serde_json::from_slice::(&bytes) { Ok(marker) => { - let pid_alive = pid_is_alive(marker.pid); + let pid_alive = previous_is_alive(&marker); Previous::Unclean { marker, pid_alive } } Err(e) => Previous::Unreadable { @@ -93,6 +99,48 @@ pub fn inspect(data_dir: &Path) -> Previous { } } +/// Is the serve that wrote `marker` still running — the SAME process, not a +/// later one that inherited its pid? +/// +/// Three questions, in order: is the pid ours (a re-used pid after a restart +/// — PID 1 in a container — is us, not a survivor); does the pid exist; and, +/// where the platform records one, does its start identity match what the +/// marker recorded. `None` where the platform can answer none of it. +#[must_use] +pub fn previous_is_alive(marker: &Marker) -> Option { + if marker.pid == std::process::id() { + return Some(false); + } + match pid_is_alive(marker.pid) { + Some(true) => match (marker.proc_start, proc_start_of(marker.pid)) { + (Some(recorded), Some(now)) if recorded != now => Some(false), + _ => Some(true), + }, + other => other, + } +} + +/// A process's start identity: Linux `/proc//stat` field 22 (starttime, +/// clock ticks since boot). `None` elsewhere or when unreadable. +#[must_use] +pub fn proc_start_of(pid: u32) -> Option { + #[cfg(target_os = "linux")] + { + let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?; + // The comm field is parenthesised and may contain spaces: split after + // the LAST ')' so a process named "a b)" cannot shift the columns. + let rest = stat.rsplit_once(')')?.1; + // `rest` begins with the state field (3); starttime is field 22, so + // index 22 - 3 = 19 among the remaining whitespace-separated fields. + rest.split_whitespace().nth(19)?.parse().ok() + } + #[cfg(not(target_os = "linux"))] + { + let _ = pid; + None + } +} + /// Does `pid` name a live process? `None` where the platform cannot say. #[must_use] pub fn pid_is_alive(pid: u32) -> Option { @@ -166,14 +214,35 @@ pub fn inspect_at_boot(data_dir: &Path) -> Previous { ); } } - if !matches!(prev, Previous::Clean) { + // A marker owned by a LIVE serve is that serve's, not ours to clear: this + // boot will fail its bind, and if the survivor is later killed the next + // boot must still find its marker (Codex on #569). Only a dead or damaged + // one is cleared, so this serve can write its own. + let keep = previous_still_running(&prev); + if !keep && !matches!(prev, Previous::Clean) { let _ = std::fs::remove_file(path(data_dir)); } prev } -/// Write this serve's marker. Called once the read API is BOUND — the instant -/// from which a response can be in flight. +/// Whether [`inspect_at_boot`] found a previous serve still running — in which +/// case this serve must NOT write a marker over it. +#[must_use] +pub fn previous_still_running(prev: &Previous) -> bool { + matches!( + prev, + Previous::Unclean { + pid_alive: Some(true), + .. + } + ) +} + +/// Write this serve's marker. Called just BEFORE the read API binds: lens-core +/// binds and exposes the accept loop inside one call, so a marker written +/// after it returns leaves a window in which a request can be accepted and the +/// process killed with no marker on disk (Codex on #569). A bind failure +/// clears it again (see `compose`). pub fn write(data_dir: &Path, listen_addr: SocketAddr, key_id: &str) -> std::io::Result<()> { let marker = Marker { pid: std::process::id(), @@ -181,6 +250,7 @@ pub fn write(data_dir: &Path, listen_addr: SocketAddr, key_id: &str) -> std::io: started_at: crate::node_identity::started_at_rfc3339(), listen_addr: listen_addr.to_string(), key_id: key_id.to_owned(), + proc_start: proc_start_of(std::process::id()), }; let p = path(data_dir); let tmp = p.with_extension("json.tmp"); @@ -238,10 +308,13 @@ mod tests { assert_eq!(marker.pid, std::process::id()); assert_eq!(marker.listen_addr, "127.0.0.1:4243"); assert_eq!(marker.key_id, "ciris-server"); - // Our own pid is alive; a platform that cannot say says None. - assert!( - pid_alive != Some(false), - "our own pid read as dead: {pid_alive:?}" + // A marker naming OUR OWN pid is by definition not a survivor + // (a re-used pid after a restart — Codex on #569), so the + // question "is the previous serve still running" is `false`. + assert_eq!( + pid_alive, + Some(false), + "our own marker is a replaced serve, not a survivor" ); } other => panic!("expected the marker we just wrote, got {other:?}"), @@ -260,6 +333,7 @@ mod tests { started_at: "2026-09-08T13:12:27Z".into(), listen_addr: "127.0.0.1:4243".into(), key_id: "ciris-server".into(), + proc_start: None, }; std::fs::write(path(&d), serde_json::to_vec(&stale).unwrap()).unwrap(); match inspect_at_boot(&d) { @@ -276,6 +350,79 @@ mod tests { let _ = std::fs::remove_dir_all(&d); } + /// A container restart hands the new serve the old serve's pid (PID 1). + /// A marker naming OUR OWN pid is a replaced serve, never a survivor. + #[test] + fn a_marker_with_our_own_pid_is_a_replaced_serve_not_a_survivor() { + let d = scratch(); + let reused = Marker { + pid: std::process::id(), + instance_id: "the-previous-life-of-pid-1".into(), + started_at: "2026-09-08T15:15:20Z".into(), + listen_addr: "0.0.0.0:4243".into(), + key_id: "ciris-server".into(), + proc_start: Some(1), + }; + std::fs::write(path(&d), serde_json::to_vec(&reused).unwrap()).unwrap(); + match inspect_at_boot(&d) { + Previous::Unclean { pid_alive, .. } => assert_eq!(pid_alive, Some(false)), + other => panic!("{other:?}"), + } + assert_eq!( + inspect(&d), + Previous::Clean, + "and it was cleared for this serve's own" + ); + let _ = std::fs::remove_dir_all(&d); + } + + /// A marker owned by a LIVE, different process is left alone: the bind + /// will fail, and the survivor's marker must outlive this failed boot. + #[cfg(unix)] + #[test] + fn a_live_previous_serve_keeps_its_marker() { + let d = scratch(); + // pid 1 (init) is alive on every unix and is never us in a test. + let live = Marker { + pid: 1, + instance_id: "the-survivor".into(), + started_at: "2026-09-08T15:15:20Z".into(), + listen_addr: "0.0.0.0:4243".into(), + key_id: "ciris-server".into(), + proc_start: proc_start_of(1), + }; + std::fs::write(path(&d), serde_json::to_vec(&live).unwrap()).unwrap(); + let prev = inspect_at_boot(&d); + assert!(previous_still_running(&prev), "{prev:?}"); + assert!( + path(&d).exists(), + "the survivor's marker is not ours to clear" + ); + let _ = std::fs::remove_dir_all(&d); + } + + #[cfg(target_os = "linux")] + #[test] + fn a_reused_pid_with_a_different_start_identity_is_not_alive() { + let d = scratch(); + // pid 1 is alive; a marker claiming pid 1 with the WRONG start identity + // describes a process that no longer exists. + let stale = Marker { + pid: 1, + instance_id: "prev".into(), + started_at: "2026-09-08T15:15:20Z".into(), + listen_addr: "0.0.0.0:4243".into(), + key_id: "ciris-server".into(), + proc_start: Some(u64::MAX), + }; + std::fs::write(path(&d), serde_json::to_vec(&stale).unwrap()).unwrap(); + match inspect(&d) { + Previous::Unclean { pid_alive, .. } => assert_eq!(pid_alive, Some(false)), + other => panic!("{other:?}"), + } + let _ = std::fs::remove_dir_all(&d); + } + #[test] fn a_damaged_marker_is_still_evidence_and_is_cleared() { let d = scratch(); diff --git a/tests/shutdown_signals.rs b/tests/shutdown_signals.rs index 98d7243d..de538799 100644 --- a/tests/shutdown_signals.rs +++ b/tests/shutdown_signals.rs @@ -140,10 +140,12 @@ fn the_broker_owns_the_signal_and_never_blocks_on_a_log() { ); } -/// The serve marker brackets the listener: inspected before anything binds, -/// written the instant the read API is bound, cleared only after the read API -/// has DRAINED (CIRISServer#568). Order is the whole contract — a marker -/// written before the bind, or cleared before the drain, would lie. +/// The serve marker brackets the listener: inspected once the first phase is +/// open (so its status mark lands), written just BEFORE lens-core binds and +/// exposes the accept loop, cleared only after the read API has DRAINED +/// (CIRISServer#568). Order is the whole contract — a marker written after the +/// bind leaves a window with a listener and no marker; one cleared before the +/// drain would lie. #[test] fn the_serve_marker_brackets_the_listener() { let src = compose_src(); @@ -159,13 +161,26 @@ fn the_serve_marker_brackets_the_listener() { let drain = src .find("read.shutdown().await") .expect("the read API drain"); - let clear = src - .find("crate::serve_marker::clear(") - .expect("compose clears the marker"); + // The bind-FAILURE arm also clears (no listener ever existed); the clear + // that closes a served life is the one AFTER the drain. + let clear = drain + + src[drain..] + .find("crate::serve_marker::clear(") + .expect("compose clears the marker after the drain"); + let first_phase = src + .find("compose_status::phase(\"halt_gate\")") + .expect("the first boot phase is stamped"); + assert!( + first_phase < inspect && inspect < write && write < bound && bound < drain && drain < clear, + "order must be first_phase < inspect < write < listener_bound < drain < clear — the \ + marker is inspected once a phase can record it, written BEFORE lens-core binds and \ + exposes the accept loop, and cleared only after the drain; got \ + first_phase={first_phase} inspect={inspect} write={write} bound={bound} drain={drain} \ + clear={clear}" + ); assert!( - inspect < bound && bound < write && write < drain && drain < clear, - "order must be inspect < listener_bound < write < drain < clear; got \ - inspect={inspect} bound={bound} write={write} drain={drain} clear={clear}" + src.contains("serve_marker::previous_still_running(&previous_serve)"), + "a live previous serve's marker is withheld from, not overwritten by, this serve" ); } From 2b815cb442a1fb795dc90ab0a5b7d74c50cd2e18 Mon Sep 17 00:00:00 2001 From: Eric Moore Date: Tue, 8 Sep 2026 11:32:44 -0500 Subject: [PATCH 3/3] The marker clear checks ownership, the in-flight count is per listener, and every teardown line names its step in the message (Codex round 2 on #569, plus the probe) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - `serve_marker::clear` reads the file first and removes it only when its pid and instance_id are this serve's — two boots racing for one home can both write before one loses the bind, and the loser's clean-up must not take the winner's marker (unit test: another serve's marker survives our clear). - lens-core: `InFlight` is one counter per read-API listener, carried by the `ReadApiHandle` and installed on that listener's router with `from_fn_with_state`; the process-wide static is gone, so a host with two listeners sees each drain its own requests. - `stop_step` puts the step name in the MESSAGE: the log dedup collapses lines whose message normalises the same, and on the probe it hid three of the six "teardown step done" lines behind the first. Probe on the rebuilt binary (local, persistent home): SIGTERM → drained in <1 ms, `edge run loop` named at ERROR after its 10 s budget, `node stopped took_ms=10001`, exit 143, marker cleared. kill -9 then restart → WARN "the previous serve on this home did NOT stop through the shutdown door" with the dead pid, then a fresh marker. Edge's side is CIRISEdge#578. Co-Authored-By: Claude Fable 5.1 --- crates/ciris-lens-core/src/role/node.rs | 67 ++++++++++++++------- src/compose.rs | 79 ++++++++++++++++++++++--- src/main.rs | 8 ++- src/serve_marker.rs | 51 +++++++++++++++- tests/shutdown_signals.rs | 46 ++++++++++++++ 5 files changed, 219 insertions(+), 32 deletions(-) diff --git a/crates/ciris-lens-core/src/role/node.rs b/crates/ciris-lens-core/src/role/node.rs index 8cf44e16..421e8a74 100644 --- a/crates/ciris-lens-core/src/role/node.rs +++ b/crates/ciris-lens-core/src/role/node.rs @@ -951,6 +951,10 @@ pub struct ReadApiHandle { http_shutdown_tx: watch::Sender, http_join: JoinHandle<()>, listen_addr: SocketAddr, + /// THIS listener's in-flight count — one per handle, not per process, so + /// a host running two read APIs sees each drain its own requests (Codex + /// on CIRISServer#569). + in_flight: InFlight, } impl ReadApiHandle { @@ -969,7 +973,7 @@ impl ReadApiHandle { /// a caller's dropped connection can be placed on the right side of this /// door from the log alone (CIRISServer#568). pub async fn shutdown(self) -> Result<(), NodeError> { - let inside = in_flight(); + let inside = self.in_flight.get(); let t0 = std::time::Instant::now(); tracing::info!( listen_addr = %self.listen_addr, @@ -982,7 +986,7 @@ impl ReadApiHandle { tracing::info!( listen_addr = %self.listen_addr, drained = inside, - still_in_flight = in_flight(), + still_in_flight = self.in_flight.get(), took_ms = t0.elapsed().as_millis() as u64, "lens read API stopped — listener closed; no response was cut" ); @@ -990,30 +994,36 @@ impl ReadApiHandle { } } -/// HTTP requests currently inside the read API: accepted, response not yet -/// fully written. Counted by [`track_in_flight`] on every route, including the -/// routes a host merges in. Read at shutdown so the drain is a number. -pub static IN_FLIGHT: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); - -/// Requests inside the read API right now. -#[must_use] -pub fn in_flight() -> usize { - IN_FLIGHT.load(std::sync::atomic::Ordering::SeqCst) +/// HTTP requests currently inside ONE read-API listener: accepted, response +/// not yet fully written. Counted by [`track_in_flight`] on every route of that +/// listener, including the routes a host merges in; read by +/// [`ReadApiHandle::shutdown`] so the drain is a number. One per listener — +/// a process-wide static would let a long response on listener B be reported +/// as listener A's straggler (Codex on CIRISServer#569). +#[derive(Clone, Default)] +pub struct InFlight(Arc); + +impl InFlight { + /// Requests inside this listener right now. + #[must_use] + pub fn get(&self) -> usize { + self.0.load(std::sync::atomic::Ordering::SeqCst) + } } -/// One request inside the read API, from accept to the LAST BYTE of its +/// One request inside a listener, from accept to the LAST BYTE of its /// response. A guard, not a pair of calls, so a cancelled handler or a body /// the client stopped reading still decrements. -struct Inside; +struct Inside(InFlight); impl Inside { - fn enter() -> Self { - IN_FLIGHT.fetch_add(1, std::sync::atomic::Ordering::SeqCst); - Self + fn enter(counter: &InFlight) -> Self { + counter.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Self(counter.clone()) } } impl Drop for Inside { fn drop(&mut self) { - IN_FLIGHT.fetch_sub(1, std::sync::atomic::Ordering::SeqCst); + self.0 .0.fetch_sub(1, std::sync::atomic::Ordering::SeqCst); } } @@ -1048,11 +1058,13 @@ where } /// The counting layer: enter on accept, leave when the response BODY ends. +/// Installed with `from_fn_with_state(counter, track_in_flight)` per listener. pub async fn track_in_flight( + State(counter): State, request: axum::extract::Request, next: axum::middleware::Next, ) -> Response { - let inside = Inside::enter(); + let inside = Inside::enter(&counter); let response = next.run(request).await; response.map(|body| { axum::body::Body::new(CountedBody { @@ -1140,9 +1152,14 @@ impl LensCore { // The in-flight counter wraps the WHOLE router — the host's merged // routes included — so a drain at shutdown counts every response it // is about to finish writing (CIRISServer#568). - let router = extra - .merge(build_read_router(state)) - .layer(axum::middleware::from_fn(track_in_flight)); + let in_flight = InFlight::default(); + let router = + extra + .merge(build_read_router(state)) + .layer(axum::middleware::from_fn_with_state( + in_flight.clone(), + track_in_flight, + )); let (http_shutdown_tx, mut http_shutdown_rx) = watch::channel(false); // Bind SYNCHRONOUSLY, before spawning the accept loop (CIRISServer#279). // The old shape bound inside the spawned task and swallowed the error @@ -1177,6 +1194,7 @@ impl LensCore { http_shutdown_tx, http_join, listen_addr, + in_flight, }) } @@ -1913,6 +1931,7 @@ mod tests { use axum::{routing::get, Router}; use tower::ServiceExt as _; static GATE: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(0); + let counter = InFlight::default(); let app = Router::new() .route( "/slow", @@ -1922,7 +1941,11 @@ mod tests { "done" }), ) - .layer(axum::middleware::from_fn(track_in_flight)); + .layer(axum::middleware::from_fn_with_state( + counter.clone(), + track_in_flight, + )); + let in_flight = move || counter.get(); let before = in_flight(); let call = tokio::spawn( app.oneshot( diff --git a/src/compose.rs b/src/compose.rs index 27cc5edd..48b346c9 100644 --- a/src/compose.rs +++ b/src/compose.rs @@ -71,6 +71,55 @@ pub async fn serve(cfg: ServerConfig) -> Result<()> { /// `run_lifecycle` runs as a supervised background task, and its `stop` runs on /// shutdown. This is the "ciris-server + an adapter" seam (MISSION §1.2); the /// default [`serve`] passes [`NoopAdapter`], so existing behavior is unchanged. +/// How long one teardown step may take before the stop proceeds without it. +/// The read API drains under its own graceful shutdown before any of these +/// run; what these bound is the node's OWN loops and the edge's run loop. +pub const STOP_STEP_BUDGET: std::time::Duration = std::time::Duration::from_secs(10); + +/// One teardown step, TIMED and BOUNDED (CIRISServer#568). +/// +/// The stop probe that found this: after SIGTERM the read API drained in +/// under a millisecond and every loop logged "shutting down" — and the +/// process lived on for eight minutes, the capacity scorer and the trace-plane +/// watch still ticking, until SIGKILL. `edge_join.await` never returned +/// (edge's `run` joins transport tasks that do not observe its shutdown), and +/// nothing after it ran: no propagate, no exit, no line saying why. A `docker +/// stop` of the same node ended in exit 137 for the same reason. +/// +/// So every step says how long it took, and a step that outlives its budget +/// is logged at ERROR by name and LEFT BEHIND: the stop proceeds, the process +/// exits, and the runtime takes the straggler with it. A stop that can hang +/// on one join is not a stop; a stop that names the straggler is one an +/// operator can read. +async fn stop_step(name: &'static str, fut: impl std::future::Future) -> Option { + let t0 = std::time::Instant::now(); + match tokio::time::timeout(STOP_STEP_BUDGET, fut).await { + // The step name is IN the message, not only a field: the log dedup + // collapses lines whose message normalises the same, and it hid the + // retention / mesh-config / config steps behind the first three "done" + // lines on the very probe that proved this path (CIRISServer#568). + Ok(out) => { + tracing::info!( + step = name, + took_ms = t0.elapsed().as_millis() as u64, + "teardown step done: {name}" + ); + Some(out) + } + Err(_) => { + tracing::error!( + step = name, + budget_ms = STOP_STEP_BUDGET.as_millis() as u64, + "teardown step did NOT finish within its budget: {name} — proceeding without \ + it; the process will exit and the runtime takes the straggler with it. This \ + step is the reason a stop of this node used to hang until SIGKILL \ + (CIRISServer#568)" + ); + None + } + } +} + pub async fn serve_with_adapter(cfg: ServerConfig, adapter: Arc) -> Result<()> { cfg.ensure_dirs()?; @@ -1840,31 +1889,47 @@ pub async fn serve_with_adapter(cfg: ServerConfig, adapter: Arc) -> // Signal the lifecycle to return, run the adapter's `stop()`, and join the // lifecycle task — around the edge teardown so the adapter unwinds with the // rest of the shared core. + let stop_began = std::time::Instant::now(); let _ = adapter_sd_tx.send(true); - let _ = adapter.stop().await; - let _ = adapter_join.await; + stop_step("adapter.stop", adapter.stop()).await; + stop_step("adapter lifecycle", adapter_join).await; // Tear down the CEG-driven reconcile loop (if it was spawned). let _ = reconcile_sd_tx.send(true); if let Some(join) = reconcile_join { - let _ = join.await; + stop_step("replication reconciler", join).await; } // Tear down the retention loop (CIRISServer#348). Before the config // reconciler: the loop selects on the config watch, and dropping the sender // first would race its shutdown branch against a `changed()` error break. let _ = retention_sd_tx.send(true); - let _ = retention_join.await; + stop_step("retention loop", retention_join).await; // Tear down the mesh-config consumer refresh loop (CIRISServer#365). Its // readers (the read API, the ingest router) are already gone by here. let _ = mesh_config_sd_tx.send(true); - let _ = mesh_config_join.await; + stop_step("mesh-config consumer", mesh_config_join).await; // Tear down the CEG-driven config reconcile loop (Server 0.5 Phase 2). let _ = config_sd_tx.send(true); - let _ = config_reconcile_join.await; + stop_step("config reconciler", config_reconcile_join).await; let _ = edge_shutdown_tx.send(true); // `None` in the #221 fold — the agent owns the edge's run loop (init_edge_runtime). if let Some(edge_join) = edge_join { - let _ = edge_join.await; + // The step that hung: edge's `run` joins transport tasks that do not + // all observe its shutdown signal (CIRISEdge, filed from #568). When + // it outlives the budget the handle is aborted so the runtime's own + // shutdown is not held by it either. + let edge_join_abort = edge_join.abort_handle(); + if stop_step("edge run loop", edge_join).await.is_none() { + edge_join_abort.abort(); + } } + tracing::info!( + stopped_by_sigterm, + sigterm_latched = crate::node_control::terminated_now(), + took_ms = stop_began.elapsed().as_millis() as u64, + "node stopped — every teardown step above ran or was left behind by name; the \ + process exits now (a SIGTERM is re-raised with the default action, any other \ + stop returns to the host)" + ); // A SIGTERM asked for the PROCESS to end, not only this serve. Now that // the node has unwound cleanly, finish what the signal asked for unless an // embedding host owns SIGTERM and decides for itself (#556 review): an diff --git a/src/main.rs b/src/main.rs index 343f7ced..3c389f0f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,7 +13,13 @@ fn main() -> Result<()> { // path on exactly the small hosts this exists for (CIRISServer#501). let workers = ciris_server::node_runtime::worker_threads(); let runtime = ciris_server::node_runtime::build("ciris-node")?; - runtime.block_on(async_main(workers)) + let result = runtime.block_on(async_main(workers)); + // A bounded runtime shutdown (CIRISServer#568): a blocking-pool thread + // parked in a transport read would otherwise hold the process open after + // the serve has returned — the implicit `Drop` waits for it without limit. + // Every task the teardown left behind by name is cancelled here. + runtime.shutdown_timeout(std::time::Duration::from_secs(5)); + result } async fn async_main(worker_threads: usize) -> Result<()> { diff --git a/src/serve_marker.rs b/src/serve_marker.rs index 236334e4..8d154916 100644 --- a/src/serve_marker.rs +++ b/src/serve_marker.rs @@ -267,10 +267,33 @@ pub fn write(data_dir: &Path, listen_addr: SocketAddr, key_id: &str) -> std::io: Ok(()) } -/// Remove this serve's marker: the read API has drained and the listener is -/// closed, so no response can be in flight any more. +/// Remove this serve's marker — and ONLY this serve's: the file is read first +/// and left alone unless its `pid` and `instance_id` are ours. Two boots +/// racing for one home can both write before one loses the bind; the loser's +/// clean-up must not take the winner's marker with it (Codex on #569). Called +/// after the read API has drained (no response can be in flight) and on the +/// bind-failure arm (no listener ever existed). pub fn clear(data_dir: &Path) { let p = path(data_dir); + match std::fs::read(&p) + .ok() + .and_then(|b| serde_json::from_slice::(&b).ok()) + { + Some(m) + if m.pid != std::process::id() + || m.instance_id != crate::node_identity::instance_id() => + { + tracing::info!( + path = %p.display(), + owner_pid = m.pid, + owner_instance_id = %m.instance_id, + "serve marker left in place — it belongs to another serve on this home, not \ + to this one" + ); + return; + } + _ => {} + } match std::fs::remove_file(&p) { Ok(()) => tracing::info!( path = %p.display(), @@ -423,6 +446,30 @@ mod tests { let _ = std::fs::remove_dir_all(&d); } + /// `clear` takes only its own marker: another serve's stays. + #[test] + fn clear_leaves_another_serves_marker_alone() { + let d = scratch(); + let theirs = Marker { + pid: std::process::id().wrapping_add(1), + instance_id: "someone-else".into(), + started_at: "2026-09-08T16:00:00Z".into(), + listen_addr: "0.0.0.0:4243".into(), + key_id: "ciris-server".into(), + proc_start: None, + }; + std::fs::write(path(&d), serde_json::to_vec(&theirs).unwrap()).unwrap(); + clear(&d); + assert!( + path(&d).exists(), + "another serve's marker survives our clear" + ); + write(&d, "127.0.0.1:4243".parse().unwrap(), "ciris-server").expect("write ours"); + clear(&d); + assert!(!path(&d).exists(), "our own marker is cleared"); + let _ = std::fs::remove_dir_all(&d); + } + #[test] fn a_damaged_marker_is_still_evidence_and_is_cleared() { let d = scratch(); diff --git a/tests/shutdown_signals.rs b/tests/shutdown_signals.rs index de538799..37bfee61 100644 --- a/tests/shutdown_signals.rs +++ b/tests/shutdown_signals.rs @@ -203,3 +203,49 @@ fn every_stop_request_says_who_asked() { "request_shutdown_from must log the origin before latching:\n{body}" ); } + +/// After the read API drains, every teardown join goes through `stop_step` +/// — timed, bounded, and named — and main bounds the runtime's own shutdown +/// (CIRISServer#568). A bare `.await` on a join handle after the drain is the +/// exact shape that held a stopped node open for eight minutes. +#[test] +fn every_teardown_step_after_the_drain_is_bounded_and_named() { + let src = compose_src(); + let drain = src + .find("read.shutdown().await") + .expect("the read API drain"); + let end = drain + + src[drain..] + .find("crate::node_control::propagate_terminate();") + .expect("the propagate at the end of the serve"); + let teardown = &src[drain..end]; + for bare in ["_join.await", "join.await;", "adapter.stop().await"] { + assert!( + !teardown.contains(bare), + "a bare `{bare}` after the drain is an unbounded stop step — route it through \ + stop_step(name, ..):\n{teardown}" + ); + } + for step in [ + "stop_step(\"adapter.stop\"", + "stop_step(\"retention loop\"", + "stop_step(\"config reconciler\"", + "stop_step(\"edge run loop\"", + ] { + assert!( + teardown.contains(step), + "missing named teardown step {step}" + ); + } + assert!( + teardown.contains("edge_join_abort.abort()"), + "an edge run loop that outlives its budget is aborted, not merely left" + ); + let main_rs = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/src/main.rs")) + .unwrap() + .replace("\r\n", "\n"); + assert!( + main_rs.contains("runtime.shutdown_timeout("), + "main must bound the runtime's shutdown, or a parked blocking thread holds the process" + ); +}