feat(api-server): real-time SSE event stream for block explorers

accounting/src/delta/delta_data_collection/mod.rs

@@ -158,6 +158,12 @@ impl<K: Ord + Copy, T: Clone> FromIterator<(K, DataDelta<T>)> for DeltaDataColle
     }
 }
 
+impl<K: Ord + Copy, T: Clone + Eq> Default for DeltaDataCollection<K, T> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 /// Given two deltas, combine them into one delta, this is the basic delta data composability function
 fn combine_delta_data<T: Clone + Eq>(
     lhs: DataDelta<T>,

accounting/src/delta/delta_data_collection/undo.rs

@@ -61,3 +61,9 @@ impl<K: Ord, T: Clone> DeltaDataUndoCollection<K, T> {
         self.data
     }
 }
+
+impl<K: Ord, T: Clone> Default for DeltaDataUndoCollection<K, T> {
+    fn default() -> Self {
+        Self::new()
+    }
+}

api-server/CHANGELOG.md

@@ -6,6 +6,16 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/
 
 ## [Unreleased]
 
+### Added
+- New endpoint `/v2/stream` (Server-Sent Events) that streams `tx_seen`, `block` and `reorg` events in real time, with an optional `types` filter parameter.\
+  The stream carries keepalive comments, an in-stream reconnection hint, and a `lag` advisory for clients that fall behind; there is no replay, so missed events must be recovered through the regular REST endpoints.
+- New web server options: `--stream-events-broadcast-capacity`, `--stream-events-poll-interval-secs`, `--stream-events-keepalive-interval-secs`.
+- The scanner now emits block/reorg events into a new `ml.emitted_events` table (transactionally consistent with the indexed data) and notifies listeners on commit.\
+  Transactions seen in the node's mempool are bridged into `tx_seen` events by the web server.
+
+### Changed
+- The api-server storage version was bumped from 25 to 26 (new `ml.emitted_events` table); the scanner re-initializes the database when it finds a different version, as before. Full resync is required.
+
 ### Fixed
 - `/v2/token` and `/v2/token/ticker/{ticker}` no longer return the same id more than once.\
   Both tokens and NFTs are stored with a row per block height they changed at, and every one

api-server/README.md

@@ -20,12 +20,76 @@ The blockchain scanner daemon is a tool that runs in the backend, scans the bloc
 
 The architecture of the API server is made to be distributed as much as desired. You can run the database on as many servers as you wish in master-slave mode. This is achieved by separating the "API web server" from the "blockchain scanner daemon". You can have a single "blockchain scanner daemon", communicating with the `node-daemon` of Mintlayer, collecting information about new blocks and writing it to the master database, while having as many instances of the API web server reading from the slave databases. This ensures virtually an infinitely scalable infrastructure.
 
+#### Real-time event streaming
+
+The scanner daemon and the API web server also cooperate to provide a real-time event stream for block explorer clients. When the scanner indexes new blocks, it writes the corresponding events into the `ml.emitted_events` table *inside the same database transaction* that performs the indexing, and a notification is delivered to listeners when the transaction commits. The API web server runs an "event pump" that listens for these notifications on a dedicated database connection (with automatic reconnection and a periodic polling fallback) and forwards the events to connected clients. Independently of this, transactions reaching the node's mempool are bridged from the node's WebSocket RPC and delivered as `tx_seen` events immediately, before the block containing them is indexed. The key guarantee for clients: once a `block` event is delivered, the referenced block and its transactions are immediately queryable through the regular REST endpoints; there is no race between the event and the data it refers to.
+
 ### Using other database infrastructures
 
 Currently, the API server uses PostgreSQL for storage, but the design is extremely flexible and any desired database can be added if needed by implementing some interface (trait) in the rust code.
 
 In addition to the PostgreSQL, an implementation of a full in-memory storage exists, which we use for testing and as a reference implementation. Hence, when adding a new database implementation, the in-memory implementation can be used as a reference one. Our tests ensure that both PostgreSQL and in-memory implementation, through the beautiful abstractions of rust, arrive to the same result. Any additional implementation can be added to the same test suite.
 
+#### Database schema versions and upgrades
+
+The storage schema is versioned (`CURRENT_STORAGE_VERSION` in the source code, currently 26). When the blockchain scanner daemon encounters a database with a different version than the one it expects, it re-initializes the database from scratch, dropping the old data and performing a full re-scan of the blockchain; this is also when missing tables are created. Version 26 added the `ml.emitted_events` table, the append-only log that backs the [real-time event stream](#real-time-event-stream). Since the event stream has no replay and old events are never read again, the scanner prunes stream events older than the most recent 10,000.
+
+## Real-time event stream
+
+The API web server exposes a real-time event stream for block explorer clients, based on Server-Sent Events (SSE). Any standard SSE client works: the connection stays open, and the server pushes events as they happen.
+
+### Endpoint
+
+```
+GET /api/v2/stream
+```
+
+The endpoint requires no authentication, consistent with the other `/api/v2` GET endpoints, and responds with the `text/event-stream` content type.
+
+An optional `types` query parameter restricts the streamed event kinds to a comma-separated subset of `tx_seen`, `block`, and `reorg`; by default, all kinds are streamed. Invalid values are rejected with HTTP 400. For example, to receive only block and reorganization events:
+
+```
+GET /api/v2/stream?types=block,reorg
+```
+
+### Event format
+
+Events are *named* SSE events, so clients can subscribe to specific kinds with `EventSource.addEventListener("block", ...)` and so on. Every payload is a JSON object of the form `{"type": <event name>, "content": {...}}`. A `block` event looks like this on the wire:
+
+```
+event: block
+data: {"type":"block","content":{"block_id":"<hex>","height":3,"timestamp":1639975460,"tx_ids":["<hex>"]}}
+```
+
+The content per event kind:
+
+| Event | Content fields |
+| ----- | -------------- |
+| `tx_seen` | `tx_id` (hex); `origin`, which is `local` when the transaction was submitted through this node and `remote` when it was observed coming from the network |
+| `block` | `block_id` (hex); `height`; `timestamp` (unix seconds); `tx_ids`, the list of the block's transaction ids (hex) |
+| `reorg` | `common_ancestor_height`; `removed_block_ids`, the blocks disconnected by the reorganization (hex); `new_tip_height` |
+
+A few additional frames to be aware of:
+
+- The first frame is a `retry:` hint of 3 seconds, telling conforming clients how long to wait before reconnecting.
+- Keepalive comment lines (`: keepalive`) are sent while the stream is idle, every 30 seconds by default, so that proxies and clients can tell the connection is alive.
+- The response carries an `x-accel-buffering: no` header to keep reverse proxies from buffering the stream. If you operate a reverse proxy in front of the web server, make sure response buffering stays disabled, or the events will not reach the clients in real time.
+- If a client falls further behind than the server's per-client event buffer (1024 events by default), it receives a `lag` advisory event, `data: {"skipped": N}`, instead of the missed events. When this happens, reconcile the current state through the regular REST endpoints.
+
+A minimal browser client:
+
+```js
+const source = new EventSource("http://127.0.0.1:3000/api/v2/stream?types=block,reorg");
+source.addEventListener("block", (e) => console.log("block:", JSON.parse(e.data).content));
+source.addEventListener("reorg", (e) => console.log("reorg:", JSON.parse(e.data).content));
+```
+
+### Semantics and limitations
+
+- There is **no replay**: events missed while disconnected, or skipped on lag, are not re-sent, and the `Last-Event-ID` header is not supported. Recover missed data through the regular REST endpoints.
+- `tx_seen` events refer to transactions currently in the node's mempool; such a transaction may never be mined. Only successfully processed transactions are streamed.
+- A `block` event means the block has been fully indexed: the block and its transactions are immediately available through the REST endpoints.
+
 ## How to run
 
 In the following we present the minimal requirements to run the API server in action. In this example, we will be using the testnet. Replace every `testnet` with `mainnet` for the mainnet.
@@ -89,6 +153,14 @@ api-web-server --network testnet --bind-address 127.0.0.1:3000
 
 The API web server will immediately start and connect to the database locally. A specific remote database can be specified using command line arguments. Add `--help` to the previously mentioned commands to see how to do this.
 
+The [real-time event stream](#real-time-event-stream) works out of the box, with no configuration changes needed for existing deployments. Three options are available for tuning:
+
+- `--stream-events-broadcast-capacity` (default 1024): how many events are buffered per connected client before the client receives a `lag` advisory event instead of the missed events.
+- `--stream-events-poll-interval-secs` (default 30): how often the event pump polls the database for new events, as a safety net for missed notifications.
+- `--stream-events-keepalive-interval-secs` (default 30): how often keepalive comments are sent to connected stream clients.
+
+Note that the event pump behind the stream uses the PostgreSQL LISTEN/NOTIFY mechanism on a dedicated connection, so the real-time stream requires the PostgreSQL backend.
+
 ### Testing the API web server
 
 Once the previous steps are complete, you're ready to communicate with the API web server. The following curl command should work (or you can put the link in your browser directly):

api-server/api-server-common/Cargo.toml

@@ -21,6 +21,8 @@ clap = { workspace = true, features = ["derive"] }
 futures = { workspace = true, default-features = false }
 itertools.workspace = true
 parity-scale-codec.workspace = true
+serde = { workspace = true, features = ["derive"] }
+serde_json.workspace = true
 thiserror.workspace = true
 tokio = { workspace = true, features = ["full"] }
 tokio-postgres = "0.7"

api-server/api-server-common/src/lib.rs

@@ -14,6 +14,7 @@
 // limitations under the License.
 
 pub mod storage;
+pub mod streaming;
 
 use clap::Parser;
 use common::chain::config::ChainType;

api-server/api-server-common/src/storage/impls/mod.rs

@@ -13,7 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub const CURRENT_STORAGE_VERSION: u32 = 25;
+pub const CURRENT_STORAGE_VERSION: u32 = 26;
 
 pub mod in_memory;
 pub mod postgres;

api-server/api-server-common/src/storage/impls/postgres/listener.rs

@@ -0,0 +1,187 @@
+// Copyright (c) 2026 RBB S.r.l
+// opensource@mintlayer.org
+// SPDX-License-Identifier: MIT
+// Licensed under the MIT License;
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://github.com/mintlayer/mintlayer-core/blob/master/LICENSE
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Postgres LISTEN/NOTIFY based wakeup source for the stream event pump.
+
+use std::future::poll_fn;
+use std::sync::Arc;
+use std::time::Duration;
+
+use tokio_postgres::AsyncMessage;
+
+use crate::streaming::{
+    STREAM_EVENTS_NOTIFY_CHANNEL, StreamEvent, StreamEventId, StreamEventReadError,
+    StreamEventSource,
+};
+
+use super::TransactionalApiServerPostgresStorage;
+use crate::storage::storage_api::{ApiServerStorageError, ApiServerStorageRead, Transactional};
+
+/// A dedicated Postgres connection that listens for stream event notifications.
+///
+/// Note: Postgres only delivers notifications on the connection that issued `LISTEN`, so this
+/// connection cannot be taken from the connection pool.
+pub struct PostgresEventListener {
+    /// Note: the connection stays alive as long as the client is not dropped; dropping the client
+    /// also terminates the driver task.
+    _client: tokio_postgres::Client,
+    _driver: tokio::task::JoinHandle<()>,
+    wakeups_rx: tokio::sync::mpsc::UnboundedReceiver<()>,
+}
+
+impl PostgresEventListener {
+    /// Wait for the next notification wakeup; `None` means the connection is no longer alive.
+    pub async fn recv(&mut self) -> Option<()> {
+        self.wakeups_rx.recv().await
+    }
+}
+
+/// The production [StreamEventSource]: wakes up on database notifications, with a periodic poll
+/// as a safety net for missed notifications, and reads the new events through the regular
+/// read-only storage transactions.
+pub struct PostgresStreamEventSource {
+    storage: Arc<TransactionalApiServerPostgresStorage>,
+    listener: PostgresEventListener,
+    poll_interval: Duration,
+}
+
+impl PostgresStreamEventSource {
+    pub fn new(
+        storage: Arc<TransactionalApiServerPostgresStorage>,
+        listener: PostgresEventListener,
+        poll_interval: Duration,
+    ) -> Self {
+        Self {
+            storage,
+            listener,
+            poll_interval,
+        }
+    }
+
+    /// Re-establish the listener connection, bounded by the poll interval so that a slow or
+    /// failing database cannot stall the pump for long.
+    async fn reconnect(&mut self) -> Result<(), ApiServerStorageError> {
+        let listener = tokio::time::timeout(self.poll_interval, self.storage.new_event_listener())
+            .await
+            .map_err(|_| {
+                ApiServerStorageError::InitializationError(
+                    "Stream event listener reconnection timed out".to_owned(),
+                )
+            })??;
+
+        logging::log::info!("Stream event listener connection re-established");
+        self.listener = listener;
+        Ok(())
+    }
+}
+
+#[async_trait::async_trait]
+impl StreamEventSource for PostgresStreamEventSource {
+    async fn wait_for_wakeup(&mut self) {
+        tokio::select! {
+            // Note: the pump keeps track of the last seen id itself, so the notification is just
+            // a wakeup signal.
+            wakeup = self.listener.recv() => {
+                if wakeup.is_none() {
+                    // Note: the listener connection is dead; without the reconnect below, the
+                    // pump would fall back to pure polling forever and the notifications would
+                    // never be received again.
+                    if self.reconnect().await.is_err() {
+                        tokio::time::sleep(self.poll_interval).await;
+                    }
+                }
+            }
+            _ = tokio::time::sleep(self.poll_interval) => {}
+        }
+    }
+
+    async fn read_events_after(
+        &mut self,
+        last_seen_id: StreamEventId,
+    ) -> Result<Vec<(StreamEventId, StreamEvent)>, StreamEventReadError> {
+        let db_tx = self
+            .storage
+            .transaction_ro()
+            .await
+            .map_err(|e: ApiServerStorageError| StreamEventReadError(e.to_string()))?;
+        db_tx
+            .read_stream_events_after(last_seen_id)
+            .await
+            .map_err(|e| StreamEventReadError(e.to_string()))
+    }
+}
+
+impl TransactionalApiServerPostgresStorage {
+    /// Create a dedicated connection that listens for stream event notifications.
+    pub async fn new_event_listener(&self) -> Result<PostgresEventListener, ApiServerStorageError> {
+        let (client, mut connection) =
+            self.connection_config.connect(tokio_postgres::NoTls).await.map_err(|e| {
+                ApiServerStorageError::InitializationError(format!(
+                    "Stream event listener connection failed: {e}"
+                ))
+            })?;
+
+        let (wakeups_tx, wakeups_rx) = tokio::sync::mpsc::unbounded_channel();
+
+        // Note: the connection driver task must be spawned before issuing any queries, because
+        // the client only queues the requests and the driver is the one that performs them.
+        let driver = tokio::spawn(async move {
+            // Note: the stream of messages ends when the connection is closed, e.g. when the
+            // client is dropped.
+            loop {
+                match poll_fn(|cx| connection.poll_message(cx)).await {
+                    // Note: the notification payload (the id of the last emitted event) is
+                    // intentionally ignored; the pump keeps track of the last seen id itself.
+                    Some(Ok(AsyncMessage::Notification(_))) => {
+                        let _ = wakeups_tx.send(());
+                    }
+                    Some(Ok(_)) => {}
+                    Some(Err(err)) => {
+                        logging::log::error!("Stream event listener connection error: {err}");
+                        break;
+                    }
+                    None => break,
+                }
+            }
+        });
+
+        client
+            .batch_execute(&format!("LISTEN {STREAM_EVENTS_NOTIFY_CHANNEL};"))
+            .await
+            .map_err(|e| {
+                ApiServerStorageError::InitializationError(format!(
+                    "Stream event listener setup failed: {e}"
+                ))
+            })?;
+
+        Ok(PostgresEventListener {
+            _client: client,
+            _driver: driver,
+            wakeups_rx,
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // The listener and the event source are exercised by the postgres backend test suite and the
+    // stack tests, which require a real database.
+    #[test]
+    fn notify_channel_name_is_stable() {
+        assert_eq!(STREAM_EVENTS_NOTIFY_CHANNEL, "mintlayer_events");
+    }
+}