The stop is legible from the node's own log: serve marker, numbered drain, named origin (#568)

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"] }

crates/ciris-lens-core/src/role/node.rs

@@ -951,6 +951,10 @@ pub struct ReadApiHandle {
     http_shutdown_tx: watch::Sender<bool>,
     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<std::sync::atomic::AtomicUsize>);
+
+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<B> {
+    inner: B,
+    _inside: Inside,
+}
+impl<B> http_body::Body for CountedBody<B>
+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<Option<Result<http_body::Frame<Self::Data>, 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<InFlight>,
+    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"
+        );
+    }
 }