diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 8196e4270..519b431ab 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -79,7 +79,14 @@ jobs: # and cannot live in the proc-macro crate's own lib tests. # streamlib-engine's lib tests (host-body tier-1 + engine twin) are a # tracked follow-up, pending a fix to a parallel-run test flake. + # streamlib-api-server carries the control plane's route-surface, auth-gate + # and WebSocket frame locks, and ran nowhere in CI before. + # core::pubsub is named explicitly for the same reason python-wheel.yml + # names core::signals: the engine lib tests are excluded wholesale, and + # without this the subscription-live locks run nowhere. - name: Run unit tests run: | cargo test --locked -p streamlib -p streamlib-macros --lib cargo test --locked -p streamlib-engine --test attribute_macro_test + cargo test --locked -p streamlib-api-server --lib + cargo test --locked -p streamlib-engine --lib -- core::pubsub diff --git a/Cargo.lock b/Cargo.lock index f6d782ed4..e8718c52c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4756,6 +4756,7 @@ dependencies = [ "tempfile", "thiserror 2.0.18", "tokio", + "tokio-tungstenite", "tower", "tower-http", "tracing", diff --git a/docs/decisions/one-monotonic-clock.md b/docs/decisions/one-monotonic-clock.md index d8c2a4746..d500246b8 100644 --- a/docs/decisions/one-monotonic-clock.md +++ b/docs/decisions/one-monotonic-clock.md @@ -12,13 +12,18 @@ language, and before writing code that assumes a timestamp starts near zero. One concept — the machine's monotonic clock — in every language the project speaks, on the data plane. Scoped by the owner (2026-08-03) to what a processor stamps, reads, or -compares: frames, bags, audio ticks, `ctx.time`. Wall clock survives on exactly four -observability surfaces — log record `host_ts` and `source_ts`, log file naming, and the -control-plane pubsub event timestamp — because correlating with the outside world and +compares: frames, bags, audio ticks, `ctx.time`. Wall clock survives on exactly three +observability surfaces — log record `host_ts` and `source_ts`, and log file naming — +because correlating with the outside world and with other hosts' logs is a job monotonic time cannot do. Everything else is monotonic; a wall-clock value never enters the data plane and is never compared against a media -timestamp. Adding a fifth wall-clock surface is a plan change, not a judgement call, and +timestamp. Adding a further wall-clock surface is a plan change, not a judgement call, and `cargo xtask check-clock-usage` enforces the list mechanically. + +> ~~A fourth surface, the control-plane pubsub event timestamp, also keeps wall clock.~~ +> — Superseded 2026-08-13 by #1783. The control-plane event bus became an in-process +> registry, so its events no longer cross a wire and carry no timestamp to stamp. The +> surface ceased to exist rather than being retracted. Timestamps are raw `clock_gettime(CLOCK_MONOTONIC)` on Linux and `mach_absolute_time` on Apple: the same epoch V4L2 and ALSA stamp their buffers with, and the same value any other process on the host would read. No process-relative epoch, and exactly one diff --git a/docs/learnings/README.md b/docs/learnings/README.md index b1d0a83a7..ef8763b02 100644 --- a/docs/learnings/README.md +++ b/docs/learnings/README.md @@ -67,8 +67,11 @@ Avoid the two failure modes: Validate camera→display end-to-end via virtual camera + PNG sampling - [@docs/learnings/vulkanalia-empty-slice-cast.md](vulkanalia-empty-slice-cast.md) — Cryptic `Cast` trait error when passing `&[]` to vulkanalia Vulkan methods -- [@docs/learnings/pubsub-lazy-init-silent-noop.md](pubsub-lazy-init-silent-noop.md) — - Test hangs indefinitely because PUBSUB silently no-ops without `init()` +- ~~`pubsub-lazy-init-silent-noop.md`~~ — Removed 2026-08-13 by #1783. It taught that + `PUBSUB` silently no-ops until `init()` and prescribed a 150 ms sleep before publishing + in tests; the control-plane bus became an in-process registry with no `init`, no + buffering and no subscriber thread, so every instruction in it was wrong and the failure + it described cannot recur. - [@docs/learnings/cdylib-make-borrow-cached-fields.md](cdylib-make-borrow-cached-fields.md) — Plugin pipeline runs end-to-end clean but produces zero/black output when host-side `make_*_borrow` helpers leave the PluginAbiObject's cached diff --git a/docs/learnings/pubsub-lazy-init-silent-noop.md b/docs/learnings/pubsub-lazy-init-silent-noop.md deleted file mode 100644 index 4c61ccc29..000000000 --- a/docs/learnings/pubsub-lazy-init-silent-noop.md +++ /dev/null @@ -1,82 +0,0 @@ -# PUBSUB silently no-ops without init(), causing test hangs - -## Symptom - -A test that uses `PUBSUB.subscribe()` + `PUBSUB.publish()` hangs -indefinitely with no error output. The test thread never exits, no -panic, no timeout — just blocks forever on `handle.join()`. - -``` -running 1 test -test core::utils::loop_control::tests::test_shutdown_event_exits_loop ... -``` -(never completes) - -## Root cause - -`PUBSUB` is a `LazyLock` that uses `OnceLock` for its internal -`runtime_id` and iceoryx2 `node`. It is only fully functional after -`PUBSUB.init("name", node)` is called — which happens inside -`StreamRuntime::new()`. - -Without `init()`: -- `subscribe()` **buffers the subscription** (does not fail) -- `publish()` **silently drops the event** (does not fail) - -Combined with the common pattern of `thread::spawn(|| subscribe(...))` + -`publish(event)` + `handle.join()`, this creates an infinite hang: -- The subscriber thread opens an iceoryx2 service and waits for events -- The publish drops silently — the event never arrives -- `join()` blocks forever waiting for the thread to exit - -There are zero error messages or warnings. The test looks correct. The -hang is the only symptom. - -## Compound failure (iceoryx2 interaction) - -Even with PUBSUB initialized, a second failure mode exists: if the -iceoryx2 service is in `ServiceInCorruptedState` (from parallel test -teardown), the subscriber thread silently exits without receiving the -event. The event is published to... nothing. `join()` may complete -(thread exited) but the expected event was never received. - -## Fix (all three parts) - -1. **Initialize PUBSUB in the test** if a `StreamRuntime` isn't being created: -```rust -if let Ok(node) = Iceoryx2Node::new() { - PUBSUB.init("test-name", node); -} -``` - -2. **Use `mpsc::channel` + `recv_timeout` instead of `handle.join()`**: -```rust -let (done_tx, done_rx) = mpsc::channel(); -std::thread::spawn(move || { - let result = shutdown_aware_loop(|| { ... }); - done_tx.send(result).ok(); -}); - -// Publish the event... - -match done_rx.recv_timeout(Duration::from_secs(5)) { - Ok(result) => assert!(result.is_ok()), - Err(_) => panic!("loop did not exit within 5s — PUBSUB may not be initialized"), -} -``` - -3. **Allow setup time** — `std::thread::sleep(Duration::from_millis(150))` - between spawning the subscriber thread and publishing. The iceoryx2 - service open is async; publishing before the subscriber is listening - loses the event. - -## Where this hits - -Any test that uses PUBSUB events (shutdown, reconfigure, etc.) outside -of a full `StreamRuntime`. Currently: -- `runtime/streamlib-engine/src/core/utils/loop_control.rs` — `test_shutdown_event_exits_loop` - -## Reference -- Fix commit in #252 (ash → vulkanalia migration branch) -- PUBSUB implementation: `runtime/streamlib-engine/src/core/pubsub.rs` -- iceoryx2 node: `runtime/streamlib-engine/src/iceoryx2/mod.rs` diff --git a/docs/plan/ARCHITECTURE.md b/docs/plan/ARCHITECTURE.md index 42b1d8199..630a78414 100644 --- a/docs/plan/ARCHITECTURE.md +++ b/docs/plan/ARCHITECTURE.md @@ -234,11 +234,15 @@ Legend: **DECIDED** — build exactly this. **OPEN** — do not build; needs an (`CLOCK_MONOTONIC` on Linux, `mach_absolute_time` on Apple), the same epoch the V4L2 and ALSA driver stamps carry, comparable across every node on a host. No process-relative epoch anywhere, and each language exports exactly one name for it. - Wall clock is permitted on exactly four observability surfaces and nowhere else: log - record `host_ts` and `source_ts`, log file naming, and the control-plane pubsub event - timestamp — their job is correlating with the outside world, which monotonic time - cannot do. A wall-clock value never enters the data plane and is never compared against - a media timestamp; a fifth surface is a plan change, not a judgement call. + Wall clock is permitted on exactly three observability surfaces and nowhere else: log + record `host_ts` and `source_ts`, and log file naming — their job is correlating with + the outside world, which monotonic time cannot do. A wall-clock value never enters the + data plane and is never compared against a media timestamp; a further surface is a plan + change, not a judgement call. + ~~The control-plane pubsub event timestamp is a fourth permitted surface.~~ — Superseded + 2026-08-13 by #1783: the control-plane event bus became an in-process registry, so its + events no longer cross a wire and carry no timestamp to stamp. The surface ceased to + exist rather than being retracted; the rule it was an instance of is unchanged. [one-monotonic-clock] - **OPEN** — Audio backend: PipeWire-native on Linux is the intent (the current CPAL → ALSA path is interim); do not build until a research memo settles it. A/V diff --git a/docs/plan/changes/one-monotonic-clock.md b/docs/plan/changes/one-monotonic-clock.md index 40751dda8..7b604848e 100644 --- a/docs/plan/changes/one-monotonic-clock.md +++ b/docs/plan/changes/one-monotonic-clock.md @@ -70,10 +70,14 @@ driver stamp. Rust reaches it through `MediaClock`; Python through `monotonic_no siblings). 3. Log file naming — `started_at_millis` (`core/logging/init.rs:185`, `core/logging/paths.rs:22`) and the CLI's rendering of both (`commands/logs.rs:222`). -4. Control-plane pubsub event `timestamp_ns` (`core/pubsub/bus.rs:263-268`). +4. ~~Control-plane pubsub event `timestamp_ns` (`core/pubsub/bus.rs:263-268`).~~ — + Superseded 2026-08-13 by #1783: the control-plane event bus became an in-process + registry, so its events no longer cross a wire and carry no timestamp to stamp. The + surface ceased to exist rather than being retracted; `check-clock-usage` now permits + three. Their job is correlating StreamLib with the outside world and with other hosts' logs — a -job monotonic time cannot do. Adding a fifth surface is a plan change, not a judgement +job monotonic time cannot do. Adding a further surface is a plan change, not a judgement call. ### The rule that keeps the two from mixing diff --git a/runtime/streamlib-api-server/Cargo.toml b/runtime/streamlib-api-server/Cargo.toml index 2b9e1e429..4de7c41d3 100644 --- a/runtime/streamlib-api-server/Cargo.toml +++ b/runtime/streamlib-api-server/Cargo.toml @@ -62,3 +62,8 @@ streamlib = { path = "../../sdk/streamlib-sdk", version = "0.17.0", features = [ tempfile = "3" tower = {version = "0.5", features = ["util"]} serial_test = "3.2" +# Real WebSocket client for the /ws/events frame-ordering test: `tower::oneshot` +# drives a router but never completes an upgrade, so it cannot observe frames. +# Pinned to the generation axum itself speaks, so the test client and the server +# under test share one tungstenite rather than forking a second into the lock. +tokio-tungstenite = "0.29" diff --git a/runtime/streamlib-api-server/src/control_plane_stub_support.rs b/runtime/streamlib-api-server/src/control_plane_stub_support.rs index 5b525180f..c79f8491b 100644 --- a/runtime/streamlib-api-server/src/control_plane_stub_support.rs +++ b/runtime/streamlib-api-server/src/control_plane_stub_support.rs @@ -87,3 +87,31 @@ macro_rules! graph_mutation_ops_are_unreachable { } pub(crate) use graph_mutation_ops_are_unreachable; + +/// Implement the observation half of [`RuntimeOperations`] — the ops every stub +/// answers the same way — with an empty graph and an unreachable shutdown +/// naming `$who`. +/// +/// Paired with [`graph_mutation_ops_are_unreachable`]: between them a new +/// `RuntimeOperations` method is one edit here rather than several +/// near-identical stubs drifting apart across the crate's test modules. +macro_rules! observation_ops_answer_an_empty_graph { + ($who:literal) => { + fn to_json_async( + &self, + ) -> ::streamlib::sdk::runtime::BoxFuture< + '_, + ::streamlib::sdk::error::Result<::serde_json::Value>, + > { + Box::pin(async { Ok(::serde_json::json!({})) }) + } + fn to_json(&self) -> ::streamlib::sdk::error::Result<::serde_json::Value> { + Ok(::serde_json::json!({})) + } + fn request_runtime_shutdown(&self, _reason: &str) -> ::streamlib::sdk::error::Result<()> { + unreachable!(concat!($who, " never shuts the runtime down")) + } + }; +} + +pub(crate) use observation_ops_answer_an_empty_graph; diff --git a/runtime/streamlib-api-server/src/handlers.rs b/runtime/streamlib-api-server/src/handlers.rs index 74b62583c..727634e9b 100644 --- a/runtime/streamlib-api-server/src/handlers.rs +++ b/runtime/streamlib-api-server/src/handlers.rs @@ -8,7 +8,7 @@ use axum::{ extract::Path, extract::Query, extract::State, - extract::ws::{Message, WebSocket, WebSocketUpgrade}, + extract::ws::{CloseFrame, Message, WebSocket, WebSocketUpgrade}, http::StatusCode, response::IntoResponse, routing::{get, post}, @@ -17,6 +17,7 @@ use futures_util::{SinkExt, StreamExt}; use parking_lot::Mutex; use serde::Deserialize; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use streamlib::sdk::error::{Error, Result}; use streamlib::sdk::json_schema::{ProcessorDescriptorOutput, RegistryResponse}; use streamlib::sdk::processors::PROCESSOR_REGISTRY; @@ -262,10 +263,86 @@ pub(crate) async fn get_moq_catalog( Json(catalog) } +// ============================================================================ +// WebSocket subscription-live contract +// ============================================================================ + +/// Build a close frame, truncating the reason to what RFC 6455 permits. +/// +/// Every close on these sockets goes through here: the cap is a wire invariant +/// (tungstenite refuses an over-length control frame and the client gets an +/// abnormal close with no reason at all), and a per-site `truncate` call is an +/// invariant held by memory. +fn websocket_close_frame(code: u16, reason: impl Into) -> Message { + Message::Close(Some(CloseFrame { + code, + reason: truncate_on_char_boundary(reason.into(), MAX_WS_CLOSE_REASON_BYTES).into(), + })) +} + +/// Close code for a socket that could not be opened — the graph snapshot its +/// first frame carries could not be produced. App codes live in the 4000–4999 +/// private range, alongside the tap's 4404 / 4409. +const WS_CLOSE_CODE_STREAM_UNAVAILABLE: u16 = 4503; + +/// Close code for a client that fell too far behind its event stream. Distinct +/// from every other close so a client knows the cure is to reconnect and read +/// the fresh snapshot, not to retry blindly. +const WS_CLOSE_CODE_CLIENT_LAGGED: u16 = 4504; + +/// Events a client may fall behind by before it is closed as lagged. +/// +/// Bounded rather than unbounded because `publish` runs on the engine's threads: +/// an unbounded queue makes a slow socket the node's memory problem. Dropping the +/// client is acceptable only because reconnecting is lossless — it opens with a +/// fresh snapshot — which is the whole reason the stream leads with state. +const MAX_BUFFERED_EVENTS_PER_CLIENT: usize = 1024; + +/// The non-`Event` frame a control-plane WebSocket opens with. It precedes every +/// data frame on the socket and says the subscription behind it is attached. +/// +/// `/ws/events` is a best-effort realtime stream, not a record: it carries no +/// history and replays nothing. Current state is `GET /api/graph` and the +/// durable record is the JSONL log — a client that wants either asks for it, +/// which is what this frame makes safe to do. Waiting for it before reading the +/// graph composes without a gap, because the subscription is attached before the +/// read, so anything the read misses arrives as a following event. +/// +/// Wire contract: `Event` is an externally tagged enum, so an event frame is +/// always a single-key JSON object keyed by a variant name (`RuntimeGlobal`, +/// `ProcessorEvent`, `Custom`). These variants keep that grammar with keys that +/// are none of them, so a client discriminates on the key alone and a strict +/// `Event` decoder rejects one as an unknown variant rather than mis-reading it. +/// On `/ws/tap/{channel}` the separation is stronger still — this is the only +/// text frame that socket carries, and every bag stays a verbatim binary frame. +#[derive(serde::Serialize)] +enum ControlPlaneWebSocketOpeningFrame { + EventStreamSubscriptionLive { topic: String }, + TapSubscriptionLive { channel: String }, +} + +impl ControlPlaneWebSocketOpeningFrame { + /// Render as the text frame to put on the wire. + fn to_websocket_text_frame(&self) -> Result { + serde_json::to_string(self) + .map(|json| Message::Text(json.into())) + .map_err(|e| Error::Runtime(format!("opening frame could not be serialized: {e}"))) + } +} + // ============================================================================ // WebSocket Event Streaming // ============================================================================ +/// `GET /ws/events` — stream the node's runtime events, opening with the graph. +#[utoipa::path( + get, + path = "/ws/events", + tag = "events", + responses( + (status = 101, description = "WebSocket upgraded. Best-effort realtime event stream: it carries no history and replays nothing. The first frame is text — {\"EventStreamSubscriptionLive\":{\"topic\":\"*\"}} — sent once the subscription is attached; every frame after it is one runtime Event as JSON. Wait for that frame before reading GET /api/graph if you want current state, and the two compose without a gap: the subscription is attached before the read, so anything the read misses arrives as a following event. Current state is /api/graph and the durable record is the JSONL log — neither is served here. Frames are discriminated by their single top-level key: the opening frame's key is never an Event variant name (RuntimeGlobal / ProcessorEvent / Custom), so a strict Event decoder rejects it rather than mis-reading it. A client that falls more than 1024 events behind is closed with 4504 and should reconnect.") + ) +)] pub(crate) async fn websocket_handler(ws: WebSocketUpgrade) -> impl IntoResponse { ws.on_upgrade(handle_websocket) } @@ -274,15 +351,48 @@ async fn handle_websocket(socket: WebSocket) { let (mut sender, mut receiver) = socket.split(); // Channel to bridge sync EventListener -> async WebSocket - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); + let (tx, mut rx) = tokio::sync::mpsc::channel::(MAX_BUFFERED_EVENTS_PER_CLIENT); + let client_lagged = Arc::new(AtomicBool::new(false)); // Listener that forwards events to channel - let listener = Arc::new(Mutex::new(WebSocketEventForwarder { tx })); - - // Subscribe to ALL topics via wildcard + let listener = Arc::new(Mutex::new(WebSocketEventForwarder { + tx: Some(tx), + client_lagged: Arc::clone(&client_lagged), + })); + + // Subscribe to ALL topics via wildcard. Registration is synchronous, so + // every event caused from here on is queued for this socket — including any + // caused while the snapshot below is being taken. PUBSUB.subscribe(topics::ALL, listener.clone()); - tracing::info!("WebSocket client connected, subscribed to all events"); + // The subscription is attached above, so this frame reports a fact rather + // than a promise. axum runs this callback after the 101, so the upgrade + // alone cannot tell a client that — which is the whole reason a first frame + // exists on a stream that otherwise just streams. + let opening_frame = ControlPlaneWebSocketOpeningFrame::EventStreamSubscriptionLive { + topic: topics::ALL.to_string(), + } + .to_websocket_text_frame(); + + let opening_frame = match opening_frame { + Ok(opening_frame) => opening_frame, + Err(e) => { + tracing::error!("WebSocket opening frame could not be built: {e}"); + let _ = sender + .send(websocket_close_frame( + WS_CLOSE_CODE_STREAM_UNAVAILABLE, + format!("event stream could not start: {e}"), + )) + .await; + return; + } + }; + + if sender.send(opening_frame).await.is_err() { + return; + } + + tracing::info!("WebSocket client connected, subscription to all events is live"); // Task: forward channel events to WebSocket let send_task = tokio::spawn(async move { @@ -290,7 +400,7 @@ async fn handle_websocket(socket: WebSocket) { match serde_json::to_string(&event) { Ok(json) => { if sender.send(Message::Text(json.into())).await.is_err() { - break; + return; } } Err(e) => { @@ -298,20 +408,41 @@ async fn handle_websocket(socket: WebSocket) { } } } + + // The loop ends only when every sender is gone, and the forwarder drops + // its sender exactly when it latches the lag — so the flag is read after + // the fact rather than raced against the drain. + if client_lagged.load(Ordering::Relaxed) { + tracing::info!("WebSocket client fell behind its event stream, closing"); + let _ = sender + .send(websocket_close_frame( + WS_CLOSE_CODE_CLIENT_LAGGED, + "client lagged; reconnect for a fresh snapshot", + )) + .await; + } }); - // Receive loop (keep-alive, handle close) - while let Some(msg) = receiver.next().await { - match msg { - Ok(Message::Close(_)) => { - tracing::info!("WebSocket client closed connection"); - break; - } - Err(e) => { - tracing::warn!("WebSocket error: {}", e); - break; - } - _ => {} // axum handles ping/pong automatically + // Keep-alive / close, raced against the send task: that task ends when it + // has closed the socket itself, and a client closed for lagging is exactly + // the one that may never answer the close handshake — waiting only on the + // client would keep this subscription alive cloning events for a corpse. + let mut send_task = send_task; + loop { + tokio::select! { + _ = &mut send_task => break, + message = receiver.next() => match message { + Some(Ok(Message::Close(_))) | None => { + tracing::info!("WebSocket client closed connection"); + break; + } + Some(Err(e)) => { + tracing::warn!("WebSocket error: {}", e); + break; + } + // axum handles ping/pong automatically + Some(Ok(_)) => {} + }, } } @@ -322,12 +453,36 @@ async fn handle_websocket(socket: WebSocket) { } struct WebSocketEventForwarder { - tx: tokio::sync::mpsc::UnboundedSender, + /// Taken when the client is found lagging, which closes the channel and + /// ends the send task's drain — the signal the task acts on. + tx: Option>, + client_lagged: Arc, } impl EventListener for WebSocketEventForwarder { + /// Hands the event to the socket's send task without blocking: `on_event` + /// runs on the engine thread that published, so it must never wait on a + /// client. A full queue latches the lag rather than dropping the event + /// quietly — the socket then closes and the client re-snapshots. fn on_event(&mut self, event: &Event) -> Result<()> { - let _ = self.tx.send(event.clone()); + // Reserve before cloning: a lagging client would otherwise pay a full + // Event clone per publish only to have it dropped. The borrow on `tx` + // ends with the permit, so the sender can be taken below. + let queue_is_full = match self.tx.as_ref() { + Some(tx) => match tx.try_reserve() { + Ok(permit) => { + permit.send(event.clone()); + false + } + Err(_) => true, + }, + None => return Ok(()), + }; + + if queue_is_full { + self.client_lagged.store(true, Ordering::Relaxed); + self.tx = None; + } Ok(()) } } @@ -360,7 +515,7 @@ pub(crate) struct TapQuery { ("count" = Option, Query, description = "Stream exactly this many bags then close; absent streams live until the client disconnects") ), responses( - (status = 101, description = "WebSocket upgraded. Read-only observability tap: each channel bag is forwarded verbatim (FrameHeader-framed) as a binary WS frame with no encode, containerize, or transcode — decoding is the client's concern. To observe a viewable video feed, tap an encoded (h264/h265/jpeg) or container (CMAF/fMP4) channel; a raw video channel carries zero-copy DMA-BUF/VkImage frame descriptors (meaningless off-host), not pixels, and this is not a realtime-video transport (use the WebRTC/MoQ/display processors)."), + (status = 101, description = "WebSocket upgraded. The first frame is text — {\"TapSubscriptionLive\":{\"channel\":\"…\"}} — sent once the tap is attached, so a client can act without racing the attach; it is the only text frame the socket carries. Every frame after it is a channel bag forwarded verbatim (FrameHeader-framed) as a binary WS frame with no encode, containerize, or transcode — decoding is the client's concern. To observe a viewable video feed, tap an encoded (h264/h265/jpeg) or container (CMAF/fMP4) channel; a raw video channel carries zero-copy DMA-BUF/VkImage frame descriptors (meaningless off-host), not pixels, and this is not a realtime-video transport (use the WebRTC/MoQ/display processors)."), (status = 401, description = "Missing or malformed bearer token", body = UnauthorizedResponse), (status = 403, description = "Invalid bearer token", body = ForbiddenResponse) ) @@ -401,25 +556,45 @@ async fn handle_tap_websocket( tracing::info!(channel = %channel, "tap client attached"); - // Own the subscription in this scope: forward bags until the tap ends - // (bounded count reached / channel gone) or the client disconnects. - loop { - tokio::select! { - maybe_bag = subscription.recv() => match maybe_bag { - Some(bytes) => { - if sender.send(Message::Binary(bytes.into())).await.is_err() { + // `tap_async` resolves only once the tap's subscriber exists, so this frame + // reports an attach that has already happened — a client that acts on it + // cannot race the attach, and can tell "attached" from "attached but idle". + let live_frame_was_sent = match (ControlPlaneWebSocketOpeningFrame::TapSubscriptionLive { + channel: channel.clone(), + }) + .to_websocket_text_frame() + { + Ok(live_frame) => sender.send(live_frame).await.is_ok(), + Err(e) => { + tracing::error!(channel = %channel, "tap live frame could not be sent: {e}"); + false + } + }; + + // Forwarding is skipped by falling through rather than returning: the + // detach below must stay on the path out, because dropping the subscription + // here would join an OS thread on an async worker. + if live_frame_was_sent { + // Own the subscription in this scope: forward bags until the tap ends + // (bounded count reached / channel gone) or the client disconnects. + loop { + tokio::select! { + maybe_bag = subscription.recv() => match maybe_bag { + Some(bytes) => { + if sender.send(Message::Binary(bytes.into())).await.is_err() { + break; + } + } + None => { + let _ = sender.send(Message::Close(None)).await; break; } - } - None => { - let _ = sender.send(Message::Close(None)).await; - break; - } - }, - maybe_msg = receiver.next() => match maybe_msg { - Some(Ok(Message::Close(_))) | Some(Err(_)) | None => break, - _ => {} - }, + }, + maybe_msg = receiver.next() => match maybe_msg { + Some(Ok(Message::Close(_))) | Some(Err(_)) | None => break, + _ => {} + }, + } } } @@ -434,7 +609,7 @@ async fn handle_tap_websocket( tracing::info!(channel = %channel, "tap client detached"); } -/// Longest tap close reason RFC 6455 permits: a control frame caps its payload +/// Longest close reason RFC 6455 permits: a control frame caps its payload /// at 125 bytes and the 2-byte close code consumes the first two, leaving 123 /// for the UTF-8 reason. tungstenite refuses to write an over-length close /// frame, so an untruncated tap error string (`NotSupported` runs ~180 bytes) @@ -835,3 +1010,281 @@ mod router_surface_and_auth_gate_tests { ); } } + +#[cfg(test)] +mod websocket_subscription_live_frame_tests { + //! The subscription-live contract on the control plane's WebSockets. + //! + //! axum runs the upgrade callback after the 101, so the upgrade alone tells + //! a client nothing about whether anything is listening on its behalf yet. + //! These cover what the socket says instead: an opening frame that is + //! unambiguously not an `Event`, arriving before any event does, and + //! carrying the state a client would otherwise have to infer from deltas. + + use super::*; + use crate::control_plane_stub_support::observation_ops_answer_an_empty_graph; + use std::time::Duration; + use streamlib::sdk::pubsub::RuntimeEvent; + use streamlib::sdk::runtime::{BoxFuture, TapSubscription}; + + /// The `Event` variant names an event frame can be keyed by. An opening + /// frame keyed by any of these would be ambiguous with a real event. + const EVENT_VARIANT_KEYS: &[&str] = &["RuntimeGlobal", "ProcessorEvent", "Custom"]; + + /// Hands out one preset `TapSubscription`, so `/ws/tap/{channel}` can be + /// driven over a real socket with no engine behind it. + /// + /// Separate from [`EventStreamStubRuntime`] for its `tap_async` alone; the + /// observation ops both answer come from one macro so a new + /// `RuntimeOperations` method is one edit rather than several diverging + /// ones. + struct TapStubRuntime { + subscription: Mutex>, + } + + impl RuntimeOperations for TapStubRuntime { + observation_ops_answer_an_empty_graph!("the tap test"); + + fn tap_async( + &self, + channel: String, + _count: Option, + ) -> BoxFuture<'_, Result> { + let taken = self.subscription.lock().take(); + Box::pin(async move { taken.ok_or(Error::TapSlotOccupied(channel)) }) + } + + crate::control_plane_stub_support::graph_mutation_ops_are_unreachable!("route"); + } + + /// Serve the real router on a loopback ephemeral port, returning the port + /// and the task to abort when done. + /// + /// A real bind rather than `tower::oneshot`, which drives a router but never + /// completes an upgrade — so it can never observe a frame. + fn serve_on_ephemeral_port( + runtime: Arc, + ) -> (u16, tokio::task::JoinHandle<()>) { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port"); + listener + .set_nonblocking(true) + .expect("the tokio listener needs a non-blocking socket"); + let port = listener.local_addr().expect("local addr").port(); + let listener = + tokio::net::TcpListener::from_std(listener).expect("adopt the bound listener"); + + let router = build_router( + runtime, + None, + #[cfg(feature = "moq")] + "test-runtime-id".to_string(), + ); + let server = tokio::spawn(async move { + let _ = axum::serve(listener, router).await; + }); + (port, server) + } + + /// How long a test waits for a frame the server should already be sending. + /// Every read is bounded by it: a hung suite reports nothing, a red one + /// names the frame that never came. + const FRAME_ARRIVAL_BUDGET: Duration = Duration::from_secs(5); + + /// `/ws/events` reads nothing off the runtime — it subscribes to `PUBSUB` + /// directly — but `build_router` needs one. + struct EventStreamStubRuntime; + + impl RuntimeOperations for EventStreamStubRuntime { + observation_ops_answer_an_empty_graph!("the event-stream test"); + + fn tap_async( + &self, + channel: String, + _count: Option, + ) -> BoxFuture<'_, Result> { + Box::pin(async move { Err(Error::TapChannelNotFound(channel)) }) + } + + crate::control_plane_stub_support::graph_mutation_ops_are_unreachable!("route"); + } + + fn opening_frame_json(frame: &ControlPlaneWebSocketOpeningFrame) -> serde_json::Value { + serde_json::to_value(frame).expect("opening frame serializes") + } + + /// The wire contract that lets a client tell the live frame from an event + /// without out-of-band knowledge: both are single-key objects, and the live + /// frame's key is none of `Event`'s. + #[test] + fn an_opening_frame_is_never_mistakable_for_an_event() { + let frames = [ + ControlPlaneWebSocketOpeningFrame::EventStreamSubscriptionLive { + topic: topics::ALL.to_string(), + }, + ControlPlaneWebSocketOpeningFrame::TapSubscriptionLive { + channel: "some-processor/some-output".to_string(), + }, + ]; + + for frame in &frames { + let json = opening_frame_json(frame); + let object = json.as_object().expect("opening frame is a JSON object"); + assert_eq!( + object.len(), + 1, + "an opening frame keeps Event's single-key grammar: {json}" + ); + + let key = object.keys().next().expect("single key"); + assert!( + !EVENT_VARIANT_KEYS.contains(&key.as_str()), + "opening frame key '{key}' collides with an Event variant" + ); + + // The other direction: a strict `Event` decoder must reject it + // outright rather than mis-read it as some event. + assert!( + serde_json::from_value::(json.clone()).is_err(), + "an opening frame must not deserialize as an Event: {json}" + ); + } + } + + /// Every event frame stays decodable as an `Event` — the opening frame is + /// prepended, and no envelope is wrapped around the events themselves. + #[test] + fn an_event_frame_is_unchanged_by_the_opening_frame() { + let event = Event::RuntimeGlobal(RuntimeEvent::GraphDidChange); + let encoded = serde_json::to_string(&event).expect("event serializes"); + + let decoded: Event = serde_json::from_str(&encoded).expect("event frame decodes as Event"); + assert_eq!(decoded, event); + } + + /// The guarantee a client acts on: once the opening frame arrives, nothing + /// published after it is missed. + /// + /// The publish below is not racing anything — `subscribe` registers + /// synchronously inside the handler, before the frame goes out — so this + /// locks the wire contract, and the engine's `core::pubsub::integration_tests` + /// lock the delivery guarantee it rests on. + /// + /// The publish below is not racing anything — `subscribe` registers + /// synchronously inside the handler — so this locks the wire contract, and + /// the engine's `core::pubsub::integration_tests` lock the delivery + /// guarantee it rests on. + /// + /// `#[serial]`: this test publishes, and `PUBSUB` is process-global — an + /// unserialized publish here lands inside the sample window of any other + /// test reading the same bus. + #[tokio::test] + #[serial_test::serial] + async fn ws_events_sends_the_subscription_live_frame_before_any_event() { + use tokio_tungstenite::tungstenite::Message as ClientMessage; + + let (port, server) = serve_on_ephemeral_port(Arc::new(EventStreamStubRuntime)); + + let (mut socket, _) = + tokio_tungstenite::connect_async(format!("ws://127.0.0.1:{port}/ws/events")) + .await + .expect("WebSocket upgrade on /ws/events"); + + // Frame 1 says the subscription is attached, before anything is published. + // Bounded like every other read here: a regression that stops the frame + // being sent must name itself, not hang the suite until the job's cap. + let first = tokio::time::timeout(FRAME_ARRIVAL_BUDGET, socket.next()) + .await + .expect("a first frame within the budget") + .expect("a first frame") + .expect("first frame is not an error"); + let ClientMessage::Text(first) = first else { + panic!("the first frame must be text, got {first:?}"); + }; + let first: serde_json::Value = serde_json::from_str(&first).expect("first frame is JSON"); + assert!( + first.get("EventStreamSubscriptionLive").is_some(), + "first frame must be the subscription-live frame, got {first}" + ); + + let published = Event::RuntimeGlobal(RuntimeEvent::GraphDidChange); + PUBSUB.publish(&published.topic(), &published); + + let next = tokio::time::timeout(FRAME_ARRIVAL_BUDGET, socket.next()) + .await + .expect("an event frame within the budget") + .expect("a second frame") + .expect("second frame is not an error"); + let ClientMessage::Text(next) = next else { + panic!("an event frame must be text, got {next:?}"); + }; + let received: Event = serde_json::from_str(&next) + .expect("the frame after the live frame decodes as an Event"); + assert_eq!(received, published); + + let _ = socket.close(None).await; + server.abort(); + } + + /// The tap socket's half of the same contract: a text live frame, then the + /// bag — verbatim and binary. + /// + /// The OpenAPI 101 description promises exactly this shape, and nothing + /// else proves the server sends it: the frame-grammar test only serializes + /// the enum, and the MCP tap tool never touches this route. + #[tokio::test] + async fn ws_tap_sends_the_live_frame_before_any_bag() { + use tokio_tungstenite::tungstenite::Message as ClientMessage; + + const BAG: &[u8] = b"\x00\x01\x02 not-an-event"; + + let (bag_sender, bag_receiver) = tokio::sync::mpsc::channel(1); + bag_sender.send(BAG.to_vec()).await.expect("queue one bag"); + + let (port, server) = serve_on_ephemeral_port(Arc::new(TapStubRuntime { + subscription: Mutex::new(Some(TapSubscription::from_forward_channel( + "some-processor/some-output".to_string(), + bag_receiver, + 0, + ))), + })); + + let (mut socket, _) = tokio_tungstenite::connect_async(format!( + "ws://127.0.0.1:{port}/ws/tap/some-processor%2Fsome-output" + )) + .await + .expect("WebSocket upgrade on /ws/tap/{channel}"); + + let first = tokio::time::timeout(FRAME_ARRIVAL_BUDGET, socket.next()) + .await + .expect("a first frame within the budget") + .expect("a first frame") + .expect("first frame is not an error"); + let ClientMessage::Text(first) = first else { + panic!("the tap's first frame must be text, got {first:?}"); + }; + let first: serde_json::Value = serde_json::from_str(&first).expect("first frame is JSON"); + assert_eq!( + first + .get("TapSubscriptionLive") + .and_then(|live| live.get("channel")) + .and_then(serde_json::Value::as_str), + Some("some-processor/some-output"), + "first frame must be the tap's subscription-live frame, got {first}" + ); + + // The bag is binary and byte-identical: the live frame is prepended to + // the stream, never wrapped around what the channel carries. + let next = tokio::time::timeout(FRAME_ARRIVAL_BUDGET, socket.next()) + .await + .expect("a bag frame within the budget") + .expect("a second frame") + .expect("second frame is not an error"); + let ClientMessage::Binary(bag) = next else { + panic!("a bag must arrive as a binary frame, got {next:?}"); + }; + assert_eq!(bag.as_ref(), BAG, "the bag must be forwarded verbatim"); + + let _ = socket.close(None).await; + server.abort(); + } +} diff --git a/runtime/streamlib-api-server/src/mcp.rs b/runtime/streamlib-api-server/src/mcp.rs index 6f36e804c..ee657642b 100644 --- a/runtime/streamlib-api-server/src/mcp.rs +++ b/runtime/streamlib-api-server/src/mcp.rs @@ -342,6 +342,8 @@ async fn call_logs(runtime: &Arc, arguments: Value) -> Va let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); let listener = Arc::new(Mutex::new(McpEventForwarder { tx })); + // Registration is synchronous, so the sample window below starts against a + // subscription that can already receive. PUBSUB.subscribe(topics::ALL, listener.clone()); let mut events: Vec = Vec::with_capacity(sample); @@ -440,8 +442,11 @@ fn hex_encode(bytes: &[u8]) -> String { hex } -/// Forwards runtime events into the `logs` tool's bounded collection channel, -/// mirroring the REST WebSocket event forwarder. +/// Forwards runtime events into the `logs` tool's bounded collection channel. +/// +/// Unbounded, unlike the WebSocket forwarder: this listener is drained +/// continuously for one bounded sample window and then dropped, so its queue +/// cannot outlive the call the way a long-lived socket's can. struct McpEventForwarder { tx: tokio::sync::mpsc::UnboundedSender, } @@ -884,11 +889,12 @@ mod tests { } #[tokio::test] + #[serial_test::serial] async fn tools_call_logs_returns_bounded_window_sample() { - // Hermetic: PUBSUB is uninitialized here, so no event is delivered and - // the collection is bounded by the monotonic sample window, returning an - // empty sample rather than hanging. Live event delivery rides iceoryx2 - // and is exercised by the engine's pubsub integration tests, not here. + // Nobody publishes during this window, so the tool collects nothing and + // the monotonic sample window bounds the wait rather than letting it + // hang. `#[serial]` keeps another test's publish out of the window — + // `PUBSUB` is process-global. let started = tokio::time::Instant::now(); let (status, body) = mcp_call( Arc::new(ControlPlaneMcpDispatchStubRuntime::new()), diff --git a/runtime/streamlib-engine/src/core/pubsub/bus.rs b/runtime/streamlib-engine/src/core/pubsub/bus.rs index 517565243..8542e8187 100644 --- a/runtime/streamlib-engine/src/core/pubsub/bus.rs +++ b/runtime/streamlib-engine/src/core/pubsub/bus.rs @@ -1,38 +1,86 @@ // Copyright (c) 2025 Jonathan Fontanez // SPDX-License-Identifier: BUSL-1.1 -use parking_lot::Mutex; -use std::cell::RefCell; -use std::collections::HashMap; -use std::sync::{Arc, LazyLock, OnceLock, Weak}; +use parking_lot::{Mutex, RwLock}; +use std::sync::mpsc::{Receiver, SyncSender, sync_channel}; +use std::sync::{Arc, LazyLock, Weak}; use super::events::{Event, EventListener, topics}; -use crate::iceoryx2::{EventPayload, Iceoryx2EventService, Iceoryx2Node, MAX_EVENT_PAYLOAD_SIZE}; -type EventPublisher = - iceoryx2::port::publisher::Publisher; +/// Process-wide pub/sub handle. +pub static PUBSUB: LazyLock = LazyLock::new(PubSub::new); + +/// Stated once so the release log and the debug assertion cannot drift apart. +const TEMPORARY_ARC_SUBSCRIBE_DIAGNOSIS: &str = "the listener will be dropped immediately and never receive events. Store the Arc \ + in a variable that outlives the subscription."; + +/// Deliveries a publisher may run ahead of the dispatcher before it blocks. +/// +/// Bounded so a runaway publisher cannot grow the queue without limit. Reaching +/// it means the dispatcher is starved, which for a control-plane bus carrying +/// lifecycle events means a listener is violating its no-blocking contract — +/// back-pressuring the publisher is the honest response, and it is visible, +/// where dropping would not be. +const MAX_QUEUED_DELIVERIES: usize = 4096; + +/// One listener's registration: the topic it asked for, and a weak handle to it. +/// +/// Weak, so dropping the caller's `Arc` unsubscribes with no bookkeeping — the +/// entry is pruned by the next publish that finds it dead. +struct PubSubTopicSubscription { + topic: String, + listener_weak: Weak>, +} -thread_local! { - /// Per-thread cache of iceoryx2 publishers keyed by service name. +impl PubSubTopicSubscription { + /// Whether an event published to `published_topic` reaches this listener. /// - /// iceoryx2 Publisher uses Rc internally (!Send), so it cannot be stored - /// in shared state. thread_local satisfies the !Send constraint while - /// keeping publishers alive so sent samples remain in shared memory - /// for subscribers to receive. - static PUBLISHER_CACHE: RefCell> = - RefCell::new(HashMap::new()); + /// A wildcard subscriber matches everything, every other subscriber matches + /// its own topic, and either way a subscription is visited once — a wildcard + /// listener does not also receive a second copy through the specific topic. + fn receives(&self, published_topic: &str) -> bool { + self.topic == topics::ALL || self.topic == published_topic + } } -/// Process-wide pub/sub handle. -pub static PUBSUB: LazyLock = LazyLock::new(PubSub::new); +/// Work the dispatcher thread drains in order. +enum PubSubDispatch { + /// One event and the listeners it was addressed to when it was published. + /// + /// Recipients are resolved at publish time, not delivery time, so a listener + /// receives exactly the events published after it subscribed — never one + /// that was already in flight when it arrived. + Deliver { + topic: String, + event: Event, + recipients: Vec>>, + }, + /// Acknowledges once everything queued before it has been delivered. + Barrier(SyncSender<()>), +} -/// iceoryx2-backed pub/sub for runtime events. +/// In-process pub/sub for control-plane events. +/// +/// Subscribing is synchronous: a registration is visible to the next publish on +/// any thread the moment [`PubSub::subscribe`] returns, so an event caused after +/// subscribing is delivered. There is no service to open, no connection to +/// establish, and no window in which a subscription exists but cannot receive. +/// +/// Publishing is a queue-and-return. Every event goes through one FIFO, so all +/// listeners observe one order — the order events were published in — rather +/// than an order that depends on which thread happened to publish. It also means +/// no listener ever runs on an engine thread: the engine publishes from inside +/// its own graph write lock, and running a callback there would put a listener +/// underneath a lock it knows nothing about. +/// +/// Control plane only, and in-process by construction: every publisher and every +/// listener lives in the app process, and an out-of-process observer reads the +/// control plane's `/ws/events` rather than this. Cross-process data movement is +/// the iceoryx2 channel plane, which this does not touch. pub struct PubSub { - // Set once via init() - runtime_id: OnceLock, - node: OnceLock, - // Subscriptions registered before init() — replayed when init() is called - pending_subscriptions: Mutex>)>>, + subscriptions: RwLock>, + dispatch_sender: SyncSender, + dispatcher: Mutex>>, } impl Default for PubSub { @@ -43,363 +91,165 @@ impl Default for PubSub { impl PubSub { pub fn new() -> Self { - Self { - runtime_id: OnceLock::new(), - node: OnceLock::new(), - pending_subscriptions: Mutex::new(Vec::new()), - } - } - - /// Initialize with iceoryx2 backend. Called once from Runner::new(). - /// - /// Replays any subscriptions that were registered before initialization. - pub fn init(&self, runtime_id: &str, node: Iceoryx2Node) { - let _ = self.runtime_id.set(runtime_id.to_string()); - let _ = self.node.set(node); + let (dispatch_sender, dispatch_receiver) = sync_channel(MAX_QUEUED_DELIVERIES); + let dispatcher = std::thread::Builder::new() + .name("pubsub-dispatch".to_string()) + .spawn(move || run_dispatch_loop(dispatch_receiver)) + .inspect_err(|e| tracing::error!("Failed to spawn the pubsub dispatcher: {}", e)) + .ok(); - tracing::info!("PUBSUB initialized for runtime '{}'", runtime_id); - - // Replay pending subscriptions - let pending = std::mem::take(&mut *self.pending_subscriptions.lock()); - for (topic, listener) in pending { - tracing::debug!("Replaying pending subscription for topic '{}'", topic); - self.subscribe_inner(&topic, listener); + Self { + subscriptions: RwLock::new(Vec::new()), + dispatch_sender, + dispatcher: Mutex::new(dispatcher), } } /// Subscribe a listener to a topic. /// - /// The subscriber thread holds only a Weak reference to the listener. - /// The caller MUST keep the Arc alive for the lifetime of the subscription. - /// When the Arc is dropped, the subscriber thread exits automatically. - /// - /// ```ignore - /// // WRONG — Arc dropped immediately, listener never receives events: - /// PUBSUB.subscribe(topic, Arc::new(Mutex::new(listener))); - /// - /// // RIGHT — Arc stored, subscription lives until variable is dropped: - /// let sub = Arc::new(Mutex::new(listener)); - /// PUBSUB.subscribe(topic, Arc::clone(&sub)); - /// ``` + /// The registry holds only a `Weak`, so the caller MUST keep the `Arc` alive + /// for the lifetime of the subscription; dropping it unsubscribes. pub fn subscribe(&self, topic: &str, listener: Arc>) { - // Caller must keep a strong Arc — we only store a Weak in the - // subscriber thread. strong_count == 1 means this parameter is the - // only reference and will be dropped when this call returns. - debug_assert!( - Arc::strong_count(&listener) > 1, - "PUBSUB.subscribe() called with a temporary Arc for topic '{}' — \ - the listener will be dropped immediately and never receive events. \ - Store the Arc in a variable that outlives the subscription.", - topic, - ); + // Caller must keep a strong Arc — the registry stores only a Weak. + // strong_count == 1 means this parameter is the only reference and will + // be dropped when this call returns. if Arc::strong_count(&listener) <= 1 { tracing::error!( - "PUBSUB.subscribe() called with a temporary Arc for topic '{}' — \ - the listener will be dropped immediately and never receive events", + "PUBSUB.subscribe() called with a temporary Arc for topic '{}' — {}", topic, + TEMPORARY_ARC_SUBSCRIBE_DIAGNOSIS, ); - } - - if self.runtime_id.get().is_none() { - // Not yet initialized — buffer for replay - tracing::debug!( - "PUBSUB not initialized, buffering subscription for '{}'", - topic + debug_assert!( + false, + "PUBSUB.subscribe() called with a temporary Arc for topic '{}' — {}", + topic, TEMPORARY_ARC_SUBSCRIBE_DIAGNOSIS, ); - self.pending_subscriptions - .lock() - .push((topic.to_string(), listener)); - return; } - self.subscribe_inner(topic, listener); - } - - fn subscribe_inner(&self, topic: &str, listener: Arc>) { - let runtime_id = self.runtime_id.get().unwrap().clone(); - let node = self.node.get().unwrap().clone(); - let weak_listener = Arc::downgrade(&listener); - let topic_owned = topic.to_string(); - - let service_name = topic_to_service_name(&runtime_id, topic); - let service_name_for_log = service_name.clone(); - - // Spawn a dedicated OS thread for polling. - // iceoryx2 Subscriber uses Rc internally (!Send), so it must be - // created and used on the same thread. - let builder = std::thread::Builder::new().name(format!("pubsub-{}", topic)); - if let Err(e) = builder.spawn(move || { - // Retry `open_or_create` — iceoryx2 can transiently report - // `ServiceInCorruptedState` when a concurrent node (e.g. another - // streamlib process or another test binary on the same machine) - // is scanning/cleaning dead-node state under `/tmp/iceoryx2/`. - // The state stabilizes within a few tens of milliseconds. - let mut service = None; - for attempt in 0..10 { - match node.open_or_create_event_service(&service_name) { - Ok(s) => { - service = Some(s); - break; - } - Err(e) => { - tracing::warn!( - "Failed to create event service '{}' (attempt {}): {}", - service_name, - attempt + 1, - e - ); - std::thread::sleep(std::time::Duration::from_millis(20)); - } - } - } - let Some(service) = service else { - tracing::error!( - "Giving up after 10 attempts to create event service '{}'", - service_name - ); - return; - }; + self.subscriptions.write().push(PubSubTopicSubscription { + topic: topic.to_string(), + listener_weak: Arc::downgrade(&listener), + }); - let subscriber = match service.create_subscriber() { - Ok(s) => s, - Err(e) => { - tracing::error!("Failed to create subscriber for '{}': {}", service_name, e); - return; - } - }; + tracing::debug!("Listener subscribed to topic '{}'", topic); + } - subscriber_poll_loop(&subscriber, &weak_listener, &topic_owned); - }) { - tracing::error!( - "Failed to spawn subscriber thread for '{}': {}", - service_name_for_log, - e - ); - } else { - tracing::debug!( - "Listener subscribed to topic '{}' (service: {})", - topic, - service_name_for_log - ); - } + /// How many registrations the bus is holding, live or not yet pruned. + #[cfg(test)] + pub(crate) fn registration_count(&self) -> usize { + self.subscriptions.read().len() } - /// Publish event to topic (serializes and sends via iceoryx2). + /// Queue an event for every listener subscribed to `topic` or to + /// [`topics::ALL`]. /// - /// Events are dispatched to: - /// 1. All subscribers of the specific topic - /// 2. All subscribers of `topics::ALL` (wildcard) + /// Returns once the event is queued, not once it is delivered. Ordering is + /// the guarantee: events are delivered in the order they were published + /// here, and every listener observes that same order. pub fn publish(&self, topic: &str, event: &Event) { - let Some(runtime_id) = self.runtime_id.get() else { - tracing::trace!( - "PUBSUB not initialized, dropping event: {}", - event.log_name() - ); - return; - }; - - // Serialize event to MessagePack - let bytes = match rmp_serde::to_vec_named(event) { - Ok(b) => b, - Err(e) => { - tracing::warn!("Failed to serialize event: {}", e); - return; + let mut recipients = Vec::new(); + let mut found_dead_subscription = false; + { + let subscriptions = self.subscriptions.read(); + for subscription in subscriptions.iter() { + // Liveness is checked on every entry, not only the matching + // ones: a subscription on a topic nothing publishes to would + // otherwise never be visited, and so never pruned. + match ( + subscription.receives(topic), + subscription.listener_weak.upgrade(), + ) { + (true, Some(listener)) => recipients.push(listener), + (false, Some(_)) => {} + (_, None) => found_dead_subscription = true, + } } - }; - - if bytes.len() > MAX_EVENT_PAYLOAD_SIZE { - tracing::warn!( - "Event too large ({} bytes, max {}): {}", - bytes.len(), - MAX_EVENT_PAYLOAD_SIZE, - event.log_name() - ); - return; } - let timestamp_ns = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_nanos() as i64) - .unwrap_or(0); - - let payload = EventPayload::new(topic, timestamp_ns, &bytes); - - // Send to topic-specific service - self.send_payload(runtime_id, topic, &payload); + if found_dead_subscription { + self.subscriptions + .write() + .retain(|subscription| subscription.listener_weak.strong_count() > 0); + } - // Also send to /all aggregate service (if not already wildcard) - if topic != topics::ALL { - self.send_payload(runtime_id, topics::ALL, &payload); + let recipient_count = recipients.len(); + if !recipients.is_empty() + && self + .dispatch_sender + .send(PubSubDispatch::Deliver { + topic: topic.to_string(), + event: event.clone(), + recipients, + }) + .is_err() + { + tracing::error!( + "Dropping [{}] on topic '{}': the pubsub dispatcher is gone", + event.log_name(), + topic + ); + return; } tracing::debug!( - "Published [{}] to topic [{}] ({} bytes)", + "Queued [{}] for topic [{}] ({} listener(s))", event.log_name(), topic, - bytes.len() + recipient_count ); } - fn send_payload(&self, runtime_id: &str, topic: &str, payload: &EventPayload) { - let service_name = topic_to_service_name(runtime_id, topic); - let node = self.node.get().unwrap(); - - PUBLISHER_CACHE.with(|cache| { - let mut cache = cache.borrow_mut(); - - // Get or create a cached publisher for this service name. - // Publishers must stay alive so sent samples remain in shared memory. - if !cache.contains_key(&service_name) { - // Same `ServiceInCorruptedState` retry as in `subscribe_inner` - // — iceoryx2 dead-node cleanup can transiently flag a fresh - // service as corrupted when concurrent nodes scan at the same - // time. - let mut service = None; - for attempt in 0..10 { - match node.open_or_create_event_service(&service_name) { - Ok(s) => { - service = Some(s); - break; - } - Err(e) => { - tracing::warn!( - "Failed to open event service '{}' (attempt {}): {}", - service_name, - attempt + 1, - e - ); - std::thread::sleep(std::time::Duration::from_millis(20)); - } - } - } - let Some(service) = service else { - tracing::error!( - "Giving up after 10 attempts to open event service '{}'", - service_name - ); - return; - }; - - let publisher = match service.create_publisher() { - Ok(p) => p, - Err(e) => { - tracing::warn!("Failed to create publisher for '{}': {}", service_name, e); - return; - } - }; - - cache.insert(service_name.clone(), (service, publisher)); - } - - let (_, publisher) = cache.get(&service_name).unwrap(); - - match publisher.loan_uninit() { - Ok(sample) => { - let sample = sample.write_payload(*payload); - if let Err(e) = sample.send() { - tracing::warn!("Failed to send event to '{}': {:?}", service_name, e); - } - } - Err(e) => { - tracing::warn!("Failed to loan sample for '{}': {:?}", service_name, e); - } - } - }); + /// Block until everything queued before this call has been delivered. + /// + /// A barrier through the same FIFO, so it needs no timing assumption: it + /// cannot be acknowledged before the deliveries ahead of it have run. + pub fn flush(&self) { + let (delivered, wait_for_delivery) = sync_channel(1); + if self + .dispatch_sender + .send(PubSubDispatch::Barrier(delivered)) + .is_ok() + { + let _ = wait_for_delivery.recv(); + } } } -/// Blocking poll loop for an iceoryx2 event subscriber. -/// -/// Runs on a dedicated OS thread, polling the subscriber for new events. -/// Exits when the listener is dropped (weak ref upgrade fails). -fn subscriber_poll_loop( - subscriber: &iceoryx2::port::subscriber::Subscriber< - iceoryx2::service::ipc::Service, - EventPayload, - (), - >, - weak_listener: &Weak>, - topic: &str, -) { - loop { - // Drain all available events before sleeping - let mut received_any = false; - loop { - match subscriber.receive() { - Ok(Some(sample)) => { - received_any = true; - let payload: &EventPayload = &sample; - - // Deserialize event from MessagePack - let event: Event = match rmp_serde::from_slice(payload.data()) { - Ok(e) => e, - Err(e) => { - tracing::warn!( - "Failed to deserialize event on topic '{}': {}", - topic, - e - ); - continue; - } - }; +impl Drop for PubSub { + fn drop(&mut self) { + // Replacing the sender closes the channel, which ends the loop; joining + // keeps a test's dispatcher from outliving the bus it belonged to. + let (closed_sender, _) = sync_channel(1); + let _ = std::mem::replace(&mut self.dispatch_sender, closed_sender); + if let Some(dispatcher) = self.dispatcher.lock().take() { + let _ = dispatcher.join(); + } + } +} - // Deliver to listener (try_lock to avoid blocking, same as old rayon dispatch) - if let Some(listener) = weak_listener.upgrade() { - if let Some(mut guard) = listener.try_lock() { - let _ = guard.on_event(&event); - } else { - tracing::trace!( - "Listener busy on topic '{}', skipping (fire-and-forget)", - topic - ); - } - } else { - // Listener dropped, exit loop - tracing::debug!( - "Listener dropped for topic '{}', stopping poll thread", - topic +/// Deliver queued events in order until the bus is dropped. +fn run_dispatch_loop(dispatch_receiver: Receiver) { + while let Ok(dispatch) = dispatch_receiver.recv() { + match dispatch { + PubSubDispatch::Deliver { + topic, + event, + recipients, + } => { + for listener in &recipients { + if let Err(e) = listener.lock().on_event(&event) { + tracing::warn!( + "Listener on topic '{}' failed to handle [{}]: {}", + topic, + event.log_name(), + e ); - return; } } - Ok(None) => { - // No more data in buffer - break; - } - Err(e) => { - tracing::warn!("Event subscriber error on topic '{}': {:?}", topic, e); - return; - } + } + PubSubDispatch::Barrier(delivered) => { + let _ = delivered.send(()); } } - - // Check if listener is still alive before sleeping - if weak_listener.strong_count() == 0 { - tracing::debug!( - "Listener dropped for topic '{}', stopping poll thread", - topic - ); - return; - } - - // Sleep between polls. Events are infrequent (lifecycle, graph changes), - // so 5ms polling is more than sufficient. - std::thread::sleep(std::time::Duration::from_millis(5)); - - // Yield if we processed events for responsiveness - if received_any { - std::thread::yield_now(); - } - } -} - -/// Map a topic string to an iceoryx2 service name. -fn topic_to_service_name(runtime_id: &str, topic: &str) -> String { - if topic == topics::ALL { - format!("streamlib/{}/events/all", runtime_id) - } else { - // Replace colons with slashes for iceoryx2 service naming - let sanitized = topic.replace(':', "/"); - format!("streamlib/{}/events/{}", runtime_id, sanitized) } } diff --git a/runtime/streamlib-engine/src/core/pubsub/events.rs b/runtime/streamlib-engine/src/core/pubsub/events.rs index 817dcac34..5cbabf2e3 100644 --- a/runtime/streamlib-engine/src/core/pubsub/events.rs +++ b/runtime/streamlib-engine/src/core/pubsub/events.rs @@ -29,7 +29,13 @@ pub mod topics { } } -/// Trait for objects that can receive events +/// Trait for objects that can receive events. +/// +/// `on_event` runs on the bus's single dispatch thread, never on the thread that +/// published, so it holds no engine lock and cannot stall the engine. It must +/// still be a short handoff — set a flag, send on a channel, spawn a task — +/// because that one thread delivers every listener's events in order, so a slow +/// callback delays everyone's. pub trait EventListener: Send { fn on_event(&mut self, event: &Event) -> Result<()>; } @@ -320,8 +326,7 @@ pub enum RuntimeEvent { processor_type: ProcessorClassImportPath, }, /// Emitted when a processor type is unregistered from the factory - /// (`remove_module`). Additive variant — appended so existing msgpack - /// consumers keep decoding earlier variants unchanged. + /// (`remove_module`). RuntimeDidUnregisterProcessorType { processor_type: ProcessorClassImportPath, }, @@ -663,12 +668,10 @@ mod tests { #[test] fn test_event_serialization_roundtrip() { - // Verify events can be serialized/deserialized via MessagePack - // (critical for iceoryx2 transport). Locks **full** value - // equality, not just topic/log_name — a regression where a - // discriminator is lost on the wire but topic()/log_name() are - // computed from a fallback variant would slip past the older - // assertion. + // Locks **full** value equality, not just topic/log_name: a + // regression where a discriminator is lost on the wire but + // topic()/log_name() are computed from a fallback variant would slip + // past the weaker assertion. let events = vec![ Event::RuntimeGlobal(RuntimeEvent::RuntimeStarted), Event::RuntimeGlobal(RuntimeEvent::GraphDidChange), diff --git a/runtime/streamlib-engine/src/core/pubsub/integration_tests.rs b/runtime/streamlib-engine/src/core/pubsub/integration_tests.rs index 21087b4d1..ecf2de112 100644 --- a/runtime/streamlib-engine/src/core/pubsub/integration_tests.rs +++ b/runtime/streamlib-engine/src/core/pubsub/integration_tests.rs @@ -1,1082 +1,567 @@ // Copyright (c) 2025 Jonathan Fontanez // SPDX-License-Identifier: BUSL-1.1 -//! Integration tests for the pubsub module — exercises the full -//! iceoryx2 transport layer. +//! Tests for the in-process event bus. //! -//! Each test that requires iceoryx2 creates its own `PubSub::new()` + -//! `Iceoryx2Node::new()` instance with a unique runtime_id for -//! isolation (no global state). +//! Every test here is deterministic and free of sleeps, retries and timeouts, +//! and that is the point rather than a nicety. `subscribe` registers a listener +//! synchronously, so an event published after it returns is addressed to that +//! listener; `flush` then blocks until the FIFO ahead of it has drained. Neither +//! step involves a duration, so a regression fails rather than flakes. //! -//! `PubSub::subscribe()` takes ownership of the `Arc` but only stores -//! a `Weak` ref internally. Callers MUST keep a strong reference -//! alive for the subscriber thread to run. Always use -//! `bus.subscribe(topic, listener.clone())` and keep `listener` on -//! the stack. +//! The one bounded wait is the re-entrancy test, where a bound is the only way +//! to turn a deadlock into a named failure instead of a hung suite. //! -//! Synchronization strategy: -//! - Uses `std::sync::mpsc` channels for delivery notification (no -//! sleep-based waits) -//! - Uses retry-publish pattern to handle the race between subscriber -//! thread startup and the first publish (PubSub provides no -//! readiness signal) -//! -//! Lives in-source (rather than `tests/`) to access `super::bus::PubSub` -//! directly — the tests construct ad-hoc `PubSub` instances per case -//! for isolation, which is not exposed through the public surface. +//! Lives in-source (rather than `tests/`) to construct ad-hoc `PubSub` +//! instances, which the public surface does not expose. Each test owns its own +//! bus, so there is no shared state and no ordering between them. use super::bus::PubSub; use super::events::{ Event, EventListener, KeyCode, KeyState, Modifiers, MouseButton, MouseState, ProcessorEvent, RuntimeEvent, topics, }; -use crate::core::machine_global_unique_name::mint_machine_global_unique_name_suffix; -use crate::iceoryx2::{Iceoryx2Node, MAX_EVENT_PAYLOAD_SIZE}; use parking_lot::Mutex; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::mpsc; -use std::time::{Duration, Instant}; +use std::time::Duration; // --------------------------------------------------------------------------- // Test helpers // --------------------------------------------------------------------------- -struct CountingListener { - count: Arc, -} - -impl CountingListener { - fn new() -> Self { - Self { - count: Arc::new(AtomicUsize::new(0)), - } - } - - fn count(&self) -> usize { - self.count.load(Ordering::SeqCst) - } -} - -impl EventListener for CountingListener { - fn on_event(&mut self, _event: &Event) -> crate::core::error::Result<()> { - self.count.fetch_add(1, Ordering::SeqCst); - Ok(()) +/// Records every event it receives, in order. +#[derive(Default)] +struct RecordingListener { + received: Arc>>, +} + +impl RecordingListener { + fn with_shared_log() -> (Self, Arc>>) { + let received = Arc::new(Mutex::new(Vec::new())); + ( + Self { + received: Arc::clone(&received), + }, + received, + ) } } -/// Listener that sends received events through an mpsc channel. -struct ChannelListener { - sender: mpsc::Sender, -} - -impl EventListener for ChannelListener { +impl EventListener for RecordingListener { fn on_event(&mut self, event: &Event) -> crate::core::error::Result<()> { - let _ = self.sender.send(event.clone()); + self.received.lock().push(event.clone()); Ok(()) } } -/// Create an initialized PubSub instance with its own iceoryx2 node and unique runtime_id. -fn create_initialized_bus(test_name: &str) -> PubSub { - let runtime_id = format!("test-{}-{}", test_name, uuid::Uuid::new_v4()); - let node = Iceoryx2Node::new().expect("Failed to create iceoryx2 node"); - let bus = PubSub::new(); - bus.init(&runtime_id, node); - bus +/// Subscribe a recording listener, returning the strong `Arc` the caller must +/// keep alive and the log it appends to. +fn subscribe_recorder( + bus: &PubSub, + topic: &str, +) -> (Arc>, Arc>>) { + let (listener, received) = RecordingListener::with_shared_log(); + let listener: Arc> = Arc::new(Mutex::new(listener)); + bus.subscribe(topic, Arc::clone(&listener)); + (listener, received) } -/// Publish an event in a retry loop until the channel receives it (or timeout). -/// -/// Handles the race between subscriber thread startup and the first publish. -/// PubSub's subscribe() spawns a thread that creates the iceoryx2 subscriber -/// asynchronously — this function retries until the subscriber is ready. -fn publish_until_received( - bus: &PubSub, - event: &Event, - rx: &mpsc::Receiver, - timeout: Duration, -) -> Option { - let deadline = Instant::now() + timeout; - while Instant::now() < deadline { - bus.publish(&event.topic(), event); - match rx.recv_timeout(Duration::from_millis(50)) { - Ok(received) => return Some(received), - Err(mpsc::RecvTimeoutError::Timeout) => continue, - Err(mpsc::RecvTimeoutError::Disconnected) => return None, - } - } - None +fn keyboard_event() -> Event { + Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed) } // =========================================================================== -// A. Pre-init and routing tests (no iceoryx2 needed) +// A. Delivery — synchronous, with nothing to wait for // =========================================================================== +/// The contract the bus exists for, and the reason it has no live signal: an +/// event published after `subscribe` returns is already delivered when `publish` +/// returns. No retry, no budget, no poll. #[test] -fn test_pre_init_publish_is_noop() { - // Before init(), publish should silently drop events (no crash) +fn an_event_published_after_subscribe_is_delivered_before_publish_returns() { let bus = PubSub::new(); - let event = Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed); + let (_listener, received) = subscribe_recorder(&bus, topics::KEYBOARD); + + let event = keyboard_event(); bus.publish(&event.topic(), &event); - // No assertion needed — we just verify it doesn't panic + bus.flush(); + + assert_eq!(received.lock().len(), 1, "delivery is synchronous"); + assert_eq!(received.lock()[0].topic(), topics::KEYBOARD); } #[test] -fn test_pre_init_subscribe_buffers() { - // Before init(), subscribe should buffer (not crash) +fn every_subscriber_on_a_topic_receives_one_copy() { let bus = PubSub::new(); - let concrete = Arc::new(Mutex::new(CountingListener::new())); - let listener: Arc> = concrete.clone(); - bus.subscribe(topics::KEYBOARD, listener); - // Subscription is buffered, no events delivered yet - assert_eq!(concrete.lock().count(), 0); -} + let (_first, first_received) = subscribe_recorder(&bus, topics::KEYBOARD); + let (_second, second_received) = subscribe_recorder(&bus, topics::KEYBOARD); -#[test] -fn test_keyboard_event_topic_routing() { - let event = Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed); - assert_eq!(event.topic(), topics::KEYBOARD); -} + let event = keyboard_event(); + bus.publish(&event.topic(), &event); + bus.flush(); -#[test] -fn test_mouse_event_topic_routing() { - let event = Event::mouse(MouseButton::Left, (100.0, 200.0), MouseState::Pressed); - assert_eq!(event.topic(), topics::MOUSE); + assert_eq!(first_received.lock().len(), 1); + assert_eq!(second_received.lock().len(), 1); } #[test] -fn test_processor_event_topic_routing() { - let processor_id = "audio-mixer"; - let topic = topics::processor(processor_id); - let event = Event::processor(processor_id, ProcessorEvent::Started); - assert_eq!(event.topic(), topic); -} +fn an_event_reaches_only_its_own_topic() { + let bus = PubSub::new(); + let (_keyboard, keyboard_received) = subscribe_recorder(&bus, topics::KEYBOARD); + let (_mouse, mouse_received) = subscribe_recorder(&bus, topics::MOUSE); -#[test] -fn test_event_msgpack_serialization() { - // Verify events survive MessagePack round-trip (used by iceoryx2 transport) - let events = vec![ - Event::RuntimeGlobal(RuntimeEvent::RuntimeStarted), - Event::RuntimeGlobal(RuntimeEvent::GraphDidChange), - Event::processor("test-proc", ProcessorEvent::Started), - Event::custom("my-topic", serde_json::json!({"key": "value"})), - Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed), - Event::mouse(MouseButton::Right, (50.0, 75.0), MouseState::Released), - ]; + let event = Event::mouse(MouseButton::Left, (10.0, 20.0), MouseState::Pressed); + bus.publish(&event.topic(), &event); + bus.flush(); - for event in events { - let bytes = rmp_serde::to_vec_named(&event).unwrap(); - let deserialized: Event = rmp_serde::from_slice(&bytes).unwrap(); - assert_eq!(event.topic(), deserialized.topic()); - assert_eq!(event.log_name(), deserialized.log_name()); - } + assert_eq!(mouse_received.lock().len(), 1); + assert!( + keyboard_received.lock().is_empty(), + "a keyboard subscriber must not see a mouse event" + ); } -// =========================================================================== -// B. Diagnostic: verify iceoryx2 transport works at the low level -// =========================================================================== - +/// A wildcard subscriber sees every topic, and sees each event exactly once — +/// not twice for having matched both the specific topic and the wildcard. #[test] -fn test_iceoryx2_direct_delivery() { - // Bypass PubSub layer entirely — verify iceoryx2 pub/sub works in-process - let node = Iceoryx2Node::new().expect("Failed to create iceoryx2 node"); - - let service_name = format!( - "streamlib/diag-{}/events/test", - mint_machine_global_unique_name_suffix() - ); +fn a_wildcard_subscriber_receives_every_topic_exactly_once() { + let bus = PubSub::new(); + let (_listener, received) = subscribe_recorder(&bus, topics::ALL); - // Create subscriber FIRST (must exist before publisher sends) - let sub_service = node - .open_or_create_event_service(&service_name) - .expect("subscriber service"); - let subscriber = sub_service.create_subscriber().expect("subscriber"); - - // Create publisher - let pub_service = node - .open_or_create_event_service(&service_name) - .expect("publisher service"); - let publisher = pub_service.create_publisher().expect("publisher"); - - // Publish - let event = Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed); - let bytes = rmp_serde::to_vec_named(&event).unwrap(); - let payload = crate::iceoryx2::EventPayload::new("test", 12345, &bytes); - - let sample = publisher.loan_uninit().expect("loan"); - let sample = sample.write_payload(payload); - sample.send().expect("send"); - - // Receive - match subscriber.receive() { - Ok(Some(sample)) => { - let p: &crate::iceoryx2::EventPayload = &*sample; - let received: Event = rmp_serde::from_slice(p.data()).unwrap(); - assert_eq!(received.topic(), event.topic()); - } - Ok(None) => { - panic!("iceoryx2 subscriber received None — message not delivered"); - } - Err(e) => { - panic!("iceoryx2 subscriber error: {:?}", e); - } + let events = [ + keyboard_event(), + Event::mouse(MouseButton::Right, (5.0, 10.0), MouseState::Released), + Event::processor("test-proc", ProcessorEvent::Started), + Event::RuntimeGlobal(RuntimeEvent::GraphDidChange), + ]; + for event in &events { + bus.publish(&event.topic(), event); + bus.flush(); } + + assert_eq!(received.lock().len(), events.len()); } #[test] -fn test_iceoryx2_cross_thread_delivery() { - // Verify iceoryx2 delivery works across threads (mimics PubSub pattern) - let node = Iceoryx2Node::new().expect("Failed to create iceoryx2 node"); - let service_name = format!( - "streamlib/diag-xthread-{}/events/test", - uuid::Uuid::new_v4() +fn a_delivered_event_is_identical_to_the_one_published() { + let bus = PubSub::new(); + let (_listener, received) = subscribe_recorder(&bus, topics::KEYBOARD); + + let event = Event::keyboard( + KeyCode::Z, + Modifiers { + shift: true, + ctrl: false, + alt: true, + meta: false, + }, + KeyState::Released, ); + bus.publish(&event.topic(), &event); + bus.flush(); - let (tx, rx) = mpsc::channel::<()>(); - let node_clone = node.clone(); - let sn = service_name.clone(); - - // Subscriber on a separate thread - std::thread::spawn(move || { - let service = node_clone - .open_or_create_event_service(&sn) - .expect("sub service"); - let subscriber = service.create_subscriber().expect("subscriber"); - - let deadline = Instant::now() + Duration::from_secs(2); - while Instant::now() < deadline { - match subscriber.receive() { - Ok(Some(_)) => { - let _ = tx.send(()); - return; - } - Ok(None) => { - std::thread::yield_now(); - } - Err(_) => return, - } - } - }); - - // Brief yield to let thread start, then publish in a retry loop - std::thread::yield_now(); - let pub_service = node - .open_or_create_event_service(&service_name) - .expect("pub service"); - let publisher = pub_service.create_publisher().expect("publisher"); - - let deadline = Instant::now() + Duration::from_secs(2); - while Instant::now() < deadline { - let payload = crate::iceoryx2::EventPayload::new("test", 12345, b"hello"); - let sample = publisher.loan_uninit().expect("loan"); - let sample = sample.write_payload(payload); - sample.send().expect("send"); - - match rx.recv_timeout(Duration::from_millis(50)) { - Ok(()) => return, // success - Err(mpsc::RecvTimeoutError::Timeout) => continue, - Err(mpsc::RecvTimeoutError::Disconnected) => { - panic!("Subscriber thread exited without receiving"); - } - } - } - panic!("Cross-thread iceoryx2 delivery timed out"); + assert_eq!(received.lock()[0], event); } #[test] -fn test_iceoryx2_pubsub_pattern_mimic() { - // Exactly mimic what PubSub does: subscriber thread + fresh publisher per call - let node = Iceoryx2Node::new().expect("Failed to create iceoryx2 node"); - let runtime_id = format!("test-mimic-{}", uuid::Uuid::new_v4()); - let topic = "input/keyboard"; - let service_name = format!("streamlib/{}/events/{}", runtime_id, topic); - - let (tx, rx) = mpsc::channel::<()>(); - let node_clone = node.clone(); - let sn_clone = service_name.clone(); - - // Subscriber thread (mimics subscribe_inner) - std::thread::spawn(move || { - let service = node_clone - .open_or_create_event_service(&sn_clone) - .expect("sub service"); - let subscriber = service.create_subscriber().expect("subscriber"); - - let deadline = Instant::now() + Duration::from_secs(2); - while Instant::now() < deadline { - match subscriber.receive() { - Ok(Some(_)) => { - let _ = tx.send(()); - return; - } - Ok(None) => { - std::thread::yield_now(); - } - Err(_) => return, - } - } - }); +fn a_custom_event_carries_its_payload_intact() { + let bus = PubSub::new(); + let (_listener, received) = subscribe_recorder(&bus, "my-custom-topic"); - // Retry publish until subscriber receives (handles startup race) - let deadline = Instant::now() + Duration::from_secs(2); - while Instant::now() < deadline { - // Fresh service + publisher per call (mimics PubSub::send_payload) - let pub_service = node - .open_or_create_event_service(&service_name) - .expect("pub service"); - let publisher = pub_service.create_publisher().expect("publisher"); - - let event = Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed); - let bytes = rmp_serde::to_vec_named(&event).unwrap(); - let payload = crate::iceoryx2::EventPayload::new(topic, 12345, &bytes); - - let sample = publisher.loan_uninit().expect("loan"); - let sample = sample.write_payload(payload); - sample.send().expect("send"); - - match rx.recv_timeout(Duration::from_millis(50)) { - Ok(()) => return, - Err(mpsc::RecvTimeoutError::Timeout) => continue, - Err(mpsc::RecvTimeoutError::Disconnected) => { - panic!("Subscriber thread exited without receiving"); - } - } - } - panic!("PubSub pattern mimic delivery timed out"); + let event = Event::custom("my-custom-topic", serde_json::json!({"key": "value"})); + bus.publish(&event.topic(), &event); + bus.flush(); + + assert_eq!(received.lock()[0], event); } // =========================================================================== -// C. Diagnostic: verify PubSub's publish actually sends data to iceoryx2 +// B. Lifecycle — no initialization step, no buffering // =========================================================================== +/// There is no `init`: a bus delivers from its first instruction, so nothing is +/// ever buffered waiting for a backend that has not come up. #[test] -fn test_pubsub_publish_sends_to_iceoryx2() { - // Verify that PubSub::publish() actually sends data through iceoryx2 - // by creating a manual subscriber on the same service name. - // - // This bypasses PubSub's subscriber thread to isolate whether the bug - // is in publish (send side) or subscribe (receive side). - - let runtime_id = format!("test-pub-sends-{}", uuid::Uuid::new_v4()); - let node = Iceoryx2Node::new().expect("Failed to create iceoryx2 node"); - let node_probe = node.clone(); +fn a_fresh_bus_delivers_with_no_initialization_step() { let bus = PubSub::new(); - bus.init(&runtime_id, node); - - // Compute the service name PubSub will use for topics::KEYBOARD - let sanitized_topic = topics::KEYBOARD.replace(':', "/"); - let service_name = format!("streamlib/{}/events/{}", runtime_id, sanitized_topic); - - // Create a manual iceoryx2 subscriber on that service BEFORE publishing - let probe_service = node_probe - .open_or_create_event_service(&service_name) - .expect("probe service"); - let _probe_subscriber = probe_service.create_subscriber().expect("probe subscriber"); - - // KEY INSIGHT: send() reports delivering to N subscribers, but receive() - // returns None. Test if the issue is PortFactory creation order. - // - // Hypothesis: when send_payload creates a PortFactory via open_or_create, - // and the service already exists (created by probe subscribers), the new - // PortFactory's publisher sends to different subscriber slots. - // - // Test: force PubSub to create the service FIRST (via a warm-up publish), - // THEN create probe subscribers, THEN publish the real event. - - let event = Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed); - - // === Test A: Subscribers created BEFORE PubSub publishes (current pattern) === - let pre_probe_service = node_probe - .open_or_create_event_service(&service_name) - .expect("pre-probe service"); - let pre_probe_sub = pre_probe_service - .create_subscriber() - .expect("pre-probe subscriber"); + let (_listener, received) = subscribe_recorder(&bus, topics::RUNTIME_GLOBAL); + let event = Event::RuntimeGlobal(RuntimeEvent::RuntimeStarted); bus.publish(&event.topic(), &event); + bus.flush(); - let pre_result = match pre_probe_sub.receive() { - Ok(Some(_)) => "RECEIVED", - Ok(None) => "NONE", - Err(e) => panic!("Pre-probe error: {:?}", e), - }; - eprintln!("[diag] Test A (sub before pub): {}", pre_result); - - // === Test B: Warm-up publish FIRST, then create subscriber, then publish again === - // Use a different service name to avoid interference - let service_name_b = format!("streamlib/{}/events/input/mouse", runtime_id); - let mouse_event = Event::mouse(MouseButton::Left, (0.0, 0.0), MouseState::Pressed); - - // Warm-up: force PubSub to create the service - bus.publish(&mouse_event.topic(), &mouse_event); - eprintln!("[diag] Test B: warm-up publish done"); - - // Now create subscriber (service already exists from warm-up) - let post_probe_service = node_probe - .open_or_create_event_service(&service_name_b) - .expect("post-probe service"); - let post_probe_sub = post_probe_service - .create_subscriber() - .expect("post-probe subscriber"); - - // Publish again - bus.publish(&mouse_event.topic(), &mouse_event); - - let post_result = match post_probe_sub.receive() { - Ok(Some(_)) => "RECEIVED", - Ok(None) => "NONE", - Err(e) => panic!("Post-probe error: {:?}", e), - }; - eprintln!("[diag] Test B (sub after warm-up pub): {}", post_result); - - // === Test C: Keep publisher alive across receive === - // Maybe the issue is that send_payload drops publisher before we receive - let service_name_c = format!("streamlib/{}/events/input/window", runtime_id); - let c_probe_service = node_probe - .open_or_create_event_service(&service_name_c) - .expect("c-probe service"); - let c_probe_sub = c_probe_service - .create_subscriber() - .expect("c-probe subscriber"); - - // Mimic send_payload but keep publisher alive - let c_pub_service = node_probe - .open_or_create_event_service(&service_name_c) - .expect("c-pub service"); - let c_publisher = c_pub_service.create_publisher().expect("c-publisher"); - let bytes = rmp_serde::to_vec_named(&event).unwrap(); - let c_payload = crate::iceoryx2::EventPayload::new("input:window", 12345, &bytes); - let c_sample = c_publisher.loan_uninit().expect("loan"); - let c_sample = c_sample.write_payload(c_payload); - c_sample.send().expect("send"); - - let c_result = match c_probe_sub.receive() { - Ok(Some(_)) => "RECEIVED", - Ok(None) => "NONE", - Err(e) => panic!("C-probe error: {:?}", e), - }; - eprintln!("[diag] Test C (keep publisher alive): {}", c_result); - - // Report - assert!( - pre_result == "RECEIVED" || post_result == "RECEIVED", - "PubSub::publish() should send data to iceoryx2. \ - TestA(sub-before-pub)={}, TestB(sub-after-warmup)={}, TestC(keep-pub-alive)={}", - pre_result, - post_result, - c_result - ); + assert_eq!(received.lock().len(), 1); } -/// Minimal reproduction of PubSub's send_payload using OnceLock, -/// to determine if OnceLock storage causes the iceoryx2 delivery failure. #[test] -fn test_oncelock_node_delivery() { - let node = Iceoryx2Node::new().expect("Failed to create iceoryx2 node"); - let runtime_id = format!("test-oncelock-{}", uuid::Uuid::new_v4()); - let service_name = format!("streamlib/{}/events/input/keyboard", runtime_id); - - // Store node in OnceLock (mimics PubSub's storage) - let node_in_lock: std::sync::OnceLock = std::sync::OnceLock::new(); - let _ = node_in_lock.set(node.clone()); - - // Create subscriber from the direct node clone - let sub_service = node - .open_or_create_event_service(&service_name) - .expect("sub service"); - let subscriber = sub_service.create_subscriber().expect("subscriber"); - - // Publish from OnceLock-stored node (mimics PubSub::send_payload) - let node_ref = node_in_lock.get().unwrap(); - let pub_service = node_ref - .open_or_create_event_service(&service_name) - .expect("pub service"); - let publisher = pub_service.create_publisher().expect("publisher"); - - let event = Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed); - let bytes = rmp_serde::to_vec_named(&event).unwrap(); - let payload = crate::iceoryx2::EventPayload::new("input:keyboard", 12345, &bytes); - - // Use *&payload to mimic PubSub's `write_payload(*payload)` (copy from reference) - let sample = publisher.loan_uninit().expect("loan"); - let sample = sample.write_payload(payload); - sample.send().expect("send"); - - match subscriber.receive() { - Ok(Some(_)) => { - eprintln!("[oncelock test] RECEIVED — OnceLock pattern works"); - } - Ok(None) => { - panic!("OnceLock-stored node publish failed — iceoryx2 + OnceLock interaction bug"); - } - Err(e) => { - panic!("OnceLock subscriber error: {:?}", e); - } - } +fn publishing_with_no_subscribers_is_a_no_op() { + let bus = PubSub::new(); + let event = keyboard_event(); + bus.publish(&event.topic(), &event); + bus.flush(); } -// =========================================================================== -// D. End-to-end message delivery through PubSub -// =========================================================================== - #[test] -fn test_publish_delivers_to_subscriber() { - let bus = create_initialized_bus("publish_delivers"); - - let (tx, rx) = mpsc::channel(); - let listener = ChannelListener { sender: tx }; - let listener: Arc> = Arc::new(Mutex::new(listener)); - bus.subscribe(topics::KEYBOARD, listener.clone()); - - let event = Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed); - let received = publish_until_received(&bus, &event, &rx, Duration::from_secs(5)); +fn dropping_the_listener_unsubscribes_it() { + let bus = PubSub::new(); + let (listener, received) = subscribe_recorder(&bus, topics::KEYBOARD); - assert!( - received.is_some(), - "Subscriber should have received at least one event" - ); + let event = keyboard_event(); + bus.publish(&event.topic(), &event); + bus.flush(); + assert_eq!(received.lock().len(), 1); drop(listener); -} - -#[test] -fn test_publish_delivers_to_multiple_subscribers_on_same_topic() { - let bus = create_initialized_bus("multi_sub_same_topic"); - - let (tx_a, rx_a) = mpsc::channel(); - let (tx_b, rx_b) = mpsc::channel(); - let listener_a: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx_a })); - let listener_b: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx_b })); - - bus.subscribe(topics::KEYBOARD, listener_a.clone()); - bus.subscribe(topics::KEYBOARD, listener_b.clone()); - - let event = Event::keyboard(KeyCode::B, Modifiers::default(), KeyState::Pressed); - - // Retry publish until BOTH subscribers receive - let deadline = Instant::now() + Duration::from_secs(5); - let mut a_received = false; - let mut b_received = false; - while Instant::now() < deadline && !(a_received && b_received) { - bus.publish(&event.topic(), &event); - // Drain both channels - while rx_a.try_recv().is_ok() { - a_received = true; - } - while rx_b.try_recv().is_ok() { - b_received = true; - } - if !(a_received && b_received) { - std::thread::yield_now(); - } - } + bus.publish(&event.topic(), &event); + bus.flush(); - assert!( - a_received, - "First subscriber should have received the event" - ); - assert!( - b_received, - "Second subscriber should have received the event" + assert_eq!( + received.lock().len(), + 1, + "a dropped listener receives nothing further" ); - - drop(listener_a); - drop(listener_b); } +/// Dropping a listener leaves a dead registration behind; the next publish that +/// finds it removes it, so a long-lived bus does not accumulate them. +/// +/// The count is asserted, not inferred from delivery: a dead entry is skipped +/// whether or not it is ever pruned, so delivery alone proves nothing about the +/// registry shrinking. #[test] -fn test_publish_does_not_cross_topics() { - let bus = create_initialized_bus("no_cross_topics"); - - let (tx_keyboard, rx_keyboard) = mpsc::channel(); - let (tx_mouse, rx_mouse) = mpsc::channel(); - let kb_listener: Arc> = Arc::new(Mutex::new(ChannelListener { - sender: tx_keyboard, - })); - let mouse_listener: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx_mouse })); - - bus.subscribe(topics::KEYBOARD, kb_listener.clone()); - bus.subscribe(topics::MOUSE, mouse_listener.clone()); +fn a_dropped_listeners_registration_is_pruned_by_the_next_publish() { + let bus = PubSub::new(); + let (listener, _received) = subscribe_recorder(&bus, topics::KEYBOARD); + let (_survivor, survivor_received) = subscribe_recorder(&bus, topics::KEYBOARD); + assert_eq!(bus.registration_count(), 2); - // Publish a MOUSE event — only the mouse subscriber should receive it - let mouse_event = Event::mouse(MouseButton::Left, (10.0, 20.0), MouseState::Pressed); + drop(listener); + let event = keyboard_event(); + bus.publish(&event.topic(), &event); + bus.flush(); - // Use retry loop to ensure the mouse subscriber is ready - let received_mouse = - publish_until_received(&bus, &mouse_event, &rx_mouse, Duration::from_secs(5)); - assert!( - received_mouse.is_some(), - "Mouse subscriber should receive mouse events" + assert_eq!( + bus.registration_count(), + 1, + "the dead registration is gone, not merely skipped" ); - - // Verify the keyboard subscriber received nothing - assert!( - rx_keyboard.try_recv().is_err(), - "Keyboard subscriber should NOT receive mouse events" + assert_eq!( + survivor_received.lock().len(), + 1, + "pruning a dead entry must not disturb a live one" ); - - drop(kb_listener); - drop(mouse_listener); } +/// A subscriber on a topic nothing publishes to is still pruned — liveness is +/// checked on every registration, not only the ones a publish routes to. #[test] -fn test_wildcard_subscriber_receives_all_topics() { - let bus = create_initialized_bus("wildcard_all"); - - let (tx, rx) = mpsc::channel(); - let listener: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx })); - bus.subscribe(topics::ALL, listener.clone()); - - let keyboard_event = Event::keyboard(KeyCode::C, Modifiers::default(), KeyState::Pressed); - let mouse_event = Event::mouse(MouseButton::Right, (5.0, 10.0), MouseState::Released); - let processor_event = Event::processor("test-proc", ProcessorEvent::Started); - - // Ensure wildcard subscriber is ready by retrying first event - let first = publish_until_received(&bus, &keyboard_event, &rx, Duration::from_secs(5)); - assert!( - first.is_some(), - "Wildcard subscriber should receive keyboard event" - ); +fn a_dropped_listener_is_pruned_even_on_a_topic_nothing_publishes_to() { + let bus = PubSub::new(); + let (quiet_listener, _quiet) = subscribe_recorder(&bus, &topics::processor("never-published")); + let (_active, _active_received) = subscribe_recorder(&bus, topics::KEYBOARD); + assert_eq!(bus.registration_count(), 2); - // Now publish remaining events (subscriber is ready) - bus.publish(&mouse_event.topic(), &mouse_event); - bus.publish(&processor_event.topic(), &processor_event); - - // Wait for remaining events - let deadline = Instant::now() + Duration::from_secs(2); - let mut received_count = 1; // already got first - while Instant::now() < deadline && received_count < 3 { - match rx.recv_timeout(Duration::from_millis(50)) { - Ok(_) => received_count += 1, - Err(mpsc::RecvTimeoutError::Timeout) => continue, - Err(mpsc::RecvTimeoutError::Disconnected) => break, - } - } + drop(quiet_listener); + let event = keyboard_event(); + bus.publish(&event.topic(), &event); + bus.flush(); - // The wildcard subscriber receives the event from BOTH the specific topic service - // AND the /all service, so we expect at least 3 events (may receive duplicates) - assert!( - received_count >= 3, - "Wildcard subscriber should receive at least 3 events, got {}", - received_count + assert_eq!( + bus.registration_count(), + 1, + "a dead registration on an unpublished topic must not accumulate" ); - - drop(listener); } #[test] -fn test_subscriber_receives_correct_event_data() { - let bus = create_initialized_bus("correct_data"); - - let (tx, rx) = mpsc::channel(); - let listener: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx })); - bus.subscribe(topics::KEYBOARD, listener.clone()); - - let modifiers = Modifiers { - shift: true, - ctrl: false, - alt: true, - meta: false, - }; - let event = Event::keyboard(KeyCode::Z, modifiers, KeyState::Released); - let received = publish_until_received(&bus, &event, &rx, Duration::from_secs(5)); - - let received = received.expect("Should have received the event"); - - // Verify the round-tripped event has the same topic and log_name - assert_eq!(received.topic(), topics::KEYBOARD); - assert_eq!(received.log_name(), event.log_name()); - - // Verify via MessagePack that the serialized forms match - let original_bytes = rmp_serde::to_vec_named(&event).unwrap(); - let received_bytes = rmp_serde::to_vec_named(&received).unwrap(); - assert_eq!(original_bytes, received_bytes, "Payload fidelity mismatch"); +fn two_buses_are_isolated() { + let first_bus = PubSub::new(); + let second_bus = PubSub::new(); + let (_first, first_received) = subscribe_recorder(&first_bus, topics::KEYBOARD); + let (_second, second_received) = subscribe_recorder(&second_bus, topics::KEYBOARD); - drop(listener); + let event = keyboard_event(); + first_bus.publish(&event.topic(), &event); + + // Both are flushed, so the second bus's silence is isolation rather than an + // event still sitting in its queue. + first_bus.flush(); + second_bus.flush(); + + assert_eq!(first_received.lock().len(), 1); + assert!(second_received.lock().is_empty()); } // =========================================================================== -// E. Subscription lifecycle & ordering +// C. Ordering // =========================================================================== +/// Every listener observes one order, and it is publish order. +/// +/// The property a topic owes its subscribers: two observers of the same stream +/// must never disagree about what happened first. Dispatching inline on whoever +/// published cannot promise this — two publisher threads would each walk the +/// listener list independently, so one listener could see A then B while another +/// saw B then A. #[test] -fn test_subscribe_before_init_receives_events_after_init() { - let runtime_id = format!("test-sub-before-init-{}", uuid::Uuid::new_v4()); - let node = Iceoryx2Node::new().expect("Failed to create iceoryx2 node"); - let bus = PubSub::new(); - - // Subscribe BEFORE init - let (tx, rx) = mpsc::channel(); - let listener: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx })); - bus.subscribe(topics::KEYBOARD, listener.clone()); +fn every_listener_sees_events_in_publish_order() { + const EVENTS: usize = 200; - // Now init — pending subscription should be replayed - bus.init(&runtime_id, node); - - let event = Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed); - let received = publish_until_received(&bus, &event, &rx, Duration::from_secs(5)); + let bus = PubSub::new(); + let (_first, first_received) = subscribe_recorder(&bus, topics::ALL); + let (_second, second_received) = subscribe_recorder(&bus, topics::ALL); - assert!( - received.is_some(), - "Subscription registered before init should receive events after init" - ); + let published: Vec = (0..EVENTS) + .map(|sequence| Event::custom("ordering", serde_json::json!({ "sequence": sequence }))) + .collect(); + for event in &published { + bus.publish(&event.topic(), event); + } + bus.flush(); - drop(listener); + assert_eq!(*first_received.lock(), published, "publish order, exactly"); + assert_eq!(*second_received.lock(), published); } +/// A listener never runs on the thread that published. +/// +/// This is what the FIFO buys beyond ordering, and it is the half a test can +/// pin exactly. The engine publishes from inside its own graph write lock, so +/// inline dispatch would put listener code underneath a lock it knows nothing +/// about; delivery on a thread of the bus's own makes that structurally +/// impossible rather than a rule listeners have to remember. #[test] -fn test_multiple_subscribes_before_init_all_replayed() { - let runtime_id = format!("test-multi-sub-before-init-{}", uuid::Uuid::new_v4()); - let node = Iceoryx2Node::new().expect("Failed to create iceoryx2 node"); - let bus = PubSub::new(); - - // Subscribe 3 listeners to different topics BEFORE init - let (tx_kb, rx_kb) = mpsc::channel(); - let (tx_mouse, rx_mouse) = mpsc::channel(); - let (tx_rt, rx_rt) = mpsc::channel(); - let kb_handle: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx_kb })); - let mouse_handle: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx_mouse })); - let rt_handle: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx_rt })); - - bus.subscribe(topics::KEYBOARD, kb_handle.clone()); - bus.subscribe(topics::MOUSE, mouse_handle.clone()); - bus.subscribe(topics::RUNTIME_GLOBAL, rt_handle.clone()); - - // Init replays all 3 pending subscriptions - bus.init(&runtime_id, node); - - let kb_event = Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed); - let mouse_event = Event::mouse(MouseButton::Left, (0.0, 0.0), MouseState::Pressed); - let rt_event = Event::RuntimeGlobal(RuntimeEvent::RuntimeStarted); - - // Retry until all 3 subscribers receive - let deadline = Instant::now() + Duration::from_secs(5); - let mut kb_ok = false; - let mut mouse_ok = false; - let mut rt_ok = false; - while Instant::now() < deadline && !(kb_ok && mouse_ok && rt_ok) { - if !kb_ok { - bus.publish(topics::KEYBOARD, &kb_event); - } - if !mouse_ok { - bus.publish(topics::MOUSE, &mouse_event); - } - if !rt_ok { - bus.publish(topics::RUNTIME_GLOBAL, &rt_event); - } - if rx_kb.try_recv().is_ok() { - kb_ok = true; - } - if rx_mouse.try_recv().is_ok() { - mouse_ok = true; - } - if rx_rt.try_recv().is_ok() { - rt_ok = true; +fn a_listener_never_runs_on_the_publishing_thread() { + struct ThreadRecordingListener { + delivered_on: Arc>>, + } + impl EventListener for ThreadRecordingListener { + fn on_event(&mut self, _event: &Event) -> crate::core::error::Result<()> { + self.delivered_on.lock().push(std::thread::current().id()); + Ok(()) } - std::thread::yield_now(); } - assert!(kb_ok, "Keyboard listener should receive events"); - assert!(mouse_ok, "Mouse listener should receive events"); - assert!(rt_ok, "Runtime listener should receive events"); + let bus = PubSub::new(); + let delivered_on = Arc::new(Mutex::new(Vec::new())); + let listener: Arc> = Arc::new(Mutex::new(ThreadRecordingListener { + delivered_on: Arc::clone(&delivered_on), + })); + bus.subscribe(topics::ALL, Arc::clone(&listener)); - drop(kb_handle); - drop(mouse_handle); - drop(rt_handle); + let event = keyboard_event(); + bus.publish(&event.topic(), &event); + bus.flush(); + + let publishing_thread = std::thread::current().id(); + let delivered_on = delivered_on.lock(); + assert_eq!(delivered_on.len(), 1); + assert_ne!( + delivered_on[0], publishing_thread, + "delivery must not run on the publisher's thread" + ); } +/// Concurrent publishers still produce ONE order that both listeners agree on. +/// +/// The interleaving is whatever the threads race to, and that is fine — what is +/// not fine is two listeners disagreeing about it, which is what a per-publisher +/// dispatch would allow. #[test] -fn test_listener_drop_stops_subscriber_thread() { - let bus = create_initialized_bus("listener_drop"); - - let (tx, rx) = mpsc::channel(); - let listener: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx })); - - bus.subscribe(topics::KEYBOARD, listener.clone()); +fn concurrent_publishers_produce_one_order_all_listeners_agree_on() { + const PUBLISHER_THREADS: usize = 4; + const PUBLISHES_PER_THREAD: usize = 50; + + let bus = Arc::new(PubSub::new()); + let (_first, first_received) = subscribe_recorder(&bus, topics::ALL); + let (_second, second_received) = subscribe_recorder(&bus, topics::ALL); + + let publishers: Vec<_> = (0..PUBLISHER_THREADS) + .map(|publisher| { + let bus = Arc::clone(&bus); + std::thread::spawn(move || { + for sequence in 0..PUBLISHES_PER_THREAD { + let event = Event::custom( + "ordering", + serde_json::json!({ "publisher": publisher, "sequence": sequence }), + ); + bus.publish(&event.topic(), &event); + } + }) + }) + .collect(); + for publisher in publishers { + publisher.join().expect("publisher thread panicked"); + } + bus.flush(); - // Verify subscriber is working first - let event = Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed); - let received = publish_until_received(&bus, &event, &rx, Duration::from_secs(5)); - assert!( - received.is_some(), - "Should receive events before dropping listener" + let first = first_received.lock().clone(); + let second = second_received.lock().clone(); + assert_eq!( + first.len(), + PUBLISHER_THREADS * PUBLISHES_PER_THREAD, + "every publish is delivered" + ); + assert_eq!( + first, second, + "two listeners must never disagree about the order events arrived in" ); +} - // Drop the strong reference — subscriber thread should detect and exit - drop(listener); +/// Each publisher's own events keep their relative order inside that agreed one. +#[test] +fn a_publishers_own_events_stay_in_order_relative_to_each_other() { + const PUBLISHES: usize = 100; - // The channel sender is inside the dropped listener, so rx should disconnect - // Publish after the listener is dropped — should not panic or hang - bus.publish(&event.topic(), &event); + let bus = PubSub::new(); + let (_listener, received) = subscribe_recorder(&bus, topics::ALL); - // Channel should be disconnected since the sender was dropped with the listener - match rx.recv_timeout(Duration::from_millis(200)) { - Err(mpsc::RecvTimeoutError::Disconnected) => { /* expected */ } - Err(mpsc::RecvTimeoutError::Timeout) => { /* also acceptable — no events */ } - Ok(_) => { - // This might happen if an event was in-flight before the drop - // but no further events should arrive - } + for sequence in 0..PUBLISHES { + let event = Event::custom("ordering", serde_json::json!({ "sequence": sequence })); + bus.publish(&event.topic(), &event); } + bus.flush(); + + let sequences: Vec = received + .lock() + .iter() + .filter_map(|event| match event { + Event::Custom { data, .. } => data.get("sequence")?.as_u64(), + _ => None, + }) + .collect(); + assert_eq!(sequences, (0..PUBLISHES as u64).collect::>()); } // =========================================================================== -// F. Event type coverage +// C. Concurrency and re-entrancy // =========================================================================== +/// Every publish from every thread is delivered — an exact count, not "at least +/// one". The old transport could only promise the latter. #[test] -fn test_runtime_event_delivery() { - let bus = create_initialized_bus("runtime_events"); - - let (tx, rx) = mpsc::channel(); - let listener: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx })); - bus.subscribe(topics::RUNTIME_GLOBAL, listener.clone()); - - // Ensure subscriber is ready with first event - let first = Event::RuntimeGlobal(RuntimeEvent::RuntimeStarted); - let received = publish_until_received(&bus, &first, &rx, Duration::from_secs(5)); - assert!(received.is_some(), "Should receive RuntimeStarted event"); - - // Send remaining runtime events - bus.publish( - topics::RUNTIME_GLOBAL, - &Event::RuntimeGlobal(RuntimeEvent::GraphDidChange), - ); - bus.publish( - topics::RUNTIME_GLOBAL, - &Event::RuntimeGlobal(RuntimeEvent::RuntimeShutdown), - ); +fn concurrent_publishes_are_all_delivered() { + const PUBLISHER_THREADS: usize = 4; + const PUBLISHES_PER_THREAD: usize = 25; + + let bus = Arc::new(PubSub::new()); + let delivered = Arc::new(AtomicUsize::new(0)); - // Wait for remaining events - let mut count = 1; - let deadline = Instant::now() + Duration::from_secs(2); - while Instant::now() < deadline && count < 3 { - match rx.recv_timeout(Duration::from_millis(50)) { - Ok(_) => count += 1, - Err(mpsc::RecvTimeoutError::Timeout) => continue, - Err(mpsc::RecvTimeoutError::Disconnected) => break, + struct CountingListener { + delivered: Arc, + } + impl EventListener for CountingListener { + fn on_event(&mut self, _event: &Event) -> crate::core::error::Result<()> { + self.delivered.fetch_add(1, Ordering::SeqCst); + Ok(()) } } - assert!( - count >= 3, - "Should receive all 3 runtime events, got {}", - count - ); - - drop(listener); -} - -#[test] -fn test_processor_event_delivery() { - let bus = create_initialized_bus("processor_events"); - - let processor_id = "audio-mixer"; - let topic = topics::processor(processor_id); - - let (tx, rx) = mpsc::channel(); - let listener: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx })); - bus.subscribe(&topic, listener.clone()); - - let event = Event::processor(processor_id, ProcessorEvent::Started); - let received = publish_until_received(&bus, &event, &rx, Duration::from_secs(5)); + let listener: Arc> = Arc::new(Mutex::new(CountingListener { + delivered: Arc::clone(&delivered), + })); + bus.subscribe(topics::ALL, Arc::clone(&listener)); + + let publishers: Vec<_> = (0..PUBLISHER_THREADS) + .map(|_| { + let bus = Arc::clone(&bus); + std::thread::spawn(move || { + for _ in 0..PUBLISHES_PER_THREAD { + let event = keyboard_event(); + bus.publish(&event.topic(), &event); + bus.flush(); + } + }) + }) + .collect(); + for publisher in publishers { + publisher.join().expect("publisher thread panicked"); + } - assert!( - received.is_some(), - "Processor subscriber should receive the Started event" + assert_eq!( + delivered.load(Ordering::SeqCst), + PUBLISHER_THREADS * PUBLISHES_PER_THREAD, + "every publish must be delivered, not merely most of them" ); - - drop(listener); } +/// Dispatch runs on the publishing thread with the listener's lock held, so the +/// bus must not also hold its registry lock across the callback — a listener +/// that subscribes from `on_event` would deadlock against its own publisher. +/// +/// Bounded and run off-thread because the failure mode is a hang, and a hang +/// reports nothing: the timeout turns it into a named failure. #[test] -fn test_custom_event_delivery() { - let bus = create_initialized_bus("custom_events"); - - let custom_topic = "my-custom-topic"; - let (tx, rx) = mpsc::channel(); - let listener: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx })); - bus.subscribe(custom_topic, listener.clone()); - - let payload = serde_json::json!({"key": "value", "count": 42}); - let event = Event::custom(custom_topic, payload); - let received = publish_until_received(&bus, &event, &rx, Duration::from_secs(5)); +fn a_listener_that_subscribes_from_on_event_does_not_deadlock() { + struct SubscribingListener { + bus: Arc, + added: Arc>>>>, + } + impl EventListener for SubscribingListener { + fn on_event(&mut self, _event: &Event) -> crate::core::error::Result<()> { + let (listener, _log) = RecordingListener::with_shared_log(); + let listener: Arc> = Arc::new(Mutex::new(listener)); + self.bus.subscribe(topics::MOUSE, Arc::clone(&listener)); + self.added.lock().push(listener); + Ok(()) + } + } - let received = received.expect("Custom event should be delivered"); - assert_eq!(received.topic(), custom_topic); + let bus = Arc::new(PubSub::new()); + let added = Arc::new(Mutex::new(Vec::new())); + let listener: Arc> = Arc::new(Mutex::new(SubscribingListener { + bus: Arc::clone(&bus), + added: Arc::clone(&added), + })); + bus.subscribe(topics::KEYBOARD, Arc::clone(&listener)); - // Verify payload fidelity - let original_bytes = rmp_serde::to_vec_named(&event).unwrap(); - let received_bytes = rmp_serde::to_vec_named(&received).unwrap(); - assert_eq!( - original_bytes, received_bytes, - "Custom event payload should survive iceoryx2 round-trip" - ); + let (done_tx, done_rx) = mpsc::channel(); + std::thread::spawn(move || { + let event = keyboard_event(); + bus.publish(&event.topic(), &event); + bus.flush(); + let _ = done_tx.send(()); + }); - drop(listener); + done_rx + .recv_timeout(Duration::from_secs(5)) + .expect("publish deadlocked against a listener that subscribed from on_event"); + assert_eq!(added.lock().len(), 1); } // =========================================================================== -// G. Edge cases & robustness +// D. Event vocabulary // =========================================================================== #[test] -fn test_oversized_event_is_dropped() { - let bus = create_initialized_bus("oversized_event"); - - let (tx, rx) = mpsc::channel(); - let listener: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx })); - bus.subscribe("big-topic", listener.clone()); - - // First verify the subscriber is working with a normal-sized event - let normal_event = Event::custom("big-topic", serde_json::json!({"ok": true})); - let received = publish_until_received(&bus, &normal_event, &rx, Duration::from_secs(5)); - assert!(received.is_some(), "Normal event should be delivered first"); - - // Drain any extra events from the retry loop - while rx.try_recv().is_ok() {} - - // Now try an oversized event — should be silently dropped - let large_string = "x".repeat(MAX_EVENT_PAYLOAD_SIZE + 1); - let oversized_event = Event::custom("big-topic", serde_json::json!({ "data": large_string })); - bus.publish(&oversized_event.topic(), &oversized_event); - - // Should not crash, and subscriber should receive nothing - match rx.recv_timeout(Duration::from_millis(200)) { - Err(mpsc::RecvTimeoutError::Timeout) => { /* expected — no event */ } - Ok(_) => panic!("Oversized event should NOT be delivered"), - Err(mpsc::RecvTimeoutError::Disconnected) => { - panic!("Channel disconnected unexpectedly") - } - } - - drop(listener); +fn keyboard_events_route_to_the_keyboard_topic() { + assert_eq!(keyboard_event().topic(), topics::KEYBOARD); } #[test] -fn test_concurrent_publish_from_multiple_threads() { - let bus = Arc::new(create_initialized_bus("concurrent_publish")); - - let (tx, rx) = mpsc::channel(); - let listener: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx })); - bus.subscribe(topics::KEYBOARD, listener.clone()); - - // Ensure subscriber is ready - let probe = Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed); - let received = publish_until_received(&bus, &probe, &rx, Duration::from_secs(5)); - assert!(received.is_some(), "Subscriber should be ready"); - - // Drain probe events - while rx.try_recv().is_ok() {} - - // Each thread's first publish creates a new thread-local iceoryx2 publisher. - // iceoryx2's subscriber needs a beat to establish its receive-side connection - // for each new publisher — bursting N publishers in parallel can drop early - // messages with "Unable to establish connection to new sender". Publish in a - // retry loop so later messages survive the connection setup. - let thread_count = 4; - let publishes_per_thread = 20; - let stop = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let mut handles = Vec::new(); - - for _ in 0..thread_count { - let bus = bus.clone(); - let stop = stop.clone(); - let handle = std::thread::spawn(move || { - for _ in 0..publishes_per_thread { - if stop.load(std::sync::atomic::Ordering::Relaxed) { - break; - } - let event = Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed); - bus.publish(&event.topic(), &event); - std::thread::sleep(Duration::from_millis(10)); - } - }); - handles.push(handle); - } - - // Collect at least one event; signal threads to stop as soon as we have it. - let mut received_count = 0; - let deadline = Instant::now() + Duration::from_secs(5); - while Instant::now() < deadline { - match rx.recv_timeout(Duration::from_millis(50)) { - Ok(_) => { - received_count += 1; - if received_count >= 1 { - stop.store(true, std::sync::atomic::Ordering::Relaxed); - // Keep draining briefly in case more arrive after stop signal - if received_count >= thread_count { - break; - } - } - } - Err(mpsc::RecvTimeoutError::Timeout) => { - if received_count > 0 { - break; - } - } - Err(mpsc::RecvTimeoutError::Disconnected) => break, - } - } - - for handle in handles { - handle.join().expect("Publisher thread panicked"); - } - - assert!( - received_count > 0, - "Should receive at least some events from concurrent publishers, got {}", - received_count - ); +fn mouse_events_route_to_the_mouse_topic() { + let event = Event::mouse(MouseButton::Left, (100.0, 200.0), MouseState::Pressed); + assert_eq!(event.topic(), topics::MOUSE); +} - drop(listener); +#[test] +fn processor_events_route_to_their_processors_topic() { + let processor_id = "audio-mixer"; + let event = Event::processor(processor_id, ProcessorEvent::Started); + assert_eq!(event.topic(), topics::processor(processor_id)); } #[test] -fn test_separate_pubsub_instances_are_isolated() { - let bus_a = create_initialized_bus("isolated_a"); - let bus_b = create_initialized_bus("isolated_b"); - - let (tx_a, rx_a) = mpsc::channel(); - let (tx_b, rx_b) = mpsc::channel(); - let handle_a: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx_a })); - let handle_b: Arc> = - Arc::new(Mutex::new(ChannelListener { sender: tx_b })); - - bus_a.subscribe(topics::KEYBOARD, handle_a.clone()); - bus_b.subscribe(topics::KEYBOARD, handle_b.clone()); - - // Verify bus_a's subscriber is working - let event = Event::keyboard(KeyCode::A, Modifiers::default(), KeyState::Pressed); - let received_a = publish_until_received(&bus_a, &event, &rx_a, Duration::from_secs(5)); - assert!( - received_a.is_some(), - "bus_a subscriber should receive the event" - ); +fn a_processor_event_reaches_a_subscriber_on_that_processors_topic() { + let bus = PubSub::new(); + let processor_id = "audio-mixer"; + let (_listener, received) = subscribe_recorder(&bus, &topics::processor(processor_id)); - // bus_b's subscriber should NOT have received anything from bus_a - assert!( - rx_b.try_recv().is_err(), - "bus_b subscriber should NOT receive events from bus_a" - ); + let event = Event::processor(processor_id, ProcessorEvent::Started); + bus.publish(&event.topic(), &event); + bus.flush(); - drop(handle_a); - drop(handle_b); + assert_eq!(received.lock()[0], event); } diff --git a/runtime/streamlib-engine/src/core/runtime/runtime.rs b/runtime/streamlib-engine/src/core/runtime/runtime.rs index 66767fcc2..1c4820010 100644 --- a/runtime/streamlib-engine/src/core/runtime/runtime.rs +++ b/runtime/streamlib-engine/src/core/runtime/runtime.rs @@ -195,10 +195,6 @@ impl Runner { let iceoryx2_node = Iceoryx2Node::new()?; tracing::info!("[new] iceoryx2 Node created"); - // Initialize global PUBSUB with iceoryx2 backend. - // Must happen before any subscribe() calls (GraphChangeListener below). - PUBSUB.init(&runtime_id, iceoryx2_node.clone()); - // Bring up the per-runtime surface-sharing service. Each runtime owns // a unique Unix socket at $XDG_RUNTIME_DIR/streamlib-.sock that // its polyglot subprocesses connect to via STREAMLIB_SURFACE_SOCKET. @@ -219,7 +215,7 @@ impl Runner { ); let listener: Arc> = Arc::new(Mutex::new(listener)); - // Subscribe to graph changes + // Subscribe to graph changes. PUBSUB.subscribe(topics::RUNTIME_GLOBAL, Arc::clone(&listener)); Ok(Arc::new(Self { @@ -423,7 +419,6 @@ impl Runner { // Create shared timing context - clock starts now let time = Arc::new(TimeContext::new()); - // Clone iceoryx2 Node (created in new() for early PUBSUB initialization) let iceoryx2_node = self.iceoryx2_node.clone(); // Create audio clock - platform-specific for best precision diff --git a/runtime/streamlib-engine/src/core/utils/loop_control.rs b/runtime/streamlib-engine/src/core/utils/loop_control.rs index daf7c8207..22ac8adf7 100644 --- a/runtime/streamlib-engine/src/core/utils/loop_control.rs +++ b/runtime/streamlib-engine/src/core/utils/loop_control.rs @@ -31,11 +31,6 @@ impl EventListener for ShutdownListener { } /// Run a loop that automatically exits on shutdown events. -/// -/// Host-only: both of its shutdown sources are inert inside a plugin cdylib — -/// the plugin image's `PUBSUB` is never `init()`ed (so the subscription below -/// never delivers) and the funnel's cdylib arm deliberately latches nothing in -/// the plugin image's copy of the engine. pub fn shutdown_aware_loop(mut f: F) -> std::result::Result<(), E> where F: FnMut() -> std::result::Result, @@ -48,9 +43,8 @@ where shutdown_flag: Arc::clone(&shutdown_flag), }; - // Subscribe to runtime global events (includes shutdown) - // IMPORTANT: We must keep the Arc alive for the duration of the loop! - // The event bus stores only weak references, so if we drop the Arc, the listener is lost. + // The Arc must outlive the loop: the bus stores only a Weak, so dropping it + // unsubscribes. let listener_arc: Arc> = Arc::new(Mutex::new(listener)); PUBSUB.subscribe(topics::RUNTIME_GLOBAL, Arc::clone(&listener_arc)); @@ -62,7 +56,7 @@ where // Main loop loop { // The latch is polled as well as the event because a request latched - // before the subscribe above leaves no event to receive. + // before this loop started leaves no event to receive. if shutdown_flag.load(Ordering::Relaxed) || crate::core::runtime::is_runtime_shutdown_requested() { @@ -139,52 +133,46 @@ mod tests { #[test] #[serial] fn test_shutdown_event_exits_loop() { - use crate::iceoryx2::Iceoryx2Node; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::mpsc; use std::time::Duration; - // Ensure PUBSUB has an iceoryx2 backend. Use a process-unique runtime_id - // so iceoryx2's persistent service state under /tmp/iceoryx2/ doesn't - // collide with stale state left by crashed prior cargo-test invocations - // (which surfaced as PublishSubscribeOpenError(ServiceInCorruptedState)). - // If PUBSUB was already initialized by another test in this process, - // init() is a no-op (OnceLock), and the existing runtime_id is used. - if let Ok(node) = Iceoryx2Node::new() { - let runtime_id = format!("test-loop-control-{}", uuid::Uuid::new_v4()); - PUBSUB.init(&runtime_id, node); - } - let counter = Arc::new(AtomicUsize::new(0)); let counter_clone = Arc::clone(&counter); let (done_tx, done_rx) = mpsc::channel::>(); + let (entered_loop_tx, entered_loop_rx) = mpsc::channel::<()>(); std::thread::spawn(move || { let result = shutdown_aware_loop(|| { - counter_clone.fetch_add(1, Ordering::Relaxed); + // The callback runs only after `shutdown_aware_loop` has + // subscribed, so the first invocation is the handshake this + // thread owes the publisher below. + if counter_clone.fetch_add(1, Ordering::Relaxed) == 0 { + let _ = entered_loop_tx.send(()); + } std::thread::sleep(Duration::from_millis(10)); Ok::(LoopControl::Continue) }); done_tx.send(result).ok(); }); - // Give the iceoryx2 subscriber thread time to open the service and - // start polling before we send the shutdown event. - std::thread::sleep(Duration::from_millis(150)); + // Synchronising on the loop having started, not on a duration: the + // subscription is live the moment `shutdown_aware_loop` registers it, so + // the entry handshake is the only thing left to wait for. + entered_loop_rx + .recv_timeout(Duration::from_secs(5)) + .expect("the loop thread entered its callback"); - // Publish shutdown event let shutdown_event = Event::RuntimeGlobal(RuntimeEvent::RuntimeShutdown); PUBSUB.publish(&shutdown_event.topic(), &shutdown_event); - // Wait for loop to exit with a hard timeout so the test fails clearly - // rather than hanging indefinitely when PUBSUB is not functional. + // Bounded so a regression fails by name rather than hanging the suite. match done_rx.recv_timeout(Duration::from_secs(5)) { Ok(result) => assert!(result.is_ok(), "Loop returned an error"), Err(_) => panic!( "test_shutdown_event_exits_loop: loop did not exit within 5 s \ - after shutdown event — PUBSUB may be uninitialized or the \ - iceoryx2 subscriber thread failed to open its service" + after the shutdown event" ), } diff --git a/runtime/streamlib-engine/src/iceoryx2/mod.rs b/runtime/streamlib-engine/src/iceoryx2/mod.rs index 35d742041..ce2377fcf 100644 --- a/runtime/streamlib-engine/src/iceoryx2/mod.rs +++ b/runtime/streamlib-engine/src/iceoryx2/mod.rs @@ -28,17 +28,13 @@ pub(crate) use delivery_profile::delivery_profile_for_input_port; pub use delivery_profile::{DeliveryProfile, DeliveryResolution}; pub use input::{BoundedReadOutcome, InputMailboxes, InputMailboxesInner}; pub use mailbox::PortMailbox; -pub use node::{ - ChannelTapSubscribeError, Iceoryx2EventService, Iceoryx2Node, Iceoryx2NotifyService, - Iceoryx2Service, -}; +pub use node::{ChannelTapSubscribeError, Iceoryx2Node, Iceoryx2NotifyService, Iceoryx2Service}; pub use output::{ChannelEgressConfig, OutputWriter, OutputWriterInner}; pub use overflow::Overflow; pub use payload::{ - ChannelTrustTier, DEFAULT_EXPECTED_PAYLOAD_BYTES, DEFAULT_MAX_QUEUED_MESSAGES, EventPayload, - FRAME_HEADER_SIZE, FrameHeader, MAX_EVENT_PAYLOAD_SIZE, MAX_PUBLISHERS_PER_CHANNEL, - MAX_TOPIC_KEY_SIZE, PortKey, RESERVED_TAP_SUBSCRIBER_SLOTS_PER_CHANNEL, - TRUSTED_CHANNEL_PAYLOAD_CEILING_BYTES, TopicKey, + ChannelTrustTier, DEFAULT_EXPECTED_PAYLOAD_BYTES, DEFAULT_MAX_QUEUED_MESSAGES, + FRAME_HEADER_SIZE, FrameHeader, MAX_PUBLISHERS_PER_CHANNEL, PortKey, + RESERVED_TAP_SUBSCRIBER_SLOTS_PER_CHANNEL, TRUSTED_CHANNEL_PAYLOAD_CEILING_BYTES, UNTRUSTED_SESSION_CHANNEL_PAYLOAD_CEILING_BYTES, }; pub use read_mode::ReadMode; diff --git a/runtime/streamlib-engine/src/iceoryx2/node.rs b/runtime/streamlib-engine/src/iceoryx2/node.rs index 6ed0907d5..b07c235b9 100644 --- a/runtime/streamlib-engine/src/iceoryx2/node.rs +++ b/runtime/streamlib-engine/src/iceoryx2/node.rs @@ -11,7 +11,7 @@ use iceoryx2::port::notifier::Notifier; use iceoryx2::prelude::*; use parking_lot::Mutex; -use super::{EventPayload, FRAME_HEADER_SIZE, MAX_PUBLISHERS_PER_CHANNEL}; +use super::{FRAME_HEADER_SIZE, MAX_PUBLISHERS_PER_CHANNEL}; use crate::core::error::{Error, Result}; /// Thread-safe wrapper for iceoryx2 Node. @@ -35,26 +35,6 @@ impl Iceoryx2Node { }) } - /// Open or create a publish-subscribe service for EventPayload. - /// - /// The service name should follow the format: "streamlib/{runtime_id}/events/{topic}" - pub fn open_or_create_event_service(&self, service_name: &str) -> Result { - let node = self.inner.lock(); - let service_name: ServiceName = service_name.try_into().map_err(|e| { - Error::Configuration(format!("Invalid service name '{}': {:?}", service_name, e)) - })?; - - let service = node - .service_builder(&service_name) - .publish_subscribe::() - .max_publishers(16) - .subscriber_max_buffer_size(64) - .open_or_create() - .map_err(|e| Error::Runtime(format!("Failed to open/create event service: {:?}", e)))?; - - Ok(Iceoryx2EventService { inner: service }) - } - /// Open or create an iceoryx2 Event service for fd-multiplexed wakeups. /// /// Pairs with a destination's data channels for fd-multiplexed wakeups: the @@ -62,8 +42,7 @@ impl Iceoryx2Node { /// destination waits on ONE `Listener` fd regardless of fan-in, while every /// upstream source publishing into one of its channels holds a `Notifier` /// here. `max_notifiers` is the destination's compile-time fan-in (the count - /// of inbound links). Distinct from [`Iceoryx2EventService`] which is a typed - /// pub/sub for runtime events. + /// of inbound links). pub fn open_or_create_notify_service( &self, service_name: &str, @@ -227,8 +206,7 @@ pub enum ChannelTapSubscribeError { /// Handle to an iceoryx2 Event service used for fd-multiplexed wakeups. /// -/// Distinct from [`Iceoryx2EventService`] (which is a typed pub/sub for runtime events). -/// This wraps iceoryx2's `MessagingPattern::Event` — `Notifier::notify()` causes any +/// Wraps iceoryx2's `MessagingPattern::Event` — `Notifier::notify()` causes any /// `Listener` on the same service to become readable on its underlying fd. pub struct Iceoryx2NotifyService { inner: iceoryx2::service::port_factory::event::PortFactory, @@ -252,38 +230,6 @@ impl Iceoryx2NotifyService { } } -/// Handle to an iceoryx2 publish-subscribe service for events. -pub struct Iceoryx2EventService { - inner: iceoryx2::service::port_factory::publish_subscribe::PortFactory< - ipc::Service, - EventPayload, - (), - >, -} - -impl Iceoryx2EventService { - /// Create a publisher for this event service. - pub fn create_publisher( - &self, - ) -> Result> { - self.inner - .publisher_builder() - .create() - .map_err(|e| Error::Runtime(format!("Failed to create event publisher: {:?}", e))) - } - - /// Create a subscriber for this event service. - pub fn create_subscriber( - &self, - ) -> Result> { - self.inner - .subscriber_builder() - .buffer_size(64) - .create() - .map_err(|e| Error::Runtime(format!("Failed to create event subscriber: {:?}", e))) - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/runtime/streamlib-engine/src/iceoryx2/payload.rs b/runtime/streamlib-engine/src/iceoryx2/payload.rs index 75c8f2cd8..c6bc27f42 100644 --- a/runtime/streamlib-engine/src/iceoryx2/payload.rs +++ b/runtime/streamlib-engine/src/iceoryx2/payload.rs @@ -7,9 +7,8 @@ //! the wheel's helper-process transport share the same wire-compatible types. pub use streamlib_ipc_types::{ - ChannelTrustTier, DEFAULT_EXPECTED_PAYLOAD_BYTES, DEFAULT_MAX_QUEUED_MESSAGES, EventPayload, - FRAME_HEADER_SIZE, FrameHeader, MAX_EVENT_PAYLOAD_SIZE, MAX_PUBLISHERS_PER_CHANNEL, - MAX_TOPIC_KEY_SIZE, PortKey, RESERVED_TAP_SUBSCRIBER_SLOTS_PER_CHANNEL, - TRUSTED_CHANNEL_PAYLOAD_CEILING_BYTES, TopicKey, + ChannelTrustTier, DEFAULT_EXPECTED_PAYLOAD_BYTES, DEFAULT_MAX_QUEUED_MESSAGES, + FRAME_HEADER_SIZE, FrameHeader, MAX_PUBLISHERS_PER_CHANNEL, PortKey, + RESERVED_TAP_SUBSCRIBER_SLOTS_PER_CHANNEL, TRUSTED_CHANNEL_PAYLOAD_CEILING_BYTES, UNTRUSTED_SESSION_CHANNEL_PAYLOAD_CEILING_BYTES, }; diff --git a/runtime/streamlib-ipc-types/src/lib.rs b/runtime/streamlib-ipc-types/src/lib.rs index 01616b807..c8c74496d 100644 --- a/runtime/streamlib-ipc-types/src/lib.rs +++ b/runtime/streamlib-ipc-types/src/lib.rs @@ -27,8 +27,6 @@ use iceoryx2::prelude::*; /// multi-MB keyframe) free to grow rather than crash. pub const DEFAULT_EXPECTED_PAYLOAD_BYTES: usize = 65536; pub const MAX_PORT_KEY_SIZE: usize = 64; -pub const MAX_EVENT_PAYLOAD_SIZE: usize = 8192; -pub const MAX_TOPIC_KEY_SIZE: usize = 128; /// Per-channel payload ceiling for a trusted (in-process host) data channel — /// the graceful, observable layer in front of the subprocess cgroup @@ -446,100 +444,6 @@ impl FrameHeader { } } -/// Fixed-size topic name for event pub/sub IPC. -#[derive(Clone, Copy, Eq, PartialEq, Hash, Debug, ZeroCopySend)] -#[repr(C)] -pub struct TopicKey { - len: u8, - name: [u8; MAX_TOPIC_KEY_SIZE - 1], -} - -impl TopicKey { - pub fn new(name: &str) -> Self { - let bytes = name.as_bytes(); - let len = bytes.len().min(MAX_TOPIC_KEY_SIZE - 1) as u8; - let mut key = Self { - len, - name: [0u8; MAX_TOPIC_KEY_SIZE - 1], - }; - key.name[..len as usize].copy_from_slice(&bytes[..len as usize]); - key - } - - pub fn as_str(&self) -> &str { - std::str::from_utf8(&self.name[..self.len as usize]).unwrap_or("") - } -} - -impl Default for TopicKey { - fn default() -> Self { - Self { - len: 0, - name: [0u8; MAX_TOPIC_KEY_SIZE - 1], - } - } -} - -/// Event payload for iceoryx2 pub/sub communication. -/// -/// Carries serialized runtime events (lifecycle, graph changes, compiler, input) -/// between components via iceoryx2 shared memory. -#[derive(Clone, Copy, ZeroCopySend)] -#[type_name("EventPayload")] -#[repr(C)] -pub struct EventPayload { - pub topic_key: TopicKey, - pub timestamp_ns: i64, - pub len: u32, - pub data: [u8; MAX_EVENT_PAYLOAD_SIZE], -} - -impl EventPayload { - /// Create a new event payload with the given topic and serialized data. - pub fn new(topic: &str, timestamp_ns: i64, data: &[u8]) -> Self { - let len = data.len().min(MAX_EVENT_PAYLOAD_SIZE) as u32; - let mut payload = Self { - topic_key: TopicKey::new(topic), - timestamp_ns, - len, - data: [0u8; MAX_EVENT_PAYLOAD_SIZE], - }; - payload.data[..len as usize].copy_from_slice(&data[..len as usize]); - payload - } - - /// Get the actual data slice (excluding padding). - pub fn data(&self) -> &[u8] { - &self.data[..self.len as usize] - } - - /// Get the topic key as a string. - pub fn topic(&self) -> &str { - self.topic_key.as_str() - } -} - -impl Default for EventPayload { - fn default() -> Self { - Self { - topic_key: TopicKey::default(), - timestamp_ns: 0, - len: 0, - data: [0u8; MAX_EVENT_PAYLOAD_SIZE], - } - } -} - -impl std::fmt::Debug for EventPayload { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("EventPayload") - .field("topic_key", &self.topic_key.as_str()) - .field("timestamp_ns", &self.timestamp_ns) - .field("len", &self.len) - .finish() - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/xtask/src/check_clock_usage.rs b/xtask/src/check_clock_usage.rs index 96323a1aa..7f39d9c9e 100644 --- a/xtask/src/check_clock_usage.rs +++ b/xtask/src/check_clock_usage.rs @@ -1,21 +1,21 @@ // Copyright (c) 2025 Jonathan Fontanez // SPDX-License-Identifier: BUSL-1.1 -//! Bans wall-clock reads outside the four observability surfaces the plan +//! Bans wall-clock reads outside the three observability surfaces the plan //! permits them on (`docs/plan/ARCHITECTURE.md` §Media I/O //! `[one-monotonic-clock]`; rationale in `docs/decisions/one-monotonic-clock.md`). //! //! Monotonic is the only legal clock on the data plane. A wall-clock value and a //! media timestamp share a unit and are different quantities, so a subtraction //! across them is always a bug — and it is an easy bug to write, because -//! `SystemTime::now()` is the reflexive spelling for "what time is it". The four +//! `SystemTime::now()` is the reflexive spelling for "what time is it". The //! surfaces that keep wall clock correlate StreamLib with the outside world and //! with other hosts' logs, a job monotonic time cannot do. //! //! There is no per-line pragma and no opt-out attribute. The file allowlist is -//! the only way past this gate, every entry names one of exactly four -//! [`ObservabilitySurface`] variants, and a fifth surface is a plan change — so -//! widening the list means adding a variant, which no one does by accident. +//! the only way past this gate, every entry names an [`ObservabilitySurface`] +//! variant, and a further surface is a plan change — so widening the list means +//! adding a variant, which no one does by accident. //! //! Cheap substring scan, no `syn` and no compile. Whole-line `//` and `#` //! comments and Python triple-quoted spans are blanked first, so a doc comment @@ -55,18 +55,17 @@ const SCAN_ROOTS: &[&str] = &[ /// Files whose *source text* spells a banned pattern without reading a clock — /// this gate's own constants and fixtures. Not allowlist entries: the -/// permitted-surface list stays exactly the four the plan names, and nothing +/// permitted-surface list stays exactly the ones the plan names, and nothing /// here is licensed to read a wall clock. const SCAN_EXEMPT_FILES: &[&str] = &["xtask/src/check_clock_usage.rs"]; -/// The four surfaces the plan permits a wall-clock read on. Adding a fifth is a -/// plan change, so it is a variant here before it is a line in the allowlist. +/// The surfaces the plan permits a wall-clock read on. Adding one is a plan +/// change, so it is a variant here before it is a line in the allowlist. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ObservabilitySurface { LogRecordHostTimestamp, LogRecordSourceTimestamp, LogFileName, - ControlPlaneEventTimestamp, } impl ObservabilitySurface { @@ -74,7 +73,6 @@ impl ObservabilitySurface { ObservabilitySurface::LogRecordHostTimestamp, ObservabilitySurface::LogRecordSourceTimestamp, ObservabilitySurface::LogFileName, - ObservabilitySurface::ControlPlaneEventTimestamp, ]; pub const fn label(self) -> &'static str { @@ -82,9 +80,6 @@ impl ObservabilitySurface { ObservabilitySurface::LogRecordHostTimestamp => "log record `host_ts`", ObservabilitySurface::LogRecordSourceTimestamp => "log record `source_ts`", ObservabilitySurface::LogFileName => "log file naming", - ObservabilitySurface::ControlPlaneEventTimestamp => { - "control-plane pubsub event `timestamp_ns`" - } } } } @@ -118,11 +113,6 @@ const PERMITTED_WALL_CLOCK_SURFACES: &[PermittedWallClockSurface] = &[ surface: ObservabilitySurface::LogFileName, reason: "mints `started_at_millis`, which humans read off the JSONL file name", }, - PermittedWallClockSurface { - path: "runtime/streamlib-engine/src/core/pubsub/bus.rs", - surface: ObservabilitySurface::ControlPlaneEventTimestamp, - reason: "stamps control-plane events, which are correlated against outside-world clocks", - }, ]; pub struct ClockUsageLanguage { diff --git a/xtask/src/main.rs b/xtask/src/main.rs index 864a758bd..76e5c59b5 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -268,10 +268,10 @@ enum Commands { /// CI gate for the wall-clock allowlist. Fails on a wall-clock read /// (`SystemTime::now`, `Utc::now`, `time.time_ns`, `datetime.now`, …) /// anywhere under `runtime/ sdk/ adapters/ xtask/ packages/test-fixtures/` - /// outside the four + /// outside the three /// observability surfaces the plan permits it on: log record `host_ts` - /// and `source_ts`, log file naming, and the control-plane pubsub event - /// timestamp. Monotonic is the only legal clock on the data plane — a + /// and `source_ts`, and log file naming. Monotonic is the only legal clock + /// on the data plane — a /// wall-clock value and a media timestamp share a unit and are different /// quantities, so subtracting across them is always a bug. There is no /// per-line pragma: widening the list is a plan change. See