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 86f610d6..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 { @@ -960,13 +964,116 @@ 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 = self.in_flight.get(); + 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 = self.in_flight.get(), + took_ms = t0.elapsed().as_millis() as u64, + "lens read API stopped — listener closed; no response was cut" + ); Ok(()) } } +/// 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 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(InFlight); +impl Inside { + 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) { + self.0 .0.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. +/// 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(&counter); + let response = next.run(request).await; + response.map(|body| { + axum::body::Body::new(CountedBody { + inner: body, + _inside: inside, + }) + }) +} + // ─── LensCore::node ──────────────────────────────────────────────── impl LensCore { @@ -1042,7 +1149,17 @@ 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 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 @@ -1071,12 +1188,13 @@ 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, http_join, listen_addr, + in_flight, }) } @@ -1805,4 +1923,62 @@ 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 counter = InFlight::default(); + 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_with_state( + counter.clone(), + track_in_flight, + )); + let in_flight = move || counter.get(); + 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); + // 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 body ended and the count fell back" + ); + } } diff --git a/src/compose.rs b/src/compose.rs index b2b1b2bf..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()?; @@ -111,6 +160,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"); @@ -998,6 +1053,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), @@ -1586,6 +1662,11 @@ 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); } }; @@ -1793,8 +1874,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. @@ -1803,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/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/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/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..8d154916 --- /dev/null +++ b/src/serve_marker.rs @@ -0,0 +1,481 @@ +//! # 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, + /// 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. +#[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 = previous_is_alive(&marker); + Previous::Unclean { marker, pid_alive } + } + Err(e) => Previous::Unreadable { + path: p, + error: e.to_string(), + }, + } +} + +/// 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 { + #[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)" + ); + } + } + // 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 +} + +/// 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(), + 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(), + proc_start: proc_start_of(std::process::id()), + }; + 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 — 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(), + "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"); + // 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:?}"), + } + 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(), + proc_start: None, + }; + 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); + } + + /// 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); + } + + /// `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(); + 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..37bfee61 100644 --- a/tests/shutdown_signals.rs +++ b/tests/shutdown_signals.rs @@ -139,3 +139,113 @@ 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 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(); + 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"); + // 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!( + src.contains("serve_marker::previous_still_running(&previous_serve)"), + "a live previous serve's marker is withheld from, not overwritten by, this serve" + ); +} + +/// 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}" + ); +} + +/// 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" + ); +}