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.