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 27 issue(s) in this PR.
📄
|
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| 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 undecodable rows is a reasonable availability trade-off, but the pump advances last_seen_id past these ids, so subscribers permanently and silently miss those events (e.g. after a payload schema change between versions). Since StreamEvent has no version field, consider adding one, or at least surfacing skipped ids so the pump can emit a gap/lag advisory to subscribers rather than only a server-side warning.
| /// Delete the stream events that fell out of the retention window. | ||
| pub async fn prune_stream_events(&mut self) -> Result<(), ApiServerStorageError> { |
There was a problem hiding this comment.
Retention pruning is independent of consumer progress: if the event pump is stalled or its reads repeatedly fail (run_event_pump just logs the error and falls back to waiting), events can fall out of the retention window and be deleted before they are ever forwarded to subscribers. Consider tracking the minimum forwarded id (e.g. persisted by the pump) and pruning only below it, or at minimum documenting that events may be lost if the pump falls more than STREAM_EVENTS_RETENTION_COUNT behind.
Suggestion:
| /// Delete the stream events that fell out of the retention window. | |
| pub async fn prune_stream_events(&mut self) -> Result<(), ApiServerStorageError> { | |
| /// Delete the stream events that fell out of the retention window. | |
| /// | |
| /// Note: pruning is based purely on event age relative to the newest event; it does | |
| /// not take consumer progress into account, so events the pump has not yet forwarded | |
| /// can be permanently lost if the pump falls more than RETENTION_COUNT behind. | |
| pub async fn prune_stream_events(&mut self) -> Result<(), ApiServerStorageError> { |
| Err(err) => { | ||
| logging::log::error!("{err}"); | ||
| break; | ||
| } |
There was a problem hiding this comment.
After a read error the pump breaks out of the drain loop and waits for the next wakeup, which is fine, but combined with retention pruning any events not read during the failure window are eventually deleted server-side and never reach subscribers. Consider retrying the read (with backoff) a bounded number of times before falling back to the wakeup interval, or at least documenting that a read error can lead to permanent event loss due to pruning.
| 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.
prune_stream_events() uses a fixed-count retention window (deletes everything below max(id) - 10_000) with no consumer watermark. Since events are pruned on every connect_block call, a subscriber whose pump is offline, lagging, or reading slower than block production can have events deleted before it reads them; the pump then silently continues from the next surviving id, producing undetectable gaps in the event stream (e.g. a missed Reorg event leaves the client with stale, disconnected block references). Consider tracking a consumer-side last-read watermark for pruning, or at least surfacing gap detection (id discontinuity) in the event pump when a pruned range is skipped.
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 { | |
| // Note: pruning must be tied to consumer progress, not a fixed count, otherwise | |
| // lagging subscribers lose events silently. | |
| db_tx.prune_stream_events().await?; | |
| db_tx.notify_new_stream_events(last_event_id).await?; | |
| } |
| for height in (common_block_height.into_int() + 1)..=best_block_height.into_int() { | ||
| 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.
capture_main_chain_block_ids issues one get_main_chain_block_id round trip per height, so a deep reorg holds the read-write storage transaction (and the write lock) for O(depth) queries. Consider a batched range query (e.g. SELECT ... WHERE height BETWEEN $1 AND $2 ORDER BY height) to collapse this into a single round trip and shorten the lock hold time.
| Some(types) => StreamEventsFilter::parse(types).map_err(|_| { | ||
| ApiServerWebServerError::ClientError(ApiServerWebServerClientError::BadRequest) | ||
| })?, |
There was a problem hiding this comment.
The parse error is discarded and mapped to a generic BadRequest, yet StreamEventTypeParseError implements a helpful Display listing the valid event types. Clients get no indication of which types value was invalid. Since this is a client-facing input error, surfacing the message (or at least logging it) would substantially improve the developer experience of API consumers.
Suggestion:
| Some(types) => StreamEventsFilter::parse(types).map_err(|_| { | |
| ApiServerWebServerError::ClientError(ApiServerWebServerClientError::BadRequest) | |
| })?, | |
| Some(types) => StreamEventsFilter::parse(types).map_err(|err| { | |
| logging::log::debug!("Invalid stream `types` query parameter: {err}"); | |
| ApiServerWebServerError::ClientError(ApiServerWebServerClientError::BadRequest) | |
| })?, |
| Err(broadcast::error::RecvError::Lagged(skipped)) => { | ||
| // Note: the client fell too far behind; tell it what happened and | ||
| // continue with the fresh events. | ||
| return Some((Ok(lag_event(skipped)), (receiver, filter, retry_sent))); | ||
| } |
There was a problem hiding this comment.
On broadcast lag the client is only told how many events were skipped via the lag advisory, but those skipped events may include Reorg or Block events, leaving the client with an inconsistent view of the chain (e.g. it never learns a reorg happened). The shared streaming.rs explicitly states events are persisted with retention (STREAM_EVENTS_RETENTION_COUNT) and the endpoint "does not support replays anyway" — the endpoint should at least document the expected client recovery contract (resync via REST after a lag event), otherwise consider actually replaying the missed ids from storage.
Suggestion:
| Err(broadcast::error::RecvError::Lagged(skipped)) => { | |
| // Note: the client fell too far behind; tell it what happened and | |
| // continue with the fresh events. | |
| return Some((Ok(lag_event(skipped)), (receiver, filter, retry_sent))); | |
| } | |
| Err(broadcast::error::RecvError::Lagged(skipped)) => { | |
| // Note: the client fell too far behind; tell it what happened and | |
| // continue with the fresh events. Document that clients must resync | |
| // through the REST endpoints after receiving a `lag` event. | |
| return Some((Ok(lag_event(skipped)), (receiver, filter, retry_sent))); | |
| } |
| tokio::spawn(streaming::run_database_event_pump( | ||
| event_source, | ||
| stream_events.clone(), | ||
| )); |
There was a problem hiding this comment.
The database event pump and the mempool bridge are spawned before the RPC client is constructed and before web_server starts. If construction subsequently fails (e.g. RpcError), the server exits while these tasks keep running against a half-initialized state (the pump holds a dedicated Postgres LISTEN connection that is never shut down). There is also no graceful shutdown wiring for the event listener on normal exit. Consider spawning these tasks only after all fallible setup succeeds, and passing them a shutdown signal (e.g. CancellationToken) tied to server shutdown.
| 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 database event pump starts at last_seen_id = 0, so on every server restart it replays up to STREAM_EVENTS_RETENTION_COUNT (10,000) historical events into the broadcast channel. Any client that connects while this backlog is being drained receives stale Block/TxSeen events interleaved with live ones and has no way to distinguish or deduplicate them, because the SSE frames carry no id: field (and the endpoint doesn't honor Last-Event-ID). Consider starting the pump from the latest stored event id (skipping the backlog, since the endpoint explicitly does not support replays), or attaching SSE event ids so clients can detect duplicates.
Suggestion:
| pub async fn run_database_event_pump(source: impl StreamEventSource, handle: StreamEventsHandle) { | |
| run_event_pump(source, handle.channel, 0).await; | |
| } | |
| pub async fn run_database_event_pump(source: impl StreamEventSource, handle: StreamEventsHandle) { | |
| // Start from the newest persisted event so that a server restart does not replay the | |
| // retained backlog to connected clients; the endpoint does not support replays. | |
| 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.
run_mempool_bridge retries the WebSocket subscription forever with a fixed 1s delay, no backoff, and no cancellation/shutdown propagation. If the node is persistently unreachable (or auth is misconfigured), this hot-loops log spam and reconnect attempts indefinitely. Additionally, the task is spawned in main.rs with the JoinHandle dropped, so a permanently failed bridge is never observed — clients silently stop receiving TxSeen events with no health signal. Consider exponential backoff with a cap, and a shutdown token (or at minimum a fatal-error exit) so the failure is visible.
Suggestion:
| tokio::time::sleep(MEMPOOL_RESUBSCRIBE_DELAY).await; | |
| } | |
| } | |
| // Exponential backoff capped at some maximum, plus cancellation via a shutdown token. | |
| tokio::time::sleep(MEMPOOL_RESUBSCRIBE_DELAY).await; | |
| } | |
| } |
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).