feat(api-server): real-time SSE event stream for block explorers - #2117
Conversation
Mechanical, behavior-preserving fixes for the lints introduced by the newer clippy (new_without_default, let_and_return, useless_borrows / redundant references, some_filter, unused imports), so that do_checks.sh passes with the current toolchain.
Add a streaming endpoint (GET /api/v2/stream) that delivers three event kinds to explorer clients as named Server-Sent Events (tx_seen, block, reorg), with a 'types' query filter, keepalives, an x-accel-buffering response header, a lag advisory for slow clients, and an in-stream reconnection hint. Event sources: * the scanner appends block/reorg events to a new ml.emitted_events table inside the same transaction that indexes the blocks, so every event is transactionally consistent with the data the explorer can immediately fetch over REST; a pg_notify wakeup is sent on commit; * the web server bridges mempool events from the node's WebSocket RPC into tx_seen events (only successfully processed transactions). The web server runs an event pump that wakes up on the Postgres notification (LISTEN on a dedicated connection, with automatic reconnection), falls back to periodic polling, drains the backlog in bounded batches, and forwards the events into a tokio broadcast channel consumed by the SSE endpoint. Events older than a retention window are pruned by the scanner. Notes: * the storage version is bumped (25 -> 26) since the schema changed, which triggers the usual full re-initialization on upgrade; * the in-memory storage backend treats streaming as a no-op; * storage-test-suite gains streaming tests for both backends; * stack tests cover the SSE endpoint contract and, against a real Postgres, the full scanner -> emitted_events -> pump -> SSE flow including a reorg scenario; * the podman-based container helper falls back to docker when podman is not installed.
Document the /v2/stream SSE endpoint (event kinds, filtering, wire format, keepalives, lag advisory, no-replay semantics), the new web server streaming options, the event flow architecture, and the storage version 25 -> 26 upgrade (full resync) in the README, and add the corresponding changelog entries.
|
🔍 OpenCodeReview found 20 issue(s) in this PR.
📄
|
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| .execute( | ||
| "DELETE FROM ml.emitted_events | ||
| WHERE id <= (SELECT COALESCE(max(id), 0) - $1 FROM ml.emitted_events);", | ||
| &[&STREAM_EVENTS_RETENTION_COUNT], | ||
| ) |
There was a problem hiding this comment.
Retention pruning is based solely on the global max(id) minus STREAM_EVENTS_RETENTION_COUNT, with no consideration of how far the event pump (or a slow consumer) has progressed. If the pump lags behind by more than 10,000 events (e.g. after a database outage or a long indexing burst), unread events will be deleted and clients connected via the stream will silently miss them. Consider pruning only below the last consumed cursor, or at least logging a warning when the pruned boundary has overtaken last_seen_id.
Suggestion:
| .execute( | |
| "DELETE FROM ml.emitted_events | |
| WHERE id <= (SELECT COALESCE(max(id), 0) - $1 FROM ml.emitted_events);", | |
| &[&STREAM_EVENTS_RETENTION_COUNT], | |
| ) | |
| .execute( | |
| "DELETE FROM ml.emitted_events | |
| WHERE id <= (SELECT COALESCE(max(id), 0) - $1 FROM ml.emitted_events);", | |
| &[&STREAM_EVENTS_RETENTION_COUNT], | |
| ) | |
| // TODO: consider pruning below the pump's last_seen_id or warn when unread events | |
| // fall out of the retention window. |
| podman.run(); | ||
|
|
||
| let host_port = podman.get_port_mapping(5432).unwrap(); |
There was a problem hiding this comment.
Podman::new/run()/get_port_mapping() execute synchronous process invocations (container start can take seconds) directly on the tokio runtime thread of this async test. All subsequently spawned tasks (event pump, SSE collector, web server) run on the same current-thread runtime, so the blocking startup starves them and can distort the timing-sensitive assertions later in the test. Prefer running the container lifecycle via tokio::task::spawn_blocking (or std::thread) before the async work begins.
Suggestion:
| podman.run(); | |
| let host_port = podman.get_port_mapping(5432).unwrap(); | |
| tokio::task::spawn_blocking(move || { | |
| podman.run(); | |
| podman.get_port_mapping(5432).unwrap() | |
| }) | |
| .await | |
| .unwrap(); |
| tokio::spawn(async move { | ||
| let mut connect_attempts = 0u32; | ||
| let client = reqwest::Client::new(); |
There was a problem hiding this comment.
The SSE collector task's JoinHandle is dropped, so a panic inside it (e.g. the x-accel-buffering assert at line ~177, serde_json::from_str expect, or the UTF-8 unwrap) goes unnoticed: the test instead fails later with a misleading timeout on recv_event or an "unexpected extra event" panic, hiding the root cause. Retain the handle and check it (e.g. via handle.is_finished()/abort + assert! on the result) at the end of the test, or send failures through the channel.
Suggestion:
| tokio::spawn(async move { | |
| let mut connect_attempts = 0u32; | |
| let client = reqwest::Client::new(); | |
| let collector_task = tokio::spawn(async move { | |
| let mut connect_attempts = 0u32; | |
| let client = reqwest::Client::new(); |
| tokio::time::sleep(Duration::from_secs(1)).await; | ||
| match tokio::time::timeout(Duration::from_secs(1), event_rx.recv()).await { |
There was a problem hiding this comment.
This "no duplicates" negative assertion is purely timing-based: podman.run() and other synchronous work earlier in the same #[tokio::test] (single-threaded runtime by default) can delay task scheduling, so a late-but-legitimate duplicate could arrive after the 1s settle + 1s observation window and silently pass, or a slow CI machine could flake. Consider a longer observation window (e.g. a few seconds) or asserting only that no duplicate of the already-consumed events is received after a full event-pump poll cycle.
Suggestion:
| tokio::time::sleep(Duration::from_secs(1)).await; | |
| match tokio::time::timeout(Duration::from_secs(1), event_rx.recv()).await { | |
| tokio::time::sleep(Duration::from_secs(1)).await; | |
| match tokio::time::timeout(Duration::from_secs(5), event_rx.recv()).await { |
| pub async fn run_database_event_pump(source: impl StreamEventSource, handle: StreamEventsHandle) { | ||
| run_event_pump(source, handle.channel, 0).await; | ||
| } |
There was a problem hiding this comment.
The event pump is always started with last_seen_id = 0, so on every server restart it re-reads and re-broadcasts all retained events (up to STREAM_EVENTS_RETENTION_COUNT = 10,000) to connected subscribers as if they were fresh. Connected clients receive a large replay of stale Block/Reorg/TxSeen events at startup, which contradicts the documented no-replay semantics (see STREAM_EVENTS_RETENTION_COUNT doc: "the stream endpoint does not support replays anyway"). Since PostgresStreamEventSource already holds the storage handle, the initial id should be initialized from the latest committed event id (e.g. a read_last_stream_event_id storage query, or SELECT max(id) FROM ml.emitted_events) before entering run_event_pump.
Suggestion:
| pub async fn run_database_event_pump(source: impl StreamEventSource, handle: StreamEventsHandle) { | |
| run_event_pump(source, handle.channel, 0).await; | |
| } | |
| // The initial last-seen id must come from storage so that a server restart does not | |
| // replay the retained backlog; `0` causes up to 10,000 old events to be re-broadcast. | |
| pub async fn run_database_event_pump(mut source: impl StreamEventSource, handle: StreamEventsHandle) { | |
| let last_seen_id = source.latest_event_id().await.unwrap_or(0); | |
| run_event_pump(source, handle.channel, last_seen_id).await; | |
| } |
| tokio::time::sleep(MEMPOOL_RESUBSCRIBE_DELAY).await; | ||
| } | ||
| } |
There was a problem hiding this comment.
Reconnect retries with a fixed 1-second delay and no backoff, and every failed iteration logs an error. If the node RPC is down for a long period this produces an unbounded stream of log messages at 1/sec (plus a subscription attempt per second). Consider exponential backoff with a cap, consistent with the checklist's retry-loop guidance.
Suggestion:
| tokio::time::sleep(MEMPOOL_RESUBSCRIBE_DELAY).await; | |
| } | |
| } | |
| tokio::time::sleep(MEMPOOL_RESUBSCRIBE_DELAY).await; | |
| // Consider: exponential backoff (e.g. doubling up to a cap) to avoid | |
| // sustained per-second error logging while the node is unreachable. | |
| } | |
| } |
* start the event pump from the most recently committed event id, so a web server restart cannot re-broadcast old events as fresh (the stream has no replay semantics); * log a warning when stream event pruning deletes rows, and expose the latest event id through the storage read interface; * use an exponential backoff (capped at 60s) for the mempool subscription retries, resetting on success; * supervise the streaming background tasks so a terminal failure is logged instead of silently disabling the stream; * restore a doc comment line accidentally merged in block_status; * make the postgres stream test robust (container startup off the async runtime, collector panics propagated, wider no-duplicates window) and log the chosen container manager in the test helper.
|
Thanks @github-actions[bot] — findings addressed in dcf705b:
Intentionally not changed:
|
| // connected clients, so the failure is retried until the database answers. | ||
| logging::log::error!("Failed to read the latest stream event id: {err}"); | ||
| tokio::time::sleep(self.poll_interval).await; | ||
| Box::pin(self.initial_last_seen_id()).await |
There was a problem hiding this comment.
initial_last_seen_id retries a database failure via unbounded recursion (Box::pin(self.initial_last_seen_id()).await). Each retry builds and awaits a new boxed future inside the previous one, so during a prolonged database outage the chain of pinned futures grows without bound (memory growth), and there is no cancellation path. Since the sole production caller is run_database_event_pump, which simply awaits this once before the pump loop, this can be rewritten as an iterative loop with the existing poll_interval backoff (which is already bounded per attempt) with no recursive self-call.
Suggestion:
| Box::pin(self.initial_last_seen_id()).await | |
| // Retry iteratively to avoid building an unbounded chain of boxed futures | |
| // during a prolonged database outage. | |
| loop { | |
| match self.latest_event_id().await { | |
| Ok(id) => break id, | |
| Err(err) => { | |
| logging::log::error!("Failed to read the latest stream event id: {err}"); | |
| tokio::time::sleep(self.poll_interval).await; | |
| } | |
| } | |
| } |
| .await | ||
| .map_err(|e| ApiServerStorageError::LowLevelStorageError(e.to_string()))?; | ||
|
|
||
| Ok(deleted) | ||
| } |
There was a problem hiding this comment.
This DELETE runs inside every block-indexing write transaction (see scanner-lib/src/blockchain_state/mod.rs calling db_tx.prune_stream_events() before commit), and its max(id) subquery plus delete adds write-path latency on the critical indexing path even though it almost always deletes nothing. Consider tracking the last-pruned id (in memory or storage) and only issuing the DELETE when the log has actually grown past the retention window, or moving pruning to a periodic maintenance task outside the indexing transaction.
| tokio::time::sleep(Duration::from_secs(3)).await; | ||
| match tokio::time::timeout(Duration::from_secs(3), event_rx.recv()).await { | ||
| Err(_timed_out) => {} // no more events, as expected |
There was a problem hiding this comment.
This panic-on-any-extra-event check is inherently timing-sensitive: a slow CI machine could deliver committed events (e.g. via the safety-net poll path of the pump) after the 3s settle window, producing a flaky failure. Consider deriving the 'no more events' expectation from a deterministic signal (e.g. a scanner-side marker that all expected events have been consumed by the pump) or at least making the settle window part of the expected-event protocol rather than wall-clock sleeps.
Additionally, web_task is aborted but never joined, unlike collector_task; a panic inside the web server task (which could itself be the root cause of a missing event) would go unobserved and surface only as the misleading timeout/extra-event failures above.
Suggestion:
| tokio::time::sleep(Duration::from_secs(3)).await; | |
| match tokio::time::timeout(Duration::from_secs(3), event_rx.recv()).await { | |
| Err(_timed_out) => {} // no more events, as expected | |
| tokio::time::sleep(Duration::from_secs(3)).await; | |
| match tokio::time::timeout(Duration::from_secs(3), event_rx.recv()).await { | |
| Err(_timed_out) => {} // no more events, as expected | |
| } | |
| // (consider also joining web_task after abort, mirroring the collector_task handling below) |
| fn supervise(name: &'static str, handle: tokio::task::JoinHandle<()>) { | ||
| // Note: the background tasks are meant to run forever; without this supervisor, a panic or | ||
| // any other terminal failure would go completely unnoticed while the REST endpoints keep | ||
| // working, silently disabling the event stream. | ||
| tokio::spawn(async move { | ||
| if let Err(err) = handle.await { | ||
| logging::log::error!("CRITICAL: the {name} task terminated: {err}"); | ||
| } | ||
| }); | ||
| } |
There was a problem hiding this comment.
supervise only logs when a background task terminates, so a panic in the event pump or the mempool bridge silently disables the whole real-time stream while REST endpoints keep serving; operators get no restart, no health endpoint impact, and the log line is easy to miss. Consider either restarting the task with backoff, or propagating the failure (e.g. process exit, a health/ready flag in StreamEventsHandle surfaced via an endpoint) so the dead stream is detectable.
Suggestion:
| fn supervise(name: &'static str, handle: tokio::task::JoinHandle<()>) { | |
| // Note: the background tasks are meant to run forever; without this supervisor, a panic or | |
| // any other terminal failure would go completely unnoticed while the REST endpoints keep | |
| // working, silently disabling the event stream. | |
| tokio::spawn(async move { | |
| if let Err(err) = handle.await { | |
| logging::log::error!("CRITICAL: the {name} task terminated: {err}"); | |
| } | |
| }); | |
| } | |
| tokio::spawn(async move { | |
| if let Err(err) = handle.await { | |
| logging::log::error!("CRITICAL: the {name} task terminated: {err}"); | |
| // Surface the failure to health checks / restart the task here. | |
| } | |
| }); |
| let channel = StreamEventsChannel::new(args.stream_events_broadcast_capacity.max(1)); | ||
| let config = streaming::StreamingConfig { | ||
| keepalive_interval: std::time::Duration::from_secs( | ||
| args.stream_events_keepalive_interval_secs.max(1), | ||
| ), | ||
| }; |
There was a problem hiding this comment.
Invalid operator input is silently clamped instead of rejected: --stream-events-broadcast-capacity 0 becomes a capacity of 1, which makes every subscriber lag after a single buffered event, and a 0 for the interval options becomes 1 second. Since these are CLI flags with clap, prefer value_parser range validation (or an explicit validation error) over silent clamping so the operator's mistake is visible.
Suggestion:
| let channel = StreamEventsChannel::new(args.stream_events_broadcast_capacity.max(1)); | |
| let config = streaming::StreamingConfig { | |
| keepalive_interval: std::time::Duration::from_secs( | |
| args.stream_events_keepalive_interval_secs.max(1), | |
| ), | |
| }; | |
| let channel = StreamEventsChannel::new( | |
| args.stream_events_broadcast_capacity | |
| .max(1), // consider validating instead of clamping | |
| ); |
| match MempoolRpcClient::subscribe_to_events(rpc.ws_client()).await { | ||
| Ok(subscription) => { |
There was a problem hiding this comment.
The mempool bridge reconnect loop calls MempoolRpcClient::subscribe_to_events with no timeout. If the WebSocket handshake hangs (e.g. the node accepts TCP but never completes the RPC handshake), the loop blocks forever inside await with no log output and no backoff progress, silently disabling TxSeen streaming. Consider wrapping the subscribe call in tokio::time::timeout so a stalled connection is logged and retried.
Suggestion:
| match MempoolRpcClient::subscribe_to_events(rpc.ws_client()).await { | |
| Ok(subscription) => { | |
| match tokio::time::timeout(SUBSCRIBE_TIMEOUT, MempoolRpcClient::subscribe_to_events(rpc.ws_client())).await { | |
| Ok(Ok(subscription)) => { ... } | |
| Ok(Err(err)) => { ... } | |
| Err(_) => logging::log::error!("Timed out subscribing to node mempool events"), | |
| } |
…D_TESTS The test requires a container runtime, which only the ubuntu CI runners provide; skip it like the storage backend postgres test does on the other platforms.
| logging::log::error!("Failed to read the latest stream event id: {err}"); | ||
| tokio::time::sleep(self.poll_interval).await; | ||
| Box::pin(self.initial_last_seen_id()).await |
There was a problem hiding this comment.
initial_last_seen_id retries forever via unbounded async recursion (Box::pin(self.initial_last_seen_id()).await) with no backoff beyond poll_interval and no cancellation or observability: with a persistently failing database, the spawned pump task spins silently (the supervise wrapper in web-server's main.rs only logs if the task terminates, which it never does) and the event stream stays dead without anyone noticing. Consider a bounded retry that returns a Result and lets the caller decide (e.g. log CRITICAL and exit), or an iterative loop with exponential backoff, rather than unbounded recursion.
Suggestion:
| logging::log::error!("Failed to read the latest stream event id: {err}"); | |
| tokio::time::sleep(self.poll_interval).await; | |
| Box::pin(self.initial_last_seen_id()).await | |
| logging::log::error!("Failed to read the latest stream event id: {err}"); | |
| // Iterative retry with backoff; bail out after a bounded number of attempts so | |
| // a dead database is observable instead of wedging the pump forever. | |
| let mut delay = self.poll_interval; | |
| loop { | |
| tokio::time::sleep(delay).await; | |
| match self.latest_event_id().await { | |
| Ok(id) => break id, | |
| Err(err) => { | |
| logging::log::error!("Failed to read the latest stream event id: {err}"); | |
| delay = std::cmp::min(delay * 2, MAX_RETRY_DELAY); | |
| } | |
| } | |
| } |
| match serde_json::from_str(&payload) { | ||
| Ok(event) => Some((id, event)), | ||
| Err(err) => { | ||
| logging::log::warn!("Skipping undecodable stream event #{id}: {err}"); | ||
| None | ||
| } | ||
| } |
There was a problem hiding this comment.
Skipping an undecodable row silently drops that event from the stream: the pump advances last_seen_id past the id, so the event is never broadcast and clients get no indication of the gap. The warning log alone is easy to miss in production. Since the event was durably committed alongside its indexed data, consider treating a decode failure as a hard error (surfaced via StreamEventReadError, which the pump already tolerates and logs) or forwarding a placeholder/error event so clients can resync via REST instead of silently missing e.g. a Reorg event.
Suggestion:
| match serde_json::from_str(&payload) { | |
| Ok(event) => Some((id, event)), | |
| Err(err) => { | |
| logging::log::warn!("Skipping undecodable stream event #{id}: {err}"); | |
| None | |
| } | |
| } | |
| match serde_json::from_str(&payload) { | |
| Ok(event) => Some((id, event)), | |
| Err(err) => { | |
| logging::log::error!("Undecodable stream event #{id}: {err}"); | |
| return Err(ApiServerStorageError::LowLevelStorageError(format!( | |
| "Undecodable stream event #{id}: {err}" | |
| ))); | |
| } | |
| } |
| /// Delete the stream events that fell out of the retention window, returning the number of | ||
| /// deleted rows. | ||
| pub async fn prune_stream_events(&mut self) -> Result<u64, ApiServerStorageError> { | ||
| let deleted = self | ||
| .tx | ||
| .execute( | ||
| "DELETE FROM ml.emitted_events | ||
| WHERE id <= (SELECT COALESCE(max(id), 0) - $1 FROM ml.emitted_events);", | ||
| &[&STREAM_EVENTS_RETENTION_COUNT], | ||
| ) |
There was a problem hiding this comment.
Retention pruning can silently destroy unread events. prune_stream_events deletes everything older than the most recent 10,000 events, and the scanner calls it on every block-indexing commit. Meanwhile the pump (run_event_pump) drains in batches of 1000 and, on a read error, only resumes after wait_for_wakeup (up to the 30s poll interval). If the pump stalls — repeated DB errors, a restart of the web server, a slow drain during a long reorg — more than 10,000 events can accumulate and be pruned before last_seen_id reaches them, so those events are permanently lost. Unlike per-subscriber lag (which emits the lag advisory in stream.rs), this loss is invisible to clients: they simply never see Block/Reorg events, with no gap signaled. Consider (a) pruning only events with id <= last_seen_id known to have been consumed, or (b) at minimum logging/broadcasting an advisory when the pump detects that events between reads have disappeared (e.g. next read returns ids with a jump past the retention window).
Suggestion:
| /// Delete the stream events that fell out of the retention window, returning the number of | |
| /// deleted rows. | |
| pub async fn prune_stream_events(&mut self) -> Result<u64, ApiServerStorageError> { | |
| let deleted = self | |
| .tx | |
| .execute( | |
| "DELETE FROM ml.emitted_events | |
| WHERE id <= (SELECT COALESCE(max(id), 0) - $1 FROM ml.emitted_events);", | |
| &[&STREAM_EVENTS_RETENTION_COUNT], | |
| ) | |
| /// Delete the stream events that fell out of the retention window, returning the number of | |
| /// deleted rows. | |
| /// TODO: only prune events already consumed by the event pump, or emit a gap advisory, | |
| /// otherwise a stalled pump silently loses events to pruning. | |
| pub async fn prune_stream_events(&mut self) -> Result<u64, ApiServerStorageError> { | |
| let deleted = self | |
| .tx | |
| .execute( | |
| "DELETE FROM ml.emitted_events | |
| WHERE id <= (SELECT COALESCE(max(id), 0) - $1 FROM ml.emitted_events);", | |
| &[&STREAM_EVENTS_RETENTION_COUNT], | |
| ) |
| if last_event_id > 0 { | ||
| db_tx.prune_stream_events().await?; | ||
| db_tx.notify_new_stream_events(last_event_id).await?; | ||
| } |
There was a problem hiding this comment.
Pruning happens unconditionally at the start of the same transaction that just appended the new events, and the prune cutoff is max(id) - 10000 where max(id) already includes the just-appended events. If the pump ever falls more than 10,000 events behind (e.g., web server down for a while while the scanner keeps syncing), each new block commits a prune that deletes events the pump has not yet read. Combined with the pump resuming from last_seen_id, this is fine for the pump itself, but any pump that later reconnects via initial_last_seen_id (the latest committed event id) plus a client trying to resume will silently skip the deleted ids. If silent loss across a >10k backlog is intended (no replay semantics), consider pruning based on a cutoff captured before the appends, or moving the prune to a periodic/background path so appends and pruning are decoupled.
Suggestion:
| if last_event_id > 0 { | |
| db_tx.prune_stream_events().await?; | |
| db_tx.notify_new_stream_events(last_event_id).await?; | |
| } | |
| if last_event_id > 0 { | |
| // Prune against the high-water mark from before this batch so that events appended | |
| // here can never evict each other; ideally move this to a periodic maintenance path. | |
| db_tx.prune_stream_events().await?; | |
| db_tx.notify_new_stream_events(last_event_id).await?; | |
| } |
| // Note: backends without stream event support return an empty list. | ||
| assert!( | ||
| read_events.is_empty() || read_events.len() == events.len(), | ||
| "unexpected number of stream events: {}", | ||
| read_events.len() | ||
| ); |
There was a problem hiding this comment.
This assertion accepts either an empty result or the full event list, so any backend that silently drops appended stream events (a real correctness bug) passes the test. The rollback phase (Phase 1) also passes vacuously for such backends, so the transactional-visibility guarantee this test documents is not actually verified for them. Consider having backends declare stream-event support (e.g. a capability flag on the storage trait or a per-backend test parameter) and asserting strictly (non-empty, correct order, correct payloads) when support is declared.
Suggestion:
| // Note: backends without stream event support return an empty list. | |
| assert!( | |
| read_events.is_empty() || read_events.len() == events.len(), | |
| "unexpected number of stream events: {}", | |
| read_events.len() | |
| ); | |
| // Only backends that declare stream-event support must return the events. | |
| if supports_stream_events::<S>() { | |
| assert_eq!(read_events.len(), events.len(), "unexpected number of stream events"); | |
| } else { | |
| assert!(read_events.is_empty()); | |
| } |
| let command = if podman_on_path { "podman" } else { "docker" }; | ||
| logging::log::info!("Using {command} as the container manager"); | ||
| command |
There was a problem hiding this comment.
container_command() is re-invoked on every command (run, get_port_mapping, stop, restart, print_logs, remove_container), re-probing PATH each time; the result could change between calls (e.g. PATH mutation or a podman binary appearing mid-test), causing commands from different managers to be mixed against the same container. It also never verifies that docker exists when podman is absent, so the failure message would point at a missing docker binary. Consider resolving the command once (e.g. store it in Podman at construction) and failing early with a clear message if neither tool is found.
Suggestion:
| let command = if podman_on_path { "podman" } else { "docker" }; | |
| logging::log::info!("Using {command} as the container manager"); | |
| command | |
| // Resolve once per container instance and fail fast if no container manager is available. | |
| let command = if podman_on_path { | |
| "podman" | |
| } else if docker_on_path { | |
| "docker" | |
| } else { | |
| panic!("Neither `podman` nor `docker` was found on PATH; containerized tests cannot run"); | |
| }; | |
| logging::log::info!("Using {command} as the container manager"); | |
| command |
| .route("/order/pair/:pair", get(order_pair)) | ||
| // Note: the real-time event stream is exposed together with the v2 endpoints, since the | ||
| // events reference data that is served by them. | ||
| .route("/stream", get(super::stream::stream_events)) |
There was a problem hiding this comment.
Unlike the other v2 endpoints, this route holds per-connection state (a broadcast receiver and a stream task) for an unbounded lifetime, and there is no connection limit or timeout. A modest number of idle/slow clients can accumulate indefinitely since keepalives prevent proxy timeouts. Consider bounding concurrent stream connections or applying a maximum connection lifetime, unless the reverse proxy in front of the server is documented to provide this.
Suggestion:
| .route("/stream", get(super::stream::stream_events)) | |
| .route("/stream", get(super::stream::stream_events)) | |
| // TODO: consider limiting concurrent stream connections / connection lifetime. |
| if let Err(err) = handle.await { | ||
| logging::log::error!("CRITICAL: the {name} task terminated: {err}"); | ||
| } |
There was a problem hiding this comment.
The supervisor only logs a CRITICAL error when the event pump or mempool bridge task terminates; the process keeps serving REST endpoints while the event stream is silently dead, and connected SSE clients just see keepalives forever. Since a terminated task here is unrecoverable by design (no restart logic exists), consider escalating the failure so it is operationally observable — e.g. exit the process (letting the service manager restart it) or surface it via the /server_status health endpoint.
Suggestion:
| if let Err(err) = handle.await { | |
| logging::log::error!("CRITICAL: the {name} task terminated: {err}"); | |
| } | |
| if let Err(err) = handle.await { | |
| logging::log::error!("CRITICAL: the {name} task terminated: {err}"); | |
| // The streaming tasks are expected to run for the process lifetime; a termination | |
| // leaves the event stream silently dead, so fail fast and let the supervisor restart. | |
| std::process::exit(1); | |
| } |
…he event stream - retention pruning is now based on the consumption progress recorded by the event pump, so unread events are never deleted; a hard limit of 10x the retention window bounds the log during a pump outage (logged loudly) - pruning runs once per retention window in the scanner instead of on every block-indexing commit - undecodable stream event rows are a hard error instead of a silent skip - initial_last_seen_id retries iteratively with exponential backoff instead of unbounded recursion - a terminated streaming background task brings the process down instead of leaving the event stream silently dead - streaming CLI options are validated with clap range checks instead of silent clamping; new --stream-events-max-subscribers option bounds the concurrent SSE connections (429 when exceeded) - the mempool bridge subscription is bounded by a timeout - tests: strict stream-event assertions per backend capability, container manager resolved once with fail-fast, deterministic sentinel settle and web task join in the postgres stream test
| if self.reconnect().await.is_err() { | ||
| tokio::time::sleep(self.poll_interval).await; | ||
| } |
There was a problem hiding this comment.
The reconnect error is swallowed when reconnection fails: reconnect() returns an Err that is converted into a bare sleep with no logging. Once the listener connection drops, a prolonged database outage or connection failure means the pump silently degrades to pure polling forever with no diagnostic trace, making production incidents (e.g. misconfigured TLS, exhausted connection slots) very hard to diagnose. Log the error (and possibly the retry delay) before falling back to the sleep.
Suggestion:
| if self.reconnect().await.is_err() { | |
| tokio::time::sleep(self.poll_interval).await; | |
| } | |
| if let Err(err) = self.reconnect().await { | |
| logging::log::error!( | |
| "Failed to re-establish the stream event listener connection: {err}; \ | |
| falling back to polling" | |
| ); | |
| tokio::time::sleep(self.poll_interval).await; | |
| } |
| // Note: without a recorded cursor, the plain retention window is used, so that a | ||
| // database fed only by the scanner (no web server) is still bounded. | ||
| let retention_cutoff = max_id - STREAM_EVENTS_RETENTION_COUNT; |
There was a problem hiding this comment.
When no pump cursor is recorded yet, pruning falls back to the plain retention window, which can delete events the pump has never read. There is a startup race even in a normal deployment: the scanner can commit and prune events before the web server's pump has finished initial_last_seen_id() and written its first cursor, so events committed in that window are permanently lost to stream clients (a missed Reorg event). Consider only pruning below a recorded cursor, and bounding growth in the no-pump case differently (e.g. a time-based cutoff, or recording a cursor from the scanner side too).
| pub async fn run_event_pump( | ||
| mut source: impl StreamEventSource, | ||
| channel: StreamEventsChannel, | ||
| mut last_seen_id: StreamEventId, | ||
| ) { | ||
| logging::log::info!("Streaming events after #{last_seen_id}"); | ||
| loop { |
There was a problem hiding this comment.
run_event_pump has no shutdown or cancellation path: the loop never returns, and initial_last_seen_id() (see PostgresStreamEventSource) can block indefinitely during a database outage. A spawned pump task therefore cannot be terminated gracefully on server shutdown, and during an outage nothing bounds or reports the startup stall other than repeated error logs. Consider making wait_for_wakeup return a Result/Option for termination, or driving the pump from a CancellationToken/select on a shutdown signal, and bounding initial_last_seen_id with cancellation.
| let subscription = state.stream_events.try_subscribe().ok_or( | ||
| ApiServerWebServerError::TooManyStreamConnections, | ||
| )?; |
There was a problem hiding this comment.
max_subscribers is enforced via a semaphore permit held inside the SSE stream state. Releasing it relies on axum dropping the response body (and thus the unfold state) on client disconnect; there is no test covering slot release after disconnection or 429 after exhausting the limit, so a subtle leak in the response lifecycle would permanently reduce capacity. Consider adding an integration test that opens max_subscribers connections, closes one, and asserts the next subscribe succeeds.
| streaming, | ||
| }; | ||
| use clap::Parser; |
There was a problem hiding this comment.
The streaming constants live in the library (api_web_server::streaming) but the binary declares its own mod config whose crate::streaming path only resolves because main.rs re-imports the library module under the binary's crate root (use api_web_server::streaming). This is fragile: removing that single import breaks the binary's config.rs in a non-obvious way. Consider defining the constants in one shared crate (e.g. api_server_common::streaming, where the rest of the streaming machinery already lives) and referencing them explicitly, or re-exporting via pub use.
Suggestion:
| streaming, | |
| }; | |
| use clap::Parser; | |
| streaming, | |
| }; | |
| use clap::Parser; | |
| // Consider instead: use api_server_common::streaming; and keep config.rs referencing it directly. |
| logging::log::error!("CRITICAL: the {name} task terminated: {err}"); | ||
| std::process::exit(1); |
There was a problem hiding this comment.
std::process::exit(1) skips destructors and aborts in-flight work (e.g. connection-pool shutdown, in-flight requests). If a supervised task fails during process shutdown, this races with orderly termination and can turn a graceful stop into a failure exit code for the service manager. Since the tasks are tokio tasks that end with an error on the JoinHandle, consider signalling a shutdown channel (or at least tokio::task::abort of siblings plus letting the runtime return non-zero) instead of hard-exiting from a detached spawned task.
| let mut subscription = subscription; | ||
| while let Some(event) = subscription.next().await { |
There was a problem hiding this comment.
The subscribe timeout only protects the WebSocket/RPC handshake; once subscribed, the subscription.next().await loop has no timeout, ping/pong watchdog, or stalled-connection detection. If the node connection stalls (TCP alive but no frames and no close), this loop blocks forever with no log output, silently disabling mempool TxSeen events while the supervisor sees a healthy task. jsonrpsee subscriptions do not impose a receive timeout, so consider wrapping each next() in a bounded timeout and treating expiry as a connection failure that triggers re-subscription.
Suggestion:
| let mut subscription = subscription; | |
| while let Some(event) = subscription.next().await { | |
| let mut subscription = subscription; | |
| loop { | |
| let event = match tokio::time::timeout(MEMPOOL_EVENT_TIMEOUT, subscription.next()).await { | |
| Ok(event) => event, | |
| Err(_elapsed) => { | |
| logging::log::error!("No mempool event traffic for {MEMPOOL_EVENT_TIMEOUT:?}; re-subscribing"); | |
| break; | |
| } | |
| }; |
| } | ||
| } | ||
| } | ||
| logging::log::warn!("Node mempool subscription closed; re-subscribing"); |
There was a problem hiding this comment.
During the re-subscription window (connection loss plus the backoff delay, which grows to 60s), every mempool transaction is silently lost: no TxSeen event, no lag advisory, and only a single warn log. Unlike the database pump, this bridge has no poll safety net, so long node outages produce an undetectable gap in the stream. At minimum, emit a diagnostic (e.g. a lag advisory event via the channel) when a subscription is lost, and consider logging the outage duration when re-subscribing.
Suggestion:
| logging::log::warn!("Node mempool subscription closed; re-subscribing"); | |
| logging::log::warn!("Node mempool subscription closed; re-subscribing"); | |
| // Note: tell connected clients that a gap in `TxSeen` events is possible. | |
| let _ = handle.channel.send(StreamEvent::Lag { skipped: 0 }); |
| Err(err) => { | ||
| logging::log::error!("Undecodable stream event #{id}: {err}"); | ||
| return Err(ApiServerStorageError::DeserializationError(format!( | ||
| "Undecodable stream event #{id}: {err}" | ||
| ))); | ||
| } |
There was a problem hiding this comment.
An undecodable event row makes this function return a hard error, and in run_event_pump a read error just breaks the inner loop and waits for the next wakeup, which re-reads and re-fails on the same row indefinitely. The pump is permanently stalled at that id: no client receives any events (including reorg notices) until retention hard-limit pruning eventually deletes the bad row — which itself silently drops all the events in between. Consider a bounded dead-letter mechanism (e.g. log the payload, skip the id after N retries) or at least a metric/alert so operators notice before the hard limit silently prunes unread events.
| /// Subscribe to the stream events. Each subscriber receives the events sent after the | ||
| /// subscription has been created. | ||
| pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<StreamEvent> { | ||
| self.sender.subscribe() | ||
| } |
There was a problem hiding this comment.
The broadcast channel silently drops events for a slow subscriber once its capacity is exceeded (recv() returns Err(Lagged)), and nothing here documents or surfaces that. A lagging SSE client will simply miss events (e.g. a Reorg) with no log, while the pump happily advances last_seen_id. Consider at least documenting this per-subscriber loss semantics (the event log has no replay), or logging a warning when subscribe/send observes lag.
Suggestion:
| /// Subscribe to the stream events. Each subscriber receives the events sent after the | |
| /// subscription has been created. | |
| pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<StreamEvent> { | |
| self.sender.subscribe() | |
| } | |
| /// Subscribe to the stream events. Each subscriber receives the events sent after the | |
| /// subscription has been created. | |
| /// | |
| /// Note: a subscriber that falls more than `capacity` events behind receives a | |
| /// `broadcast::error::RecvError::Lagged` and the skipped events are lost permanently; | |
| /// the stream has no replay semantics. | |
| pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<StreamEvent> { | |
| self.sender.subscribe() | |
| } |
| let subscription = state | ||
| .stream_events | ||
| .try_subscribe() | ||
| .ok_or(ApiServerWebServerError::TooManyStreamConnections)?; |
There was a problem hiding this comment.
A subscriber slot (the OwnedSemaphorePermit) is only released when the response stream is dropped, and there is no idle/inactivity timeout for stream connections. A client that stalls without disconnecting (e.g. a hung NAT or a half-open TCP connection) holds its slot indefinitely; once max_subscribers permits are held this way, all new stream clients get 429 Too Many Stream Connections until the stalled connections are dropped by the OS/proxy. The keepalive writes will eventually surface write errors on a truly dead socket, but a client that merely stops reading may persist for a long time. Consider documenting this operational boundary or adding a connection-level idle timeout.
…e event stream - the listener reconnection failure is logged before falling back to pure polling, so a dead listener connection is diagnosable in production - retention pruning without a recorded pump cursor now only enforces the hard limit, closing the startup race where fresh events could be pruned before the pump recorded any progress - an undecodable event row no longer stalls the pump forever: the row is logged with its payload, skipped (dead-letter style), and the clients are told about the gap through a lag advisory - a lag advisory event is now a proper stream event type; the mempool bridge broadcasts it when the node subscription is lost and logs the outage duration when it re-subscribes - every mempool subscription receive is bounded by a 10-minute timeout, so a stalled connection cannot wedge the bridge silently - the streaming defaults moved to api_server_common::streaming, so the shared config module no longer depends on the binary's crate-root re-import; documented the pump/process termination semantics and the broadcast lag semantics - tests: subscriber limit (429) and slot release after disconnect
| let _ = channel.send(StreamEvent::Lag { skipped: 1 }); | ||
| last_seen_id = id; | ||
| } |
There was a problem hiding this comment.
When an undecodable event is skipped, last_seen_id advances past the row but update_consumed_id is not called for it. If no further events ever arrive (the stream goes idle), the DB pump cursor stays permanently behind the actually-consumed id: retention pruning will then hold back rows up to the retention window, and once the hard limit kicks in, unconsumed-by-the-clients rows get deleted with only an error log. Consider calling source.update_consumed_id(last_seen_id) on the skip path (and/or when a drained batch ends with batch_size == 0 after a skip) so the recorded cursor reflects the truly consumed position.
Suggestion:
| let _ = channel.send(StreamEvent::Lag { skipped: 1 }); | |
| last_seen_id = id; | |
| } | |
| let _ = channel.send(StreamEvent::Lag { skipped: 1 }); | |
| last_seen_id = id; | |
| // Note: record the skip so the retention pruning does not wait for an id the | |
| // pump has already consumed. | |
| source.update_consumed_id(last_seen_id).await; | |
| } |
| if let Some(block_id) = db_tx.get_main_chain_block_id(BlockHeight::new(height)).await? { | ||
| block_ids.push(block_id); | ||
| } |
There was a problem hiding this comment.
A None here for a height in (common_block_height, best_block_height] means the main-chain index has a hole even though get_best_block() reported a higher tip — an integrity violation, not a normal state. Silently skipping it produces a Reorg event with an incomplete removed_block_ids, so stream consumers will keep stale data for the missing blocks. Either return an error (aborting the still-uncommitted transaction) or at minimum log loudly so the anomaly is observable.
Suggestion:
| if let Some(block_id) = db_tx.get_main_chain_block_id(BlockHeight::new(height)).await? { | |
| block_ids.push(block_id); | |
| } | |
| let block_id = db_tx | |
| .get_main_chain_block_id(BlockHeight::new(height)) | |
| .await? | |
| .ok_or_else(|| { | |
| ApiServerStorageError::LowLevelStorageError(format!( | |
| "Main chain block missing at height {height} during reorg capture" | |
| )) | |
| })?; | |
| block_ids.push(block_id); |
| let deadline = tokio::time::Instant::now() + Duration::from_secs(1); | ||
| loop { | ||
| let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); | ||
| match tokio::time::timeout(remaining, sse.next_frame(FRAME_TIMEOUT)).await { |
There was a problem hiding this comment.
The negative-assertion window is not honored precisely: the loop computes remaining from a 1-second deadline, but then calls sse.next_frame(FRAME_TIMEOUT) with the full 5-second constant instead of remaining. On the last iteration the test can over-wait up to ~5s past the intended observation window (it still won't report a false failure, but it slows the suite and slightly weakens the negative-assertion contract). Pass the remaining time instead.
Suggestion:
| let deadline = tokio::time::Instant::now() + Duration::from_secs(1); | |
| loop { | |
| let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); | |
| match tokio::time::timeout(remaining, sse.next_frame(FRAME_TIMEOUT)).await { | |
| let deadline = tokio::time::Instant::now() + Duration::from_secs(1); | |
| loop { | |
| let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); | |
| match tokio::time::timeout(remaining, sse.next_frame(remaining)).await { |
| fn allows(&self, event: &StreamEvent) -> bool { | ||
| self.0.contains(&event.event_type()) | ||
| } |
There was a problem hiding this comment.
The lag advisory is subject to the same event-type filter as regular events. A client that subscribes with types=block will never receive lag events, yet the database event pump broadcasts StreamEvent::Lag when it has to skip an undecodable row — which can be a block event. In that case the filtered client silently misses blocks with no advisory, defeating the purpose of the gap signal. Consider always letting Lag events through the filter (they carry no data payload anyway), or documenting that the lag event must be explicitly subscribed to.
Suggestion:
| fn allows(&self, event: &StreamEvent) -> bool { | |
| self.0.contains(&event.event_type()) | |
| } | |
| fn allows(&self, event: &StreamEvent) -> bool { | |
| // The `lag` advisory is a control event relevant to every subscriber, so it bypasses | |
| // the type filter. | |
| event.event_type() == StreamEventType::Lag || self.0.contains(&event.event_type()) | |
| } |
| outage_started = Some(std::time::Instant::now()); | ||
| let _ = handle.channel.send(StreamEvent::Lag { skipped: 0 }); |
There was a problem hiding this comment.
run_mempool_bridge broadcasts StreamEvent::Lag { skipped: 0 } on every connection loss, but the actual number of transactions seen during the outage is never tracked — skipped is always 0 here, so the payload carries no information. Consider counting (even approximately, e.g. events observed just before the failure) or documenting that this variant of the advisory means 'gap possible, count unknown'. Also note that after re-subscription, a node that replays recent TransactionProcessed events would produce duplicate TxSeen events; if the mempool RPC has such replay semantics, clients have no way to distinguish replays from new events.
2 participants
Summary
Adds a real-time event stream to the api-server so block explorers can display transactions and blocks live. New endpoint:
GET /api/v2/stream(Server-Sent Events), delivering three event kinds as named SSE events:tx_seen{tx_id, origin: local|remote}block{block_id, height, timestamp, tx_ids}reorg{common_ancestor_height, removed_block_ids, new_tip_height}Payloads are kept small; clients hydrate details through the existing REST endpoints.
Design
Block/reorg events — scanner, transactionally consistent.
BlockchainState::scan_blocksappends events to a newml.emitted_eventstable inside the same RW transaction that indexes the blocks (reorg detection: local best height > common ancestor, with removed block ids captured before the disconnect marks them). After the appends, the scanner issuespg_notify('mintlayer_events', <last id>)inside the transaction, so Postgres delivers the wakeup only on commit: once ablockevent is delivered, the referenced block is immediately queryable via REST (asserted by tests).tx_seen events — web server. The web server bridges the node's mempool events from the existing
NodeRpcClientWebSocket intoTxSeenevents (onlysuccessful: truetransactions; subscription re-established on connection loss).Web server event pump. A background pump wakes on the Postgres notification (dedicated LISTEN connection — with automatic reconnection bounded by the poll interval, since Postgres only delivers notifications on the listening connection) and forwards events into a
tokio::sync::broadcastchannel. Periodic polling (default 30 s) is the safety net for missed notifications; the backlog is drained in bounded batches (1000/batch) and the cursor (last_seen_id) only ever moves forward, so a web-server restart cannot produce duplicate deliveries. Events older than the last 10 000 are pruned by the scanner (the stream has no replay, so they are never re-read).Endpoint. Named events for
EventSource.addEventListener, optional?types=block,reorgfilter (400 on invalid values),: keepalivecomments (default 30 s), spec-compliantretry:hint as the first frame,x-accel-buffering: nofor reverse proxies, and alagadvisory ({"skipped": n}) when a client exceeds the broadcast capacity. No auth, consistent with the other v2 GET endpoints. No replay/Last-Event-ID: clients recover missed events via REST.Operational notes
ml.emitted_eventstable). Per the existing convention this triggers the usual full database re-initialization on upgrade (full resync). Required regardless: without the new table the scanner would fail on insert.--stream-events-broadcast-capacity(1024),--stream-events-poll-interval-secs(30),--stream-events-keepalive-interval-secs(30).Testing
storage-test-suite(both backends; Postgres via container): events visible only after commit, rollback discards them, ascending monotonic ids, resume-after-last-seen, no-op behavior of the in-memory backend.tx_seenmapping (filterssuccessful: falseandNewTip), filter parsing/matching.GET /v2/block/:idconsistency, and a Postgres end-to-end test driving the real scanner →ml.emitted_events→ LISTEN/NOTIFY pump → SSE, including a forced reorg (exactly onereorgevent with correct removed ids/heights, followed by the new fork's block events, no duplicates), and a dedup check../do_checks.shclean (fmt, cargo-deny, cargo-vet, clippy, codecheck);cargo test --releasegreen for every touched crate (stack tests incl. containers re-run after the rebase onto current master).Review
Ran two rounds of security review (findings fixed: event-pump hot-spin after listener death → reconnect + sleep fallback; unbounded startup reads → batched reads; poison-row stall → skip undecodable rows; config-value panics → clamped; dead
retryheader → in-stream frame; event-name duplication → singleStreamEventTypesource of truth) and a code-quality review (both blockers and warnings addressed; DRY pass on shared types/test helpers).Notes for reviewers
fix: address clippy 1.98 lints across the workspace) is unrelated drive-by work:do_checks.shfails on current master with clippy 1.98's new lints; the fixes are mechanical and behavior-preserving (reviewers can skip it). Happy to split it into its own PR if preferred.CURRENT_STORAGE_VERSIONbump means existing deployments resync on upgrade — called out in the README/CHANGELOG.tx_seenis not covered end-to-end against a real node process (the repo has no harness that spins a node RPC server in tests); the bridge mapping and the SSE delivery path are covered separately.storage-test-suite/src/podman.rs) now falls back todockerwhenpodmanis not installed (identical CLI surface for the commands used).