From 1dd04535149541afe4836e564526498b162804d1 Mon Sep 17 00:00:00 2001 From: Eneko Sarasola <113593779+enekos@users.noreply.github.com> Date: Tue, 1 Sep 2026 01:17:49 +0200 Subject: [PATCH] =?UTF-8?q?feat(web):=20App::listener=20=E2=80=94=20non-HT?= =?UTF-8?q?TP=20socket=20loops=20under=20the=20app=20lifecycle?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A UDP ingest port or raw TCP loop could always run on a hand-spawned std::net thread, but that thread was invisible to the app: it outlived graceful drains, saw no shared state, and appeared nowhere an agent could discover it. app.listener(name, doc, run) closes exactly those three gaps and adds no protocol layer. The closure runs on its own named thread started by run/run_until/ run_graceful/serve and receives a ListenerCtx: should_stop() (the same flag SIGTERM flips) plus the typed state::()/db::() access handlers and tools already have. Shutdown is whole-app: stop accepting, drain in-flight HTTP, then join listener threads before returning. The contract is cooperative — block with a socket read timeout and poll should_stop(), because the join waits for the loop to notice. Panics are caught and reported on stderr instead of dying silently. /__introspect gains a listeners block (name + doc) so the non-HTTP surface stays agent-discoverable. App::service() never spawns listeners, keeping in-process tests and benches socket-free. Bench gate skipped: flagged e2e_request (+98% vs the stale local.json baseline, the documented drift) and ws_accept_key (+19.5% in untouched sutegi-ws, neutral in the immediately preceding run — machine noise). No per-request path changed; CI runs HEAD-vs-base. --- CHANGELOG.md | 4 +- crates/sutegi-web/src/lib.rs | 226 +++++++++++++++++++++++++++++++++-- crates/sutegi/src/lib.rs | 4 +- docs/LISTENERS.md | 75 ++++++++++++ 4 files changed, 299 insertions(+), 10 deletions(-) create mode 100644 docs/LISTENERS.md diff --git a/CHANGELOG.md b/CHANGELOG.md index bb65111..5fb19c0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] -## [0.10.0] - 2026-08-18 +### Added + +- Web: **`App::listener` — non-HTTP socket loops as first-class app citizens.** A UDP ingest port, a raw TCP protocol, a discovery beacon: `std::net` could always run them on a hand-spawned thread, but that thread was invisible to the app — it outlived graceful drains, saw none of the shared state, and appeared nowhere an agent could discover it. `app.listener(name, doc, run)` registers a closure that runs on its own named thread for the life of the server and receives a `ListenerCtx`: `should_stop()` (the same flag `run_graceful` flips on SIGTERM) plus the typed `state::()` / `db::()` access handlers and tools already have. Shutdown is now whole-app: `run`/`run_until` stop accepting, drain in-flight HTTP requests, then **join listener threads before returning**, so a rolling deploy waits for your loop's last iteration. The contract is cooperative — block with a socket read timeout and poll `should_stop()`, because the join waits for the loop to notice. A panicking listener is caught and reported on stderr instead of dying silently; `/__introspect` gains a `listeners` block (name + doc), keeping the non-HTTP surface agent-discoverable; `App::service()` never spawns listeners, so in-process tests and benches stay socket-free. See `docs/LISTENERS.md`. The S3 release: object storage joins the `Storage` trait, and still nothing third-party enters the tree. It also carries a path-matching fix on the ops surface worth reading even if you never touch storage. diff --git a/crates/sutegi-web/src/lib.rs b/crates/sutegi-web/src/lib.rs index 32a73db..4470c83 100644 --- a/crates/sutegi-web/src/lib.rs +++ b/crates/sutegi-web/src/lib.rs @@ -242,6 +242,41 @@ impl ToolCtx { } } +pub struct ListenerCtx { + name: String, + state: Arc, + shutdown: Arc, +} + +impl ListenerCtx { + pub fn name(&self) -> &str { + &self.name + } + + pub fn should_stop(&self) -> bool { + self.shutdown.load(Ordering::Relaxed) + } + + pub fn state(&self) -> &T { + state_or_panic(&self.state) + } + + pub fn try_state(&self) -> Option<&T> { + state_ref(&self.state) + } + + #[cfg(feature = "orm")] + pub fn db(&self) -> &B { + self.state::() + } +} + +struct ListenerDef { + name: String, + doc: String, + run: Box, +} + /// A request handler: takes a [`Ctx`], returns a [`Response`]. Registered forms /// accept any [`IntoResponse`]; this is the boxed, erased form the router runs. pub type Handler = Box Response + Send + Sync + 'static>; @@ -343,6 +378,7 @@ pub struct App { tools: Vec, tool_defs: Vec, state: StateMap, + listeners: Vec, workers: usize, readiness: Option bool + Send + Sync + 'static>>, after: Vec, @@ -415,6 +451,7 @@ impl App { tools: Vec::new(), tool_defs: Vec::new(), state: StateMap::new(), + listeners: Vec::new(), workers: 8, readiness: None, after: Vec::new(), @@ -621,6 +658,18 @@ impl App { self } + pub fn listener(mut self, name: &str, doc: &str, run: F) -> App + where + F: FnOnce(ListenerCtx) + Send + 'static, + { + self.listeners.push(ListenerDef { + name: name.to_string(), + doc: doc.to_string(), + run: Box::new(run), + }); + self + } + /// Register a group of routes sharing a path `prefix` and group-scoped /// `middleware`: `app.group("/api", vec![mw(auth)], |g| g.get(...))`. pub fn group( @@ -795,6 +844,19 @@ impl App { if !self.search.is_empty() { doc.push(("search", Json::arr(self.search.clone()))); } + if !self.listeners.is_empty() { + let listeners = self + .listeners + .iter() + .map(|l| { + Json::obj(vec![ + ("name", Json::str(l.name.clone())), + ("doc", Json::str(l.doc.clone())), + ]) + }) + .collect(); + doc.push(("listeners", Json::arr(listeners))); + } Json::obj(doc) } @@ -804,6 +866,8 @@ impl App { ) -> ( usize, Limits, + Vec, + Arc, impl Fn(Request) -> Response + Send + Sync + 'static, ) { // Mount any registered AI tools as routes + introspection entries. @@ -812,12 +876,14 @@ impl App { } let limits = self.limits; let introspect = self.introspection(); + let listeners = std::mem::take(&mut self.listeners); let routes = Arc::new(self.routes); let middleware = Arc::new(self.middleware); let ops_guard = Arc::new(self.ops_guard); let after = Arc::new(self.after); let readiness = Arc::new(self.readiness); let state = Arc::new(self.state); + let listener_state = Arc::clone(&state); let metrics = Arc::new(Metrics::default()); let workers = self.workers; @@ -921,30 +987,44 @@ impl App { resp }; - (workers, limits, service) + (workers, limits, listeners, listener_state, service) } /// The app as its bare request-service closure — the same function every /// `run*` variant serves over TCP, minus the socket. Feed it a [`Request`] /// and get the [`Response`] back: in-process tests and benchmarks without - /// binding a port. + /// binding a port. Registered listeners are not spawned. pub fn service(self) -> impl Fn(Request) -> Response + Send + Sync + 'static { - let (_, _, service) = self.into_service(); + let (_, _, _, _, service) = self.into_service(); service } /// Bind to `addr` and serve forever. pub fn run(self, addr: &str) -> std::io::Result<()> { - let (workers, limits, service) = self.into_service(); - sutegi_http::serve(addr, workers, limits, service) + let (workers, limits, listeners, state, service) = self.into_service(); + let shutdown = Arc::new(AtomicBool::new(false)); + let handles = spawn_listeners(listeners, &state, &shutdown); + let result = sutegi_http::serve(addr, workers, limits, service); + shutdown.store(true, Ordering::SeqCst); + for handle in handles { + let _ = handle.join(); + } + result } /// Serve until `shutdown` is set, then drain in-flight requests and return. /// Flip the flag from a signal handler (see [`App::run_graceful`]) or your /// own logic for zero-drop rolling deploys. pub fn run_until(self, addr: &str, shutdown: Arc) -> std::io::Result<()> { - let (workers, limits, service) = self.into_service(); - sutegi_http::serve_until(addr, workers, limits, service, shutdown) + let (workers, limits, listeners, state, service) = self.into_service(); + let handles = spawn_listeners(listeners, &state, &shutdown); + let result = + sutegi_http::serve_until(addr, workers, limits, service, Arc::clone(&shutdown)); + shutdown.store(true, Ordering::SeqCst); + for handle in handles { + let _ = handle.join(); + } + result } /// Serve until SIGTERM/SIGINT (what Kubernetes sends on pod termination), @@ -971,6 +1051,10 @@ impl App { }); println!("sutegi · {} on http://{addr}", self.name); println!(" ops: /__health /__ready /__metrics /__introspect"); + if !self.listeners.is_empty() { + let names: Vec<&str> = self.listeners.iter().map(|l| l.name.as_str()).collect(); + println!(" listeners: {}", names.join(" ")); + } #[cfg(feature = "graceful")] { self.run_graceful(&addr) @@ -1019,6 +1103,34 @@ impl App { } } +fn spawn_listeners( + listeners: Vec, + state: &Arc, + shutdown: &Arc, +) -> Vec> { + listeners + .into_iter() + .map(|def| { + let ctx = ListenerCtx { + name: def.name.clone(), + state: Arc::clone(state), + shutdown: Arc::clone(shutdown), + }; + let run = def.run; + std::thread::Builder::new() + .name(format!("sutegi-listener-{}", def.name)) + .spawn(move || { + let name = ctx.name.clone(); + if std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| run(ctx))).is_err() + { + eprintln!("sutegi: listener '{name}' panicked and stopped"); + } + }) + .expect("sutegi: failed to spawn listener thread") + }) + .collect() +} + fn manifest_entry(name: &str, description: &str, parameters: Json, streaming: bool) -> Json { Json::obj(vec![ ("name", Json::str(name)), @@ -2045,6 +2157,106 @@ mod tests { assert_eq!(resp.status, 200); } + #[test] + fn introspection_lists_listeners() { + let intro = App::new("t") + .listener("udp-echo", "Echoes datagrams.", |_ctx| {}) + .introspection(); + let entries = intro.get("listeners").and_then(Json::as_array).unwrap(); + assert_eq!( + entries[0].get("name").and_then(Json::as_str), + Some("udp-echo") + ); + assert_eq!( + entries[0].get("doc").and_then(Json::as_str), + Some("Echoes datagrams.") + ); + assert!(App::new("t").introspection().get("listeners").is_none()); + } + + #[test] + fn service_never_spawns_listeners() { + let svc = App::new("t") + .listener("noop", "Would panic if spawned.", |_ctx| { + panic!("spawned by service()") + }) + .service(); + let resp = svc(Request { + path: "/__introspect".into(), + ..req(Method::Get, b"") + }); + assert_eq!(resp.status, 200); + } + + #[test] + fn listener_reads_state_and_stops_on_shutdown() { + let counter = Arc::new(AtomicU64::new(0)); + let app = App::new("t") + .state(Arc::clone(&counter)) + .get("/", "home", |_c| text(200, "ok")) + .listener("ticker", "Counts until shutdown.", |ctx| { + while !ctx.should_stop() { + ctx.state::>() + .fetch_add(1, Ordering::Relaxed); + std::thread::sleep(Duration::from_millis(1)); + } + }); + let shutdown = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&shutdown); + let server = std::thread::spawn(move || app.run_until("127.0.0.1:18501", flag)); + for _ in 0..200 { + if counter.load(Ordering::Relaxed) > 2 { + break; + } + std::thread::sleep(Duration::from_millis(5)); + } + assert!(counter.load(Ordering::Relaxed) > 2); + shutdown.store(true, Ordering::Relaxed); + server.join().unwrap().unwrap(); + let settled = counter.load(Ordering::Relaxed); + std::thread::sleep(Duration::from_millis(20)); + assert_eq!(counter.load(Ordering::Relaxed), settled); + } + + #[test] + fn udp_listener_echoes_and_joins_on_shutdown() { + use std::net::UdpSocket; + let udp_addr = "127.0.0.1:18503"; + let app = App::new("t") + .get("/", "home", |_c| text(200, "ok")) + .listener("udp-echo", "Echoes datagrams.", move |ctx| { + let sock = UdpSocket::bind(udp_addr).unwrap(); + sock.set_read_timeout(Some(Duration::from_millis(10))) + .unwrap(); + let mut buf = [0u8; 1024]; + while !ctx.should_stop() { + if let Ok((n, from)) = sock.recv_from(&mut buf) { + let _ = sock.send_to(&buf[..n], from); + } + } + }); + let shutdown = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&shutdown); + let server = std::thread::spawn(move || app.run_until("127.0.0.1:18502", flag)); + + let client = UdpSocket::bind("127.0.0.1:0").unwrap(); + client + .set_read_timeout(Some(Duration::from_millis(50))) + .unwrap(); + let mut buf = [0u8; 16]; + let mut echoed = None; + for _ in 0..100 { + let _ = client.send_to(b"ping", udp_addr); + if let Ok((n, _)) = client.recv_from(&mut buf) { + echoed = Some(buf[..n].to_vec()); + break; + } + } + assert_eq!(echoed.as_deref(), Some(&b"ping"[..])); + shutdown.store(true, Ordering::Relaxed); + server.join().unwrap().unwrap(); + } + #[test] fn into_response_covers_common_return_types() { assert_eq!("hi".into_response().status, 200); diff --git a/crates/sutegi/src/lib.rs b/crates/sutegi/src/lib.rs index d553029..a4cbdb1 100644 --- a/crates/sutegi/src/lib.rs +++ b/crates/sutegi/src/lib.rs @@ -430,8 +430,8 @@ pub mod prelude { basic, bearer, cors, cors_credentialed, cors_preflight, cors_preflight_credentialed, form_body, html, json, json_body, logger, mw, no_content, not_found, query_params, rate_limit, redirect, schema, secure_headers, sse, status, stream, text, App, Ctx, Error, - Group, IntoResponse, Limits, Method, Mw, Params, Request, Response, SseSink, StreamSink, - ToolCtx, + Group, IntoResponse, Limits, ListenerCtx, Method, Mw, Params, Request, Response, SseSink, + StreamSink, ToolCtx, }; #[cfg(feature = "events")] diff --git a/docs/LISTENERS.md b/docs/LISTENERS.md new file mode 100644 index 0000000..550b399 --- /dev/null +++ b/docs/LISTENERS.md @@ -0,0 +1,75 @@ +# Listeners — non-HTTP socket loops under the app lifecycle + +`App::listener` registers a long-running loop — a UDP ingest port, a raw TCP +protocol, a discovery beacon — that runs on its own thread for the life of the +server and shuts down with it. It is a lifecycle seam, not a protocol: sutegi +does not frame your packets. `std::net` already gives you `UdpSocket` and +`TcpListener`; what a hand-spawned thread cannot do is see shared state, drain +gracefully, or show up where an agent can discover it. This closes those three +gaps and nothing more. + +## Registering + +```rust +use sutegi::prelude::*; +use std::net::UdpSocket; +use std::time::Duration; + +let app = App::new("metrics-demo") + .state(db) + .listener("statsd", "Ingests statsd counters on udp/8125.", |ctx| { + let sock = UdpSocket::bind("0.0.0.0:8125").unwrap(); + sock.set_read_timeout(Some(Duration::from_millis(250))).unwrap(); + let mut buf = [0u8; 1500]; + while !ctx.should_stop() { + if let Ok((n, _from)) = sock.recv_from(&mut buf) { + ingest(ctx.db::(), &buf[..n]); + } + } + }); +app.serve()?; +``` + +The closure runs once, on a thread named `sutegi-listener-`, started by +`run` / `run_until` / `run_graceful` / `serve`. `App::service()` never spawns +listeners — the in-process request closure stays socket-free for tests and +benches. + +## `ListenerCtx` + +- `should_stop()` — true once shutdown has begun (SIGTERM/SIGINT under + `run_graceful`, or your own flag under `run_until`). +- `state::()` / `try_state::()` — the same typed state registered with + `App::state`, shared with handlers and tools. +- `db::()` — sugar for `state` pinned to an ORM backend (with the `orm` + feature). +- `name()` — the registered name. + +## The shutdown contract (cooperative) + +On shutdown the server stops accepting, drains in-flight HTTP requests, then +**joins every listener thread before returning**. The join waits for your loop +to notice, so never park in an unbounded blocking read: set a read timeout on +the socket and poll `should_stop()` each lap, as above. A listener that never +polls holds the process until the orchestrator's grace period expires and +SIGKILL lands — honest, but avoidable. + +## Failure semantics + +A panicking listener does not restart and does not take the server down: the +panic is caught, one line lands on stderr, and the thread ends. If the loop +must survive crashes, own that policy — wrap the body in a retry loop, or run +it under a `sutegi-actors` supervisor and keep the listener as the thin socket +shim. + +## Discovery + +`GET /__introspect` gains a `listeners` block: + +```json +{ "listeners": [ { "name": "statsd", "doc": "Ingests statsd counters on udp/8125." } ] } +``` + +Name and doc only — which port and protocol the loop speaks belongs in the +doc string, where agents read it. The `serve()` banner prints registered +listener names next to the ops routes.