From 9bc70e59fc8358c3c3747e943bb0440c8ea18406 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Thu, 16 Jul 2026 20:46:17 +0200 Subject: [PATCH] =?UTF-8?q?feat(host/events):=20GET=20/api/v1/events=20?= =?UTF-8?q?=E2=80=94=20SSE=20lifecycle=20event=20stream=20(M1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Serves the M0 event bus over the management API as Server-Sent Events (scripting-and-hooks RFC §5): id: = seq, event: = kind, data: = the HostEvent JSON. Standard Last-Event-ID (or ?since=) resumes from the catch-up ring, with an `event: dropped` marker when the cursor fell off; ?kinds= filters server-side (exact kinds or `domain.*` prefixes). Bounds per RFC §9.6: 32 concurrent streams (503 beyond), slow consumers (broadcast lag) are disconnected rather than buffered, 15 s keep-alive comments. Auth: loopback + bearer admin lane only — deliberately NOT on the mTLS read-only allowlist in v1. Note: api/openapi.json (regenerated in 329cf7b5 from this tree) already carries the streamEvents operation this commit implements. Verified live on Linux: catch-up + mid-stream library.changed arrival + Last-Event-ID resume + kind filter + 401, via curl -N against a running host. 335 host tests green (incl. the spec drift test), clippy clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/punktfunk-host/Cargo.toml | 3 + crates/punktfunk-host/src/mgmt.rs | 5 +- crates/punktfunk-host/src/mgmt/events.rs | 236 +++++++++++++++++++++++ crates/punktfunk-host/src/mgmt/tests.rs | 209 ++++++++++++++++++++ 4 files changed, 452 insertions(+), 1 deletion(-) create mode 100644 crates/punktfunk-host/src/mgmt/events.rs diff --git a/crates/punktfunk-host/Cargo.toml b/crates/punktfunk-host/Cargo.toml index 8a562d3c..c3229daa 100644 --- a/crates/punktfunk-host/Cargo.toml +++ b/crates/punktfunk-host/Cargo.toml @@ -56,6 +56,9 @@ tokio-rustls = "0.26" hyper = { version = "1", features = ["server", "http1", "http2"] } hyper-util = { version = "0.1", features = ["server", "server-auto", "tokio", "service"] } tower = { version = "0.5", features = ["util"] } +# Stream combinators for the mgmt API's SSE event feed (`GET /api/v1/events`) — already in the +# tree transitively (axum/hyper) and as a Linux target dep; control plane only. +futures-util = "0.3" rusty_enet = "0.4" serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/crates/punktfunk-host/src/mgmt.rs b/crates/punktfunk-host/src/mgmt.rs index b35a4638..ee9eb6e5 100644 --- a/crates/punktfunk-host/src/mgmt.rs +++ b/crates/punktfunk-host/src/mgmt.rs @@ -32,6 +32,7 @@ use utoipa_scalar::{Scalar, Servable}; mod auth; mod clients; mod display; +mod events; mod gpu; mod host; mod library; @@ -215,7 +216,8 @@ fn api_router_parts() -> (Router>, utoipa::openapi::OpenApi) { stats::stats_recording_get, stats::stats_recording_delete )) - .routes(routes!(stats::logs_get)), + .routes(routes!(stats::logs_get)) + .routes(routes!(events::stream_events)), ) .split_for_parts() } @@ -251,6 +253,7 @@ pub fn openapi_json() -> String { (name = "library", description = "Game library: installed-store titles (Steam) plus user-curated custom entries"), (name = "stats", description = "Streaming performance-stats capture: arm/stop a recording, read the live + saved time-series for graphing"), (name = "logs", description = "Host log stream: the newest in-memory log entries, cursor-paged for live following"), + (name = "events", description = "Host lifecycle events: an SSE stream (client/session/stream lifecycle, pairing, displays, library, host) with Last-Event-ID resume and server-side kind filters"), ) )] struct ApiDoc; diff --git a/crates/punktfunk-host/src/mgmt/events.rs b/crates/punktfunk-host/src/mgmt/events.rs new file mode 100644 index 00000000..59e2fd20 --- /dev/null +++ b/crates/punktfunk-host/src/mgmt/events.rs @@ -0,0 +1,236 @@ +//! `GET /api/v1/events` — the host lifecycle event stream (scripting-and-hooks RFC §5, M1). +//! +//! Server-Sent Events over the existing HTTPS serve loop: each event goes out as one SSE frame +//! with `id:` = the event's `seq`, `event:` = its kind (`stream.started`, …), and `data:` = the +//! [`crate::events::HostEvent`] JSON. Consumers resume with the standard `Last-Event-ID` header +//! (or its `?since=` query twin); a consumer that fell off the catch-up ring gets a synthetic +//! `event: dropped` frame first and should resync via the REST snapshots (`/status`, `/clients`, +//! …). `?kinds=` filters server-side (exact kinds or `domain.*` prefixes, comma-separated). +//! +//! Bounds (RFC §9.6): at most [`MAX_EVENT_STREAMS`] concurrent streams (503 beyond — the +//! consumer retries; SSE clients reconnect by themselves), and a consumer too slow for the +//! live tail (broadcast lag) is **disconnected**, never buffered unboundedly — its reconnect +//! resumes from the ring via `Last-Event-ID`. +//! +//! Auth: deliberately NOT on the mTLS read-only allowlist (`auth::cert_may_access`) — the +//! stream is part of the loopback + bearer admin lane in v1 (RFC §5; revisit when paired +//! clients want an activity feed). + +use super::shared::*; +use axum::http::HeaderMap; +use axum::response::sse::{Event, KeepAlive, Sse}; +use std::collections::VecDeque; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; +use tokio::sync::broadcast; + +/// Concurrent SSE connection cap. Operators run a handful of consumers (console feed, a couple +/// of scripts/plugins); 32 is generous headroom while bounding a runaway reconnect loop. +const MAX_EVENT_STREAMS: usize = 32; + +/// SSE keep-alive comment interval — detects a dead peer and keeps middleboxes from idling +/// the connection out between (low-rate) lifecycle events. +const KEEP_ALIVE: Duration = Duration::from_secs(15); + +static LIVE_STREAMS: AtomicUsize = AtomicUsize::new(0); + +/// RAII slot in the connection cap: taken before the stream starts, released when the SSE body +/// is dropped (client disconnect, slow-consumer cut, server shutdown). `pub(crate)` only for +/// the [`test_support`] saturation helper's return type. +pub(crate) struct StreamSlot; + +fn try_acquire_slot() -> Option { + LIVE_STREAMS + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| { + (n < MAX_EVENT_STREAMS).then_some(n + 1) + }) + .ok() + .map(|_| StreamSlot) +} + +impl Drop for StreamSlot { + fn drop(&mut self) { + LIVE_STREAMS.fetch_sub(1, Ordering::SeqCst); + } +} + +/// `?kinds=stream.*,pairing.pending` — exact kind names or `domain.*` prefixes. `None` = all. +struct KindFilter(Option>); + +impl KindFilter { + fn parse(kinds: Option<&str>) -> Self { + let pats: Option> = kinds.map(|s| { + s.split(',') + .map(str::trim) + .filter(|p| !p.is_empty()) + .map(str::to_string) + .collect() + }); + KindFilter(pats.filter(|p| !p.is_empty())) + } + + fn matches(&self, kind: &str) -> bool { + match &self.0 { + None => true, + Some(pats) => pats.iter().any(|p| match p.strip_suffix(".*") { + Some(prefix) => kind + .strip_prefix(prefix) + .is_some_and(|rest| rest.starts_with('.')), + None => p == kind, + }), + } + } +} + +#[derive(Deserialize)] +pub(crate) struct EventsQuery { + /// Resume cursor: stream events with `seq > since`. The `Last-Event-ID` header (what an SSE + /// client sends on auto-reconnect) takes precedence when both are present. + since: Option, + /// Comma-separated kind filter (`stream.*,pairing.pending`). + kinds: Option, +} + +/// One [`crate::events::HostEvent`] as an SSE frame: `id:` = seq (drives `Last-Event-ID`), +/// `event:` = kind, `data:` = the full event JSON (kind included — the frame is self-contained +/// for consumers that read `data` only). +fn sse_event(ev: &crate::events::HostEvent) -> Event { + Event::default() + .id(ev.seq.to_string()) + .event(ev.kind.name()) + .data(serde_json::to_string(ev).unwrap_or_else(|_| "{}".to_string())) +} + +/// Everything the poll loop owns; dropping it (the response body going away) releases the +/// connection-cap slot and the broadcast receiver. +struct StreamState { + /// The dropped marker (when applicable) + the filtered catch-up, served before the live tail. + pending: VecDeque, + rx: broadcast::Receiver, + filter: KindFilter, + _slot: StreamSlot, +} + +/// Stream host lifecycle events (SSE) +/// +/// Server-Sent Events stream of the host's lifecycle events: client connect/disconnect, session +/// and stream start/end, pairing decisions, display create/release, library changes, host +/// start/stop — both protocol planes. Frames carry `id:` = the event's monotonic `seq`, +/// `event:` = its kind, and `data:` = the event JSON (schema-versioned, additive-only). +/// +/// Resume: standard `Last-Event-ID` (or `?since=`) replays from the in-memory ring; a consumer +/// that fell off the ring receives an `event: dropped` frame first and should resync via the +/// REST snapshots. Keep-alive comments are sent every 15 s. +#[utoipa::path( + get, + path = "/events", + tag = "events", + operation_id = "streamEvents", + params( + ("since" = Option, Query, description = "Resume cursor: only events with `seq` greater than this are sent (the ring keeps the newest ~1024). `Last-Event-ID` takes precedence."), + ("kinds" = Option, Query, description = "Comma-separated server-side kind filter: exact kinds (`pairing.pending`) or `domain.*` prefixes (`stream.*`)."), + ("Last-Event-ID" = Option, Header, description = "SSE auto-reconnect cursor — the `id:` of the last received frame."), + ), + responses( + (status = OK, description = "SSE stream; each frame's `data:` is one HostEvent", body = crate::events::HostEvent, content_type = "text/event-stream"), + (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError), + (status = SERVICE_UNAVAILABLE, description = "Concurrent event-stream cap reached — retry shortly", body = ApiError), + ) +)] +pub(crate) async fn stream_events(Query(q): Query, headers: HeaderMap) -> Response { + let Some(slot) = try_acquire_slot() else { + return api_error( + StatusCode::SERVICE_UNAVAILABLE, + "event-stream connection cap reached — close an existing stream or retry", + ); + }; + // The header is what an SSE client re-sends on auto-reconnect, so it is always the newer + // cursor when both it and the original URL's `?since=` are present. + let since = headers + .get("last-event-id") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.trim().parse::().ok()) + .or(q.since) + .unwrap_or(0); + let filter = KindFilter::parse(q.kinds.as_deref()); + let sub = crate::events::bus().subscribe(since); + + let mut pending = VecDeque::new(); + if sub.dropped { + // The consumer's cursor precedes the ring: tell it so (the `LogPage.dropped` contract) + // — its move is a REST resync, not trust in a complete replay. + pending.push_back( + Event::default() + .event("dropped") + .data(r#"{"dropped":true}"#), + ); + } + pending.extend( + sub.catch_up + .iter() + .filter(|ev| filter.matches(ev.kind.name())) + .map(sse_event), + ); + + let state = StreamState { + pending, + rx: sub.rx, + filter, + _slot: slot, + }; + let stream = futures_util::stream::unfold(state, |mut st| async move { + loop { + if let Some(ev) = st.pending.pop_front() { + return Some((Ok::<_, std::convert::Infallible>(ev), st)); + } + match st.rx.recv().await { + Ok(ev) => { + if st.filter.matches(ev.kind.name()) { + return Some((Ok(sse_event(&ev)), st)); + } + } + // Lagged = this consumer is too slow for the live tail: cut it loose (bounded + // memory, RFC §9.6) — its auto-reconnect resumes from the ring, which flags + // `dropped` if it also fell off that. Closed = host shutdown. + Err(broadcast::error::RecvError::Lagged(_)) + | Err(broadcast::error::RecvError::Closed) => return None, + } + } + }); + Sse::new(stream) + .keep_alive(KeepAlive::new().interval(KEEP_ALIVE)) + .into_response() +} + +#[cfg(test)] +pub(crate) mod test_support { + /// Fill the connection cap and hand the slots to a test ([`Drop`] frees them). The events + /// tests serialize on a lock so the full cap can't 503 an unrelated concurrent test. + pub(crate) fn saturate_slots() -> Vec { + std::iter::from_fn(super::try_acquire_slot).collect() + } +} + +#[cfg(test)] +mod unit_tests { + use super::*; + + #[test] + fn kind_filter_semantics() { + let all = KindFilter::parse(None); + assert!(all.matches("stream.started")); + + let f = KindFilter::parse(Some("stream.*, pairing.pending")); + assert!(f.matches("stream.started")); + assert!(f.matches("stream.stopped")); + assert!(f.matches("pairing.pending")); + assert!(!f.matches("pairing.completed")); + assert!(!f.matches("client.connected")); + // A prefix pattern must match on the dot boundary, not raw text. + assert!(!f.matches("streamx.started")); + + // Empty/blank filter strings mean "no filter", not "nothing matches". + assert!(KindFilter::parse(Some("")).matches("host.started")); + assert!(KindFilter::parse(Some(" , ")).matches("host.started")); + } +} diff --git a/crates/punktfunk-host/src/mgmt/tests.rs b/crates/punktfunk-host/src/mgmt/tests.rs index 0aa6a920..09c28435 100644 --- a/crates/punktfunk-host/src/mgmt/tests.rs +++ b/crates/punktfunk-host/src/mgmt/tests.rs @@ -946,3 +946,212 @@ async fn logs_endpoint_pages_by_cursor() { assert!(json["entries"].as_array().unwrap().is_empty()); assert_eq!(json["next"].as_u64().unwrap(), after); } + +// ------------------------------------------------------------------ events (SSE) + +/// Serializes the events-route tests: they share the process-global event bus and the +/// connection-cap counter, so the cap test must never 503 a concurrently running stream test. +static EVENTS_TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + +/// `get_req` + the default test bearer, pre-attached (these tests read streaming bodies +/// directly instead of going through `send`). +fn events_req(path: &str) -> axum::http::Request { + let mut req = get_req(path); + req.headers_mut().insert( + axum::http::header::AUTHORIZATION, + axum::http::HeaderValue::from_static("Bearer test-secret"), + ); + req +} + +/// The next SSE frame as text, or `None` when the stream ended / nothing arrived in time. +async fn next_sse_chunk(body: &mut Body) -> Option { + match tokio::time::timeout(std::time::Duration::from_secs(5), body.frame()).await { + Ok(Some(Ok(frame))) => frame + .into_data() + .ok() + .map(|b| String::from_utf8_lossy(&b).into_owned()), + _ => None, + } +} + +/// Every `data:` payload in accumulated SSE text, parsed as JSON. +fn sse_data_events(text: &str) -> Vec { + text.lines() + .filter_map(|l| l.strip_prefix("data: ")) + .filter_map(|d| serde_json::from_str(d).ok()) + .collect() +} + +#[tokio::test] +async fn events_stream_requires_bearer() { + let app = test_app(test_state(), None); + let mut req = get_req("/api/v1/events"); + req.headers_mut().insert( + axum::http::header::AUTHORIZATION, + axum::http::HeaderValue::from_static("Bearer wrong"), + ); + let resp = app.clone().oneshot(req).await.expect("infallible"); + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); +} + +/// The full consumer contract on one route: ring catch-up, the server-side kind filter, the +/// live tail on the same connection, `?since=`/`Last-Event-ID` resume, and the `dropped` +/// marker for a cursor that fell off the ring. +#[tokio::test] +async fn events_stream_catch_up_filter_resume_tail_and_dropped() { + use crate::events::EventKind; + let _l = EVENTS_TEST_LOCK.lock().await; + let app = test_app(test_state(), None); + let uniq = format!("evt-{}-{:p}", std::process::id(), &0u8 as *const u8); + let m1 = format!("{uniq}-one"); + + // Noise of a different kind (must be filtered out), then our marker. + crate::events::emit(EventKind::DisplayReleased { count: 424_242 }); + crate::events::emit(EventKind::LibraryChanged { source: m1.clone() }); + + let resp = app + .clone() + .oneshot(events_req("/api/v1/events?kinds=library.changed")) + .await + .expect("infallible"); + assert_eq!(resp.status(), StatusCode::OK); + let ctype = resp + .headers() + .get(axum::http::header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .unwrap_or_default() + .to_string(); + assert!( + ctype.starts_with("text/event-stream"), + "content-type: {ctype}" + ); + + // Catch-up must deliver m1 (other tests' library.changed events may interleave — scan). + let mut body = resp.into_body(); + let mut seen = String::new(); + while !seen.contains(&m1) { + let chunk = next_sse_chunk(&mut body) + .await + .expect("catch-up delivers the marker event"); + seen.push_str(&chunk); + } + assert!( + !seen.contains("event: display.released"), + "kind filter must drop other kinds: {seen}" + ); + assert!( + seen.contains("event: library.changed"), + "frame kind: {seen}" + ); + let m1_seq = sse_data_events(&seen) + .iter() + .find(|e| e["source"] == m1.as_str()) + .and_then(|e| e["seq"].as_u64()) + .expect("marker frame carries the full event JSON with its seq"); + + // Live tail on the SAME connection. If a concurrent test floods the broadcast channel the + // slow-consumer cut ends this stream — then the documented client move (reconnect with the + // last seen id) must deliver m2 instead, so follow it rather than flaking. + let m2 = format!("{uniq}-two"); + crate::events::emit(EventKind::LibraryChanged { source: m2.clone() }); + let mut tail = String::new(); + loop { + match next_sse_chunk(&mut body).await { + Some(chunk) => { + tail.push_str(&chunk); + if tail.contains(&m2) { + break; + } + } + None => { + let resp = app + .clone() + .oneshot(events_req(&format!( + "/api/v1/events?since={m1_seq}&kinds=library.changed" + ))) + .await + .expect("infallible"); + body = resp.into_body(); + } + } + } + drop(body); + + // Resume from m1's seq: m2 is caught up, m1 is not. + let resp = app + .clone() + .oneshot(events_req(&format!( + "/api/v1/events?since={m1_seq}&kinds=library.changed" + ))) + .await + .expect("infallible"); + let mut body = resp.into_body(); + let mut resumed = String::new(); + while !resumed.contains(&m2) { + let chunk = next_sse_chunk(&mut body) + .await + .expect("resume catch-up delivers m2"); + resumed.push_str(&chunk); + } + assert!(!resumed.contains(&m1), "since-cursor must exclude m1"); + drop(body); + + // Last-Event-ID beats ?since (it is the newer cursor on an SSE auto-reconnect). + let mut req = events_req("/api/v1/events?since=0&kinds=library.changed"); + req.headers_mut().insert( + "last-event-id", + axum::http::HeaderValue::from_str(&m1_seq.to_string()).unwrap(), + ); + let resp = app.clone().oneshot(req).await.expect("infallible"); + let mut body = resp.into_body(); + let mut resumed = String::new(); + while !resumed.contains(&m2) { + let chunk = next_sse_chunk(&mut body) + .await + .expect("header-resume catch-up delivers m2"); + resumed.push_str(&chunk); + } + assert!(!resumed.contains(&m1), "Last-Event-ID must exclude m1"); + drop(body); + + // A cursor that fell off the ring gets the dropped marker first. Flood the ring past + // capacity, then resume from seq 1. + for _ in 0..1100 { + crate::events::emit(EventKind::DisplayReleased { count: 1 }); + } + let resp = app + .clone() + .oneshot(events_req("/api/v1/events?since=1")) + .await + .expect("infallible"); + let mut body = resp.into_body(); + let first = next_sse_chunk(&mut body).await.expect("dropped marker"); + assert!(first.contains("event: dropped"), "first frame: {first}"); + assert!( + first.contains(r#"{"dropped":true}"#), + "marker data: {first}" + ); +} + +#[tokio::test] +async fn events_stream_connection_cap() { + let _l = EVENTS_TEST_LOCK.lock().await; + let app = test_app(test_state(), None); + + let slots = super::events::test_support::saturate_slots(); + let resp = app + .clone() + .oneshot(events_req("/api/v1/events")) + .await + .expect("infallible"); + assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE); + drop(slots); + + let resp = app + .clone() + .oneshot(events_req("/api/v1/events")) + .await + .expect("infallible"); + assert_eq!(resp.status(), StatusCode::OK, "cap frees with the slots"); +}