diff --git a/Cargo.lock b/Cargo.lock index f5bf8e6fc1..0a1de389f9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -317,6 +317,8 @@ dependencies = [ "parity-scale-codec", "pos-accounting", "rstest", + "serde", + "serde_json", "serialization", "test-utils", "thiserror 1.0.69", @@ -329,6 +331,7 @@ name = "api-server-stack-test-suite" version = "1.4.0" dependencies = [ "api-blockchain-scanner-lib", + "api-server-backend-test-suite", "api-server-common", "api-web-server", "async-trait", @@ -367,6 +370,7 @@ dependencies = [ "common", "crypto", "ctor", + "futures", "hex", "logging", "mempool", diff --git a/accounting/src/delta/delta_data_collection/mod.rs b/accounting/src/delta/delta_data_collection/mod.rs index f8b2d4f7b5..5ab261a867 100644 --- a/accounting/src/delta/delta_data_collection/mod.rs +++ b/accounting/src/delta/delta_data_collection/mod.rs @@ -158,6 +158,12 @@ impl FromIterator<(K, DataDelta)> for DeltaDataColle } } +impl Default for DeltaDataCollection { + 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( lhs: DataDelta, diff --git a/accounting/src/delta/delta_data_collection/undo.rs b/accounting/src/delta/delta_data_collection/undo.rs index f5ac64572f..927642b808 100644 --- a/accounting/src/delta/delta_data_collection/undo.rs +++ b/accounting/src/delta/delta_data_collection/undo.rs @@ -61,3 +61,9 @@ impl DeltaDataUndoCollection { self.data } } + +impl Default for DeltaDataUndoCollection { + fn default() -> Self { + Self::new() + } +} diff --git a/api-server/CHANGELOG.md b/api-server/CHANGELOG.md index 2b5400dca7..4bbb3bc8b0 100644 --- a/api-server/CHANGELOG.md +++ b/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 diff --git a/api-server/README.md b/api-server/README.md index 183c2a19f7..6649f378b6 100644 --- a/api-server/README.md +++ b/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": , "content": {...}}`. A `block` event looks like this on the wire: + +``` +event: block +data: {"type":"block","content":{"block_id":"","height":3,"timestamp":1639975460,"tx_ids":[""]}} +``` + +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): diff --git a/api-server/api-server-common/Cargo.toml b/api-server/api-server-common/Cargo.toml index 532df17ed4..839db04675 100644 --- a/api-server/api-server-common/Cargo.toml +++ b/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" diff --git a/api-server/api-server-common/src/lib.rs b/api-server/api-server-common/src/lib.rs index 9d979da77a..b0ce699678 100644 --- a/api-server/api-server-common/src/lib.rs +++ b/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; diff --git a/api-server/api-server-common/src/storage/impls/mod.rs b/api-server/api-server-common/src/storage/impls/mod.rs index d457134176..c224d5b546 100644 --- a/api-server/api-server-common/src/storage/impls/mod.rs +++ b/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; diff --git a/api-server/api-server-common/src/storage/impls/postgres/listener.rs b/api-server/api-server-common/src/storage/impls/postgres/listener.rs new file mode 100644 index 0000000000..3bd9052105 --- /dev/null +++ b/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, + listener: PostgresEventListener, + poll_interval: Duration, +} + +impl PostgresStreamEventSource { + pub fn new( + storage: Arc, + 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, 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 { + 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"); + } +} diff --git a/api-server/api-server-common/src/storage/impls/postgres/mod.rs b/api-server/api-server-common/src/storage/impls/postgres/mod.rs index a8cb06bd8f..5b15fa77e7 100644 --- a/api-server/api-server-common/src/storage/impls/postgres/mod.rs +++ b/api-server/api-server-common/src/storage/impls/postgres/mod.rs @@ -13,6 +13,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +pub mod listener; pub mod transactional; mod queries; @@ -31,8 +32,12 @@ use crate::storage::storage_api::ApiServerStorageError; use self::transactional::ApiServerPostgresTransactionalRo; use self::transactional::ApiServerPostgresTransactionalRw; +pub use self::listener::{PostgresEventListener, PostgresStreamEventSource}; + pub struct TransactionalApiServerPostgresStorage { pool: Pool>, + /// The connection configuration, used to create dedicated connections, e.g. for LISTEN/NOTIFY. + connection_config: tokio_postgres::Config, /// This task is responsible for rolling back failed RW/RO transactions, since closing connections are pooled tx_dropper_joiner: tokio::task::JoinHandle<()>, /// This channel is used to send transactions that are not manually rolled back to the tx_dropper task to roll them back @@ -81,7 +86,7 @@ impl TransactionalApiServerPostgresStorage { )) })?; - let manager = PostgresConnectionManager::new(config, NoTls); + let manager = PostgresConnectionManager::new(config.clone(), NoTls); let pool = Pool::builder().max_size(max_connections).build(manager).await.map_err(|e| { ApiServerStorageError::InitializationError(format!( "Postgres connection pool creation error: {}", @@ -106,6 +111,7 @@ impl TransactionalApiServerPostgresStorage { let result = Self { pool, + connection_config: config, tx_dropper_joiner, db_tx_conn_sender: conn_tx, chain_config, diff --git a/api-server/api-server-common/src/storage/impls/postgres/queries.rs b/api-server/api-server-common/src/storage/impls/postgres/queries.rs index b5644ffd20..bf9efe27cb 100644 --- a/api-server/api-server-common/src/storage/impls/postgres/queries.rs +++ b/api-server/api-server-common/src/storage/impls/postgres/queries.rs @@ -42,6 +42,10 @@ use crate::storage::{ block_aux_data::{BlockAuxData, BlockWithExtraData}, }, }; +use crate::streaming::{ + STREAM_EVENTS_NOTIFY_CHANNEL, STREAM_EVENTS_READ_BATCH_SIZE, STREAM_EVENTS_RETENTION_COUNT, + StreamEvent, StreamEventId, +}; const VERSION_STR: &str = "version"; @@ -1046,6 +1050,17 @@ impl<'a, 'b> QueryFromConnection<'a, 'b> { ) .await?; + // Append-only log of stream events for real-time clients; the primary key index also + // serves the `WHERE id > $1 ORDER BY id` reads of the event pump. + self.just_execute( + "CREATE TABLE ml.emitted_events ( + id BIGSERIAL PRIMARY KEY, + kind TEXT NOT NULL, + payload JSONB NOT NULL + );", + ) + .await?; + logging::log::info!("Done creating database tables"); Ok(()) @@ -3047,6 +3062,97 @@ impl<'a, 'b> QueryFromConnection<'a, 'b> { }) .collect() } + + pub async fn append_stream_event( + &mut self, + event: &StreamEvent, + ) -> Result { + let kind = event.event_name(); + let payload = serde_json::to_string(event).map_err(|e| { + ApiServerStorageError::LowLevelStorageError(format!( + "Stream event serialization failed: {e}" + )) + })?; + + let row = self + .tx + .query_one( + // Note: the payload is bound as text and cast to jsonb, because the string types + // of the postgres driver do not serialize into the jsonb type directly. + "INSERT INTO ml.emitted_events (kind, payload) VALUES ($1, ($2::text)::jsonb) + RETURNING id;", + &[&kind, &payload], + ) + .await + .map_err(|e| ApiServerStorageError::LowLevelStorageError(e.to_string()))?; + + Ok(row.get(0)) + } + + pub async fn read_stream_events_after( + &self, + last_seen_id: StreamEventId, + ) -> Result, ApiServerStorageError> { + let rows = self + .tx + .query( + // Note: the reads are batched to keep the memory use of the event pump bounded + // regardless of the backlog size. + "SELECT id, payload::text FROM ml.emitted_events WHERE id > $1 + ORDER BY id ASC LIMIT $2;", + &[&last_seen_id, &STREAM_EVENTS_READ_BATCH_SIZE], + ) + .await + .map_err(|e| ApiServerStorageError::LowLevelStorageError(e.to_string()))?; + + // Note: rows that cannot be decoded are skipped (with a warning) instead of failing the + // whole batch, so that a single corrupted row cannot stall the stream. + Ok(rows + .into_iter() + .filter_map(|row| { + let id: i64 = row.get(0); + let payload: String = row.get(1); + match serde_json::from_str(&payload) { + Ok(event) => Some((id, event)), + Err(err) => { + logging::log::warn!("Skipping undecodable stream event #{id}: {err}"); + None + } + } + }) + .collect()) + } + + /// Delete the stream events that fell out of the retention window. + pub async fn prune_stream_events(&mut self) -> Result<(), ApiServerStorageError> { + self.tx + .execute( + "DELETE FROM ml.emitted_events + WHERE id <= (SELECT COALESCE(max(id), 0) - $1 FROM ml.emitted_events);", + &[&STREAM_EVENTS_RETENTION_COUNT], + ) + .await + .map_err(|e| ApiServerStorageError::LowLevelStorageError(e.to_string()))?; + + Ok(()) + } + + pub async fn notify_new_stream_events( + &mut self, + last_event_id: StreamEventId, + ) -> Result<(), ApiServerStorageError> { + // Note: the notification is sent within the transaction that appends the events, so + // Postgres delivers it only if/when that transaction commits. + self.tx + .execute( + "SELECT pg_notify($1, $2::text);", + &[&STREAM_EVENTS_NOTIFY_CHANNEL, &last_event_id.to_string()], + ) + .await + .map_err(|e| ApiServerStorageError::LowLevelStorageError(e.to_string()))?; + + Ok(()) + } } fn decode_order_from_row( diff --git a/api-server/api-server-common/src/storage/impls/postgres/transactional/read.rs b/api-server/api-server-common/src/storage/impls/postgres/transactional/read.rs index 71000c7390..6165a2f0b0 100644 --- a/api-server/api-server-common/src/storage/impls/postgres/transactional/read.rs +++ b/api-server/api-server-common/src/storage/impls/postgres/transactional/read.rs @@ -30,6 +30,7 @@ use crate::storage::{ UtxoWithExtraInfo, block_aux_data::BlockAuxData, }, }; +use crate::streaming::{StreamEvent, StreamEventId}; use std::collections::BTreeMap; use common::chain::UtxoOutPoint; @@ -448,4 +449,14 @@ impl ApiServerStorageRead for ApiServerPostgresTransactionalRo<'_> { Ok(res) } + + async fn read_stream_events_after( + &self, + last_seen_id: StreamEventId, + ) -> Result, ApiServerStorageError> { + let conn = QueryFromConnection::new(self.connection.as_ref().expect(CONN_ERR)); + let res = conn.read_stream_events_after(last_seen_id).await?; + + Ok(res) + } } diff --git a/api-server/api-server-common/src/storage/impls/postgres/transactional/write.rs b/api-server/api-server-common/src/storage/impls/postgres/transactional/write.rs index 6429c5c208..ded120ea58 100644 --- a/api-server/api-server-common/src/storage/impls/postgres/transactional/write.rs +++ b/api-server/api-server-common/src/storage/impls/postgres/transactional/write.rs @@ -35,6 +35,7 @@ use crate::storage::{ block_aux_data::{BlockAuxData, BlockWithExtraData}, }, }; +use crate::streaming::{StreamEvent, StreamEventId}; use super::{ApiServerPostgresTransactionalRw, CONN_ERR}; @@ -403,6 +404,33 @@ impl ApiServerStorageWrite for ApiServerPostgresTransactionalRw<'_> { Ok(()) } + + async fn append_stream_event( + &mut self, + event: &StreamEvent, + ) -> Result { + let mut conn = QueryFromConnection::new(self.connection.as_ref().expect(CONN_ERR)); + let res = conn.append_stream_event(event).await?; + + Ok(res) + } + + async fn notify_new_stream_events( + &mut self, + last_event_id: StreamEventId, + ) -> Result<(), ApiServerStorageError> { + let mut conn = QueryFromConnection::new(self.connection.as_ref().expect(CONN_ERR)); + conn.notify_new_stream_events(last_event_id).await?; + + Ok(()) + } + + async fn prune_stream_events(&mut self) -> Result<(), ApiServerStorageError> { + let mut conn = QueryFromConnection::new(self.connection.as_ref().expect(CONN_ERR)); + conn.prune_stream_events().await?; + + Ok(()) + } } #[async_trait::async_trait] @@ -813,4 +841,14 @@ impl ApiServerStorageRead for ApiServerPostgresTransactionalRw<'_> { Ok(res) } + + async fn read_stream_events_after( + &self, + last_seen_id: StreamEventId, + ) -> Result, ApiServerStorageError> { + let conn = QueryFromConnection::new(self.connection.as_ref().expect(CONN_ERR)); + let res = conn.read_stream_events_after(last_seen_id).await?; + + Ok(res) + } } diff --git a/api-server/api-server-common/src/storage/storage_api/mod.rs b/api-server/api-server-common/src/storage/storage_api/mod.rs index 54c409e7fe..83f02d4ddc 100644 --- a/api-server/api-server-common/src/storage/storage_api/mod.rs +++ b/api-server/api-server-common/src/storage/storage_api/mod.rs @@ -38,6 +38,7 @@ use pos_accounting::{Error as PosError, PoolData}; use serialization::{Decode, Encode}; use self::block_aux_data::{BlockAuxData, BlockWithExtraData}; +use crate::streaming::{StreamEvent, StreamEventId}; pub mod block_aux_data; @@ -800,6 +801,16 @@ pub trait ApiServerStorageRead: Sync { len: u32, offset: u64, ) -> Result, ApiServerStorageError>; + + /// Read the stream events with an id greater than `last_seen_id`, in ascending id order. + /// + /// Note: backends that don't support stream events simply return an empty list. + async fn read_stream_events_after( + &self, + _last_seen_id: StreamEventId, + ) -> Result, ApiServerStorageError> { + Ok(Vec::new()) + } } #[async_trait::async_trait] @@ -1000,6 +1011,41 @@ pub trait ApiServerStorageWrite: ApiServerStorageRead { &mut self, block_height: BlockHeight, ) -> Result<(), ApiServerStorageError>; + + /// Append an event to the stream event log and return the assigned event id. + /// + /// The event becomes visible to readers only after the enclosing transaction has been + /// committed, which is what guarantees that an event is never observed before the data it + /// refers to. + /// + /// Note: backends that don't support stream events silently drop the event and return a + /// dummy id. + async fn append_stream_event( + &mut self, + _event: &StreamEvent, + ) -> Result { + Ok(0) + } + + /// Request a notification to be delivered to stream event listeners when (and only when) the + /// enclosing transaction commits, signaling that all events up to `last_event_id` are + /// available. + /// + /// Note: backends that don't support stream event notifications silently do nothing; listeners + /// are expected to poll for new events as a fallback anyway. + async fn notify_new_stream_events( + &mut self, + _last_event_id: StreamEventId, + ) -> Result<(), ApiServerStorageError> { + Ok(()) + } + + /// Delete the stream events that fell out of the retention window. + /// + /// Note: backends that don't support stream events silently do nothing. + async fn prune_stream_events(&mut self) -> Result<(), ApiServerStorageError> { + Ok(()) + } } #[async_trait::async_trait] diff --git a/api-server/api-server-common/src/streaming.rs b/api-server/api-server-common/src/streaming.rs new file mode 100644 index 0000000000..fab40795fe --- /dev/null +++ b/api-server/api-server-common/src/streaming.rs @@ -0,0 +1,425 @@ +// 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. + +//! Real-time event streaming for block explorer clients. +//! +//! Events are produced by two independent sources: +//! * the scanner subsystem appends [StreamEvent::Block] and [StreamEvent::Reorg] events to the +//! storage *inside* the same database transaction that performs the indexing work, so once an +//! event is committed, the data it refers to is immediately queryable through the regular REST +//! endpoints; +//! * the web server bridges mempool events from the node's WebSocket RPC into +//! [StreamEvent::TxSeen] events. +//! +//! The Postgres backend additionally issues a `pg_notify` wakeup after the transaction commits, +//! which the web server uses to promptly drain new events and forward them to all connected +//! stream subscribers. Event payloads travel in the `ml.emitted_events` table and are only +//! referenced by the notification, because `pg_notify` payloads are capped at 8000 bytes. + +use std::fmt; +use std::str::FromStr; +use std::sync::Arc; + +use common::{ + chain::{Block, Transaction, block::timestamp::BlockTimestamp}, + primitives::{BlockHeight, Id}, +}; +use serde::{Deserialize, Serialize}; + +/// The name of the Postgres notification channel used as a commit wakeup ping. +pub const STREAM_EVENTS_NOTIFY_CHANNEL: &str = "mintlayer_events"; + +/// Monotonic identifier of a stream event, as assigned by the storage backend. +pub type StreamEventId = i64; + +/// The maximum number of stream events the event pump reads per database round trip, keeping the +/// memory use of the pump bounded regardless of the backlog size. +pub const STREAM_EVENTS_READ_BATCH_SIZE: i64 = 1000; + +/// The number of the most recent stream events the storage backend retains; older events are +/// pruned, since the stream endpoint does not support replays anyway. +pub const STREAM_EVENTS_RETENTION_COUNT: i64 = 10_000; + +/// The simplified origin of a transaction seen in the mempool. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum TxOrigin { + Local, + Remote, +} + +/// An event that is streamed to block explorer clients. +/// +/// The payloads are intentionally kept small; the explorer is expected to hydrate the details +/// through the regular REST endpoints. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type", content = "content")] +pub enum StreamEvent { + /// A transaction reached the node's mempool. + TxSeen { + tx_id: Id, + origin: TxOrigin, + }, + /// A block has been fully indexed into the api-server database. + Block { + block_id: Id, + height: BlockHeight, + timestamp: BlockTimestamp, + tx_ids: Vec>, + }, + /// Previously indexed blocks have been disconnected after a chain reorganization. + Reorg { + common_ancestor_height: BlockHeight, + removed_block_ids: Vec>, + new_tip_height: BlockHeight, + }, +} + +impl StreamEvent { + /// The kind of the event. + pub fn event_type(&self) -> StreamEventType { + match self { + StreamEvent::TxSeen { .. } => StreamEventType::TxSeen, + StreamEvent::Block { .. } => StreamEventType::Block, + StreamEvent::Reorg { .. } => StreamEventType::Reorg, + } + } + + /// The name of the event as used in the Server-Sent Events protocol, so that clients can + /// subscribe to specific event kinds with `EventSource.addEventListener(name, ...)`. + pub fn event_name(&self) -> &'static str { + self.event_type().name() + } +} + +/// The kinds of the stream events; the single source of truth for the event names used in the +/// Server-Sent Events protocol. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +pub enum StreamEventType { + TxSeen, + Block, + Reorg, +} + +impl StreamEventType { + /// All the event kinds, in a stable order. + pub const ALL: &'static [StreamEventType] = &[Self::TxSeen, Self::Block, Self::Reorg]; + + /// The name of the event kind as used in the Server-Sent Events protocol. + pub fn name(self) -> &'static str { + match self { + StreamEventType::TxSeen => "tx_seen", + StreamEventType::Block => "block", + StreamEventType::Reorg => "reorg", + } + } +} + +impl fmt::Display for StreamEventType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.name()) + } +} + +impl FromStr for StreamEventType { + type Err = StreamEventTypeParseError; + + fn from_str(s: &str) -> Result { + let s = s.trim().to_lowercase(); + Self::ALL + .iter() + .copied() + .find(|kind| kind.name() == s) + .ok_or(StreamEventTypeParseError) + } +} + +/// The error returned when a string does not name a valid [StreamEventType]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct StreamEventTypeParseError; + +impl fmt::Display for StreamEventTypeParseError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "invalid stream event type, expected one of: {}", + StreamEventType::ALL + .iter() + .map(|kind| kind.name()) + .collect::>() + .join(", ") + ) + } +} + +/// The error returned when reading the stream events from the source fails. +#[derive(Debug, thiserror::Error)] +#[error("Failed to read stream events: {0}")] +pub struct StreamEventReadError(pub String); + +/// A cloneable handle for distributing stream events to subscribers. +#[derive(Clone)] +pub struct StreamEventsChannel { + sender: Arc>, +} + +impl StreamEventsChannel { + /// Create a channel that can buffer up to `capacity` events per subscriber. + pub fn new(capacity: usize) -> Self { + let (sender, _receiver) = tokio::sync::broadcast::channel(capacity); + Self { + sender: Arc::new(sender), + } + } + + /// 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 { + self.sender.subscribe() + } + + /// Send an event to all subscribers. Returns `Err` only if there are no subscribers. + pub fn send( + &self, + event: StreamEvent, + ) -> Result<(), tokio::sync::broadcast::error::SendError> { + self.sender.send(event).map(|_| ()) + } +} + +/// The source of the stream events for the [event pump][run_event_pump]. +/// +/// The wait-and-poll behavior is abstracted away so that the pump logic itself (backlog draining, +/// resume from the last seen id) can be tested independently of the database backend. +#[async_trait::async_trait] +pub trait StreamEventSource: Send { + /// Wait until new events may be available. This is caused either by a database notification + /// or by the periodic poll timeout elapsing. + async fn wait_for_wakeup(&mut self); + + /// Read the events with an id greater than `last_seen_id`, in ascending id order. + async fn read_events_after( + &mut self, + last_seen_id: StreamEventId, + ) -> Result, StreamEventReadError>; +} + +/// The event pump: forwards stream events from a [StreamEventSource] into a +/// [StreamEventsChannel], tracking the id of the last forwarded event. +/// +/// The backlog is drained in batches before waiting for the next wakeup, which also takes care of +/// draining whatever was committed before this pump was started. +pub async fn run_event_pump( + mut source: impl StreamEventSource, + channel: StreamEventsChannel, + mut last_seen_id: StreamEventId, +) { + loop { + // Note: a full batch hints at more events being available, in which case the next batch + // is read immediately; this keeps the memory use of the pump bounded while large backlogs + // are forwarded without waiting for the next wakeup. + loop { + match source.read_events_after(last_seen_id).await { + Ok(events) => { + let batch_size = events.len() as i64; + for (event_id, event) in events { + logging::log::debug!( + "Streaming event #{event_id} ({})", + event.event_name(), + ); + // Note: sending may fail when there are no subscribers; this is not an + // error, the events are persisted and can still be read later. + let _ = channel.send(event); + last_seen_id = event_id; + } + if batch_size < STREAM_EVENTS_READ_BATCH_SIZE { + break; + } + } + Err(err) => { + logging::log::error!("{err}"); + break; + } + } + } + + source.wait_for_wakeup().await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + use common::primitives::H256; + + fn test_block_event(height: u64) -> StreamEvent { + StreamEvent::Block { + block_id: Id::new(H256::from_low_u64_be(height)), + height: BlockHeight::new(height), + timestamp: BlockTimestamp::from_int_seconds(1_000_000 + height), + tx_ids: vec![Id::new(H256::from_low_u64_be(1000 + height))], + } + } + + fn test_reorg_event(height: u64) -> StreamEvent { + StreamEvent::Reorg { + common_ancestor_height: BlockHeight::new(height), + removed_block_ids: vec![Id::new(H256::from_low_u64_be(2000 + height))], + new_tip_height: BlockHeight::new(height + 1), + } + } + + /// A source that yields pre-scripted events, one batch per wakeup. + struct FakeSource { + batches: std::sync::Mutex>>, + notify_rx: tokio::sync::mpsc::UnboundedReceiver<()>, + } + + impl FakeSource { + fn new() -> FakeSourceParts { + let (batch_tx, batch_rx) = std::sync::mpsc::channel(); + let (notify_tx, notify_rx) = tokio::sync::mpsc::unbounded_channel(); + ( + Self { + batches: std::sync::Mutex::new(batch_rx), + notify_rx, + }, + batch_tx, + notify_tx, + ) + } + } + + type FakeEventBatch = Vec<(StreamEventId, StreamEvent)>; + + type FakeSourceParts = ( + FakeSource, + std::sync::mpsc::Sender, + tokio::sync::mpsc::UnboundedSender<()>, + ); + + #[async_trait::async_trait] + impl StreamEventSource for FakeSource { + async fn wait_for_wakeup(&mut self) { + // Note: the pump calls this after each read, so the test must not block forever if + // there are no more notifications; hence the short timeout. + let _ = tokio::time::timeout(Duration::from_millis(100), self.notify_rx.recv()).await; + } + + async fn read_events_after( + &mut self, + last_seen_id: StreamEventId, + ) -> Result, StreamEventReadError> { + let batch = self.batches.lock().unwrap().try_recv().ok().unwrap_or_default(); + Ok(batch.into_iter().filter(|(id, _)| *id > last_seen_id).collect()) + } + } + + #[tokio::test] + async fn pump_drains_backlog_and_resumes_from_last_seen_id() { + let (source, batch_tx, notify_tx) = FakeSource::new(); + let channel = StreamEventsChannel::new(16); + let mut rx = channel.subscribe(); + + let pump = tokio::spawn(run_event_pump(source, channel.clone(), 0)); + + // Note: the backlog, committed before the pump even saw the wakeup. + batch_tx.send(vec![(1, test_block_event(1)), (2, test_block_event(2))]).unwrap(); + notify_tx.send(()).unwrap(); + + assert_eq!(rx.recv().await.unwrap(), test_block_event(1)); + assert_eq!(rx.recv().await.unwrap(), test_block_event(2)); + + // A new batch is forwarded and the previously seen ids are not re-sent. + batch_tx.send(vec![(2, test_block_event(2)), (3, test_reorg_event(2))]).unwrap(); + notify_tx.send(()).unwrap(); + + assert_eq!(rx.recv().await.unwrap(), test_reorg_event(2)); + + pump.abort(); + } + + #[tokio::test] + async fn pump_keeps_running_after_read_errors() { + struct FailingSource { + failing: bool, + } + + #[async_trait::async_trait] + impl StreamEventSource for FailingSource { + async fn wait_for_wakeup(&mut self) { + // Note: without this, the pump would spin without ever yielding on a + // single-threaded runtime. + tokio::time::sleep(Duration::from_millis(1)).await; + } + + async fn read_events_after( + &mut self, + _last_seen_id: StreamEventId, + ) -> Result, StreamEventReadError> { + if self.failing { + self.failing = false; + Err(StreamEventReadError("storage error".to_owned())) + } else { + self.failing = true; + Ok(vec![(7, test_block_event(7))]) + } + } + } + + let channel = StreamEventsChannel::new(16); + let mut rx = channel.subscribe(); + let pump = tokio::spawn(run_event_pump(FailingSource { failing: false }, channel, 0)); + + assert_eq!(rx.recv().await.unwrap(), test_block_event(7)); + // Note: the pump survives the error and keeps polling. + assert_eq!(rx.recv().await.unwrap(), test_block_event(7)); + + pump.abort(); + } + + #[test] + fn event_serde_roundtrip() { + let events = [ + StreamEvent::TxSeen { + tx_id: Id::new(H256::from_low_u64_be(1)), + origin: TxOrigin::Remote, + }, + test_block_event(3), + test_reorg_event(2), + ]; + + for event in events { + let json = serde_json::to_string(&event).unwrap(); + let deserialized: StreamEvent = serde_json::from_str(&json).unwrap(); + assert_eq!(deserialized, event); + } + } + + #[test] + fn event_names() { + assert_eq!( + StreamEvent::TxSeen { + tx_id: Id::new(H256::zero()), + origin: TxOrigin::Local, + } + .event_name(), + "tx_seen" + ); + assert_eq!(test_block_event(0).event_name(), "block"); + assert_eq!(test_reorg_event(0).event_name(), "reorg"); + } +} diff --git a/api-server/scanner-lib/src/blockchain_state/mod.rs b/api-server/scanner-lib/src/blockchain_state/mod.rs index c5146296a3..e83a40ce78 100644 --- a/api-server/scanner-lib/src/blockchain_state/mod.rs +++ b/api-server/scanner-lib/src/blockchain_state/mod.rs @@ -27,6 +27,7 @@ use api_server_common::storage::storage_api::{ PoolDataWithExtraInfo, TransactionInfo, TxAdditionalInfo, Utxo, UtxoLock, block_aux_data::{BlockAuxData, BlockWithExtraData}, }; +use api_server_common::streaming::{StreamEvent, StreamEventId}; use chainstate::{ calculate_median_time_past_from_blocktimestamps, constraints_value_accumulator::{AccumulatedFee, ConstrainedValueAccumulator}, @@ -122,9 +123,37 @@ impl LocalBlockchainState for BlockchainState ) -> Result<(), Self::Error> { let mut db_tx = self.storage.transaction_rw().await.expect("Unable to connect to database"); + // Note: a reorg happened iff the local best block is above the common ancestor with the + // new chain, in which case the blocks above the common ancestor are about to be + // disconnected. + let local_best_height = db_tx.get_best_block().await?.block_height(); + let is_reorg = local_best_height > common_block_height; + let removed_block_ids = if is_reorg { + // Note: the removed block ids must be captured before the disconnect below, because + // the disconnect only marks the blocks as disconnected instead of removing them. + Some( + capture_main_chain_block_ids(&mut db_tx, common_block_height, local_best_height) + .await?, + ) + } else { + None + }; + disconnect_tables_above_height(&mut db_tx, common_block_height) .await .expect("Unable to disconnect tables"); + + let mut last_event_id: StreamEventId = 0; + if let Some(removed_block_ids) = removed_block_ids { + let new_tip_height = next_block_height(common_block_height, blocks.len()); + let event = StreamEvent::Reorg { + common_ancestor_height: common_block_height, + removed_block_ids, + new_tip_height, + }; + last_event_id = db_tx.append_stream_event(&event).await?; + } + let mut next_order_number = db_tx.get_last_transaction_global_index().await?.map_or(0, |idx| idx + 1); @@ -201,6 +230,14 @@ impl LocalBlockchainState for BlockchainState .await .expect("Unable to set block"); + let event = StreamEvent::Block { + block_id, + height: block_height, + timestamp: block_timestamp, + tx_ids: block.transactions().iter().map(|tx| tx.transaction().get_id()).collect(), + }; + last_event_id = db_tx.append_stream_event(&event).await?; + for (idx, tx_info) in transactions.iter().enumerate() { db_tx .set_transaction( @@ -237,6 +274,15 @@ impl LocalBlockchainState for BlockchainState .expect("Unable to update tables from block"); } + // Note: sent within the transaction, so the notification is only delivered if the + // transaction commits; this is what wakes up the stream event pump in the web server. + // Note: the guard also skips the notify for the backends that don't support stream + // events, since those return the dummy event id 0 from the appends. + if last_event_id > 0 { + db_tx.prune_stream_events().await?; + db_tx.notify_new_stream_events(last_event_id).await?; + } + db_tx.commit().await.expect("Unable to commit transaction"); logging::log::info!("Database commit completed successfully"); @@ -244,6 +290,32 @@ impl LocalBlockchainState for BlockchainState } } +/// The height of the block that follows `block_count` blocks connected on top of `base_height`. +fn next_block_height(base_height: BlockHeight, block_count: usize) -> BlockHeight { + BlockHeight::new(base_height.into_int() + block_count as u64) +} + +/// Collect the ids of the main chain blocks above `common_block_height`, up to and including +/// `best_block_height`. +/// +/// Note: this issues one query per height; the cost is linear in the reorg depth, which is +/// bounded by the number of blocks fetched in a single scanner batch. +async fn capture_main_chain_block_ids( + db_tx: &mut T, + common_block_height: BlockHeight, + best_block_height: BlockHeight, +) -> Result>, ApiServerStorageError> { + let mut block_ids = Vec::with_capacity( + best_block_height.into_int().saturating_sub(common_block_height.into_int()) as usize, + ); + 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); + } + } + Ok(block_ids) +} + // Find locked UTXOs that are unlocked at this height or time and update address balances async fn update_locked_amounts_for_current_block( db_tx: &mut T, diff --git a/api-server/scanner-lib/src/sync/tests/mod.rs b/api-server/scanner-lib/src/sync/tests/mod.rs index d8345f2463..25aabce4c5 100644 --- a/api-server/scanner-lib/src/sync/tests/mod.rs +++ b/api-server/scanner-lib/src/sync/tests/mod.rs @@ -1043,15 +1043,13 @@ fn create_block( transactions: Vec, ) -> Block { tf.progress_time_seconds_since_epoch(target_block_time.as_secs()); - let block = tf - .make_pos_block_builder() + tf.make_pos_block_builder() .with_parent(prev_block_hash) .with_stake_spending_key(staking_sk) .with_vrf_key(vrf_sk.clone()) .with_stake_pool_id(pool_id) .with_transactions(transactions) - .build(&mut *rng); - block + .build(&mut *rng) } async fn sync_and_compare( diff --git a/api-server/stack-test-suite/Cargo.toml b/api-server/stack-test-suite/Cargo.toml index aeff77856b..f0004a51dd 100644 --- a/api-server/stack-test-suite/Cargo.toml +++ b/api-server/stack-test-suite/Cargo.toml @@ -7,6 +7,7 @@ rust-version.workspace = true [dev-dependencies] api-blockchain-scanner-lib = { path = "../scanner-lib" } +api-server-backend-test-suite = { path = "../storage-test-suite" } api-server-common = { path = "../api-server-common" } api-web-server = { path = "../web-server" } chainstate = { path = "../../chainstate" } diff --git a/api-server/stack-test-suite/tests/common/mod.rs b/api-server/stack-test-suite/tests/common/mod.rs new file mode 100644 index 0000000000..72b9df91d9 --- /dev/null +++ b/api-server/stack-test-suite/tests/common/mod.rs @@ -0,0 +1,50 @@ +// Copyright (c) 2023 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. + +//! Helpers shared between the test binaries of this crate. +//! +//! Note: this module is linked separately into each test binary, so an item used by only some of +//! the binaries must not be reported as unused; hence the blanket `allow` below. + +#![allow(dead_code)] + +use api_web_server::TxSubmitClient; +use common::chain::SignedTransaction; +use mempool::FeeRate; +use node_comm::rpc_client::NodeRpcError; + +/// A no-op RPC client for the web server state under test. +pub struct DummyRPC {} + +#[async_trait::async_trait] +impl TxSubmitClient for DummyRPC { + async fn submit_tx(&self, _: SignedTransaction) -> Result<(), NodeRpcError> { + Ok(()) + } + + async fn get_feerate_points(&self) -> Result, NodeRpcError> { + Ok(vec![]) + } +} + +/// The value of the `event:` field of an SSE frame, if any. +pub fn frame_event_name(frame: &str) -> Option<&str> { + frame.lines().find_map(|line| line.strip_prefix("event: ")) +} + +/// The value of the `data:` field of an SSE frame, if any. +pub fn frame_data(frame: &str) -> Option<&str> { + frame.lines().find_map(|line| line.strip_prefix("data: ")) +} diff --git a/api-server/stack-test-suite/tests/in_memory.rs b/api-server/stack-test-suite/tests/in_memory.rs index a6bdb6d73e..8c976d0ba7 100644 --- a/api-server/stack-test-suite/tests/in_memory.rs +++ b/api-server/stack-test-suite/tests/in_memory.rs @@ -13,31 +13,19 @@ // See the License for the specific language governing permissions and // limitations under the License. +// Note: the module must not be called `common`, which would be ambiguous with the `common` +// workspace crate; the path attribute decouples the module name from the file name. +#[path = "common/mod.rs"] +mod test_common; mod v2; use api_server_common::storage::impls::in_memory::transactional::TransactionalApiServerInMemoryStorage; -use api_web_server::{ApiServerWebServerState, CachedValues, TxSubmitClient, api::web_server}; -use common::{ - chain::{SignedTransaction, config::create_unit_test_config}, - primitives::time::get_time, -}; -use mempool::FeeRate; -use node_comm::rpc_client::NodeRpcError; +use api_web_server::{ApiServerWebServerState, CachedValues, api::web_server}; +use common::{chain::config::create_unit_test_config, primitives::time::get_time}; use std::sync::{Arc, RwLock}; use tokio::net::TcpListener; -struct DummyRPC {} - -#[async_trait::async_trait] -impl TxSubmitClient for DummyRPC { - async fn submit_tx(&self, _: SignedTransaction) -> Result<(), NodeRpcError> { - Ok(()) - } - - async fn get_feerate_points(&self) -> Result, NodeRpcError> { - Ok(vec![]) - } -} +pub use test_common::DummyRPC; pub async fn spawn_webserver(url: &str) -> (tokio::task::JoinHandle<()>, reqwest::Response) { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -56,6 +44,7 @@ pub async fn spawn_webserver(url: &str) -> (tokio::task::JoinHandle<()>, reqwest feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/postgres_stream.rs b/api-server/stack-test-suite/tests/postgres_stream.rs new file mode 100644 index 0000000000..a0f941aec7 --- /dev/null +++ b/api-server/stack-test-suite/tests/postgres_stream.rs @@ -0,0 +1,358 @@ +// Copyright (c) 2023 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. + +//! End-to-end test of the real-time event streaming over Postgres: +//! scanner -> emitted_events -> event pump -> SSE endpoint, including a reorg. +//! +//! Note: the scanner side is driven through the public `BlockchainState` API, which is the exact +//! code path `sync_once` delegates to for applying blocks (including the stream event appends and +//! the pg_notify wakeup inside the same transaction). + +// Note: the module must not be called `common`, which would be ambiguous with the `common` +// workspace crate; the path attribute decouples the module name from the file name. +#[path = "common/mod.rs"] +mod test_common; + +use std::{ + sync::{Arc, RwLock}, + time::Duration, +}; + +use api_blockchain_scanner_lib::{ + blockchain_state::BlockchainState, sync::local_state::LocalBlockchainState, +}; +use api_server_backend_test_suite::podman::{Container, Podman}; +use api_server_common::storage::{ + impls::postgres::{PostgresStreamEventSource, TransactionalApiServerPostgresStorage}, + storage_api::{ApiServerStorageWrite, ApiServerTransactionRw, Transactional}, +}; +use api_server_common::streaming::{StreamEvent, StreamEventsChannel}; +use api_web_server::{ + ApiServerWebServerState, CachedValues, StreamEventsHandle, StreamingConfig, api::web_server, + streaming::run_database_event_pump, +}; +use chainstate_test_framework::TestFramework; +use common::{ + chain::Block, + primitives::{BlockHeight, Id, Idable, time::get_time}, +}; +use hex::ToHex as _; +use test_common::{DummyRPC, frame_data, frame_event_name}; +use test_utils::random::{Seed, make_seedable_rng}; + +#[ctor::ctor] +fn init() { + logging::init_logging(); +} + +/// The time to wait for a single expected stream event. +const EVENT_TIMEOUT: Duration = Duration::from_secs(5); + +/// Receive the next stream event from the collector, failing if it doesn't arrive in time. +async fn recv_event( + event_rx: &mut tokio::sync::mpsc::UnboundedReceiver<(String, StreamEvent)>, +) -> (String, StreamEvent) { + tokio::time::timeout(EVENT_TIMEOUT, event_rx.recv()) + .await + .expect("timed out waiting for a stream event") + .expect("the stream event collector has been closed") +} + +#[tokio::test] +async fn stream_events_postgres_end_to_end() { + let mut rng = make_seedable_rng(Seed::from_entropy()); + let mut tf = TestFramework::builder(&mut rng).build(); + let chain_config = tf.chain_config().clone(); + + // ----------------------------------------------------------------------------------------- + // The Postgres container and two storage instances (two pools), mirroring the scanner and + // the web server processes. + // ----------------------------------------------------------------------------------------- + let mut podman = Podman::new( + "MintlayerPostgresStreamTest", + Container::PostgresFromDockerHub, + ) + .with_env("POSTGRES_HOST_AUTH_METHOD", "trust") + .with_env( + "POSTGRES_DB", + format!("mintlayer-{}", chain_config.chain_type().name()).as_str(), + ) + .with_port_mapping(None, 5432); + podman.run(); + + let host_port = podman.get_port_mapping(5432).unwrap(); + let new_storage = || { + TransactionalApiServerPostgresStorage::new( + "127.0.0.1", + host_port, + "postgres", + None, + None, + 5, + chain_config.clone(), + ) + }; + + let mut scanner_storage = new_storage().await.unwrap(); + let web_storage = Arc::new(new_storage().await.unwrap()); + + // Initialize the schema. + { + let mut db_tx = scanner_storage.transaction_rw().await.unwrap(); + db_tx.reinitialize_storage(&chain_config).await.unwrap(); + db_tx.commit().await.unwrap(); + } + + // ----------------------------------------------------------------------------------------- + // Start the event pump first, then the web server. + // ----------------------------------------------------------------------------------------- + let handle = StreamEventsHandle::new( + StreamEventsChannel::new(64), + StreamingConfig { + keepalive_interval: Duration::from_millis(200), + }, + ); + let event_listener = web_storage.new_event_listener().await.unwrap(); + let source = PostgresStreamEventSource::new( + Arc::clone(&web_storage), + event_listener, + Duration::from_millis(200), + ); + tokio::spawn(run_database_event_pump(source, handle.clone())); + + let http_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = http_listener.local_addr().unwrap(); + + let web_storage_for_task = Arc::clone(&web_storage); + let chain_config_for_task = Arc::clone(&chain_config); + let web_task = tokio::spawn(async move { + let web_server_state = ApiServerWebServerState { + db: web_storage_for_task, + chain_config: chain_config_for_task, + rpc: Arc::new(DummyRPC {}), + cached_values: Arc::new(CachedValues { + feerate_points: RwLock::new((get_time(), vec![])), + }), + time_getter: Default::default(), + stream_events: handle, + }; + + web_server(http_listener, web_server_state, true).await.unwrap(); + }); + + // ----------------------------------------------------------------------------------------- + // The SSE connection and the event collector. The collector is fed by a task reading the + // event stream; the test waits for the connection to be established before any event is + // produced, since a stream subscriber must exist for the pump's sends to reach it. + // ----------------------------------------------------------------------------------------- + let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel::<(String, StreamEvent)>(); + let (connected_tx, connected_rx) = tokio::sync::oneshot::channel::<()>(); + + let sse_url = format!("http://{}:{}/api/v2/stream", addr.ip(), addr.port()); + tokio::spawn(async move { + let mut connect_attempts = 0u32; + let client = reqwest::Client::new(); + // Note: the listener is already bound, so this connect resolves as soon as the web + // server has taken the listener over; retry in case it hasn't yet. + let mut response = loop { + match client.get(&sse_url).send().await { + Ok(response) if response.status() == 200 => break response, + _ => { + connect_attempts += 1; + assert!( + connect_attempts < 100, + "failed to connect to the stream endpoint" + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + }; + + assert_eq!( + response + .headers() + .get("x-accel-buffering") + .expect("x-accel-buffering header must be present"), + "no" + ); + + // Note: fails the test (via the oneshot) if the endpoint cannot be connected to. + _ = connected_tx.send(()); + + let mut buffer = String::new(); + loop { + let chunk = match tokio::time::timeout(Duration::from_secs(60), response.chunk()).await + { + Ok(Ok(Some(chunk))) => chunk, + // Timed out or the stream ended; either way the collector stops here. + _ => break, + }; + buffer.push_str(std::str::from_utf8(&chunk).unwrap()); + + while let Some(frame_end) = buffer.find("\n\n") { + let frame: String = buffer.drain(..frame_end + 2).collect(); + if let (Some(name), Some(data)) = (frame_event_name(&frame), frame_data(&frame)) { + let event: StreamEvent = + serde_json::from_str(data).expect("the event payload must be valid JSON"); + if event_tx.send((name.to_owned(), event)).is_err() { + // The test has finished; stop collecting. + return; + } + } + } + } + }); + connected_rx.await.unwrap(); + + let rest_client = reqwest::Client::new(); + let block_url = |block_id: Id| { + format!( + "http://{}:{}/api/v2/block/{}", + addr.ip(), + addr.port(), + block_id.to_hash().encode_hex::() + ) + }; + + // ----------------------------------------------------------------------------------------- + // First sync: three blocks connected on top of genesis must produce exactly three block + // events, in height order. + // ----------------------------------------------------------------------------------------- + let mainchain_block_ids = tf + .create_chain_return_ids(&chain_config.genesis_block_id(), 3, &mut rng) + .unwrap(); + let mainchain_blocks: Vec = mainchain_block_ids + .iter() + .map(|id| tf.block(tf.to_chain_block_id(id))) + .collect(); + + let mut scanner = BlockchainState::new(chain_config.clone(), scanner_storage); + scanner.scan_genesis(chain_config.genesis_block().as_ref()).await.unwrap(); + scanner + .scan_blocks(BlockHeight::new(0), mainchain_blocks.clone()) + .await + .unwrap(); + + for (idx, block) in mainchain_blocks.iter().enumerate() { + let expected_event = StreamEvent::Block { + block_id: block.get_id(), + height: BlockHeight::new(idx as u64 + 1), + timestamp: block.timestamp(), + tx_ids: block.transactions().iter().map(|tx| tx.transaction().get_id()).collect(), + }; + + let (name, event) = recv_event(&mut event_rx).await; + assert_eq!(name, "block"); + assert_eq!(event, expected_event, "unexpected event for block #{idx}"); + + if idx == 0 { + // Transactional consistency: the block carried by the first event must be + // immediately queryable through the regular REST endpoint. + let response = rest_client.get(block_url(block.get_id())).send().await.unwrap(); + assert_eq!(response.status(), 200); + + let body: serde_json::Value = + serde_json::from_str(&response.text().await.unwrap()).unwrap(); + let served_tx_ids: Vec = body["body"]["transactions"] + .as_array() + .expect("transactions must be an array") + .iter() + .map(|tx| tx["id"].as_str().expect("tx id must be a string").to_owned()) + .collect(); + let expected_tx_ids: Vec = block + .transactions() + .iter() + .map(|tx| tx.transaction().get_id().to_hash().encode_hex::()) + .collect(); + assert_eq!(served_tx_ids, expected_tx_ids); + } + } + + // ----------------------------------------------------------------------------------------- + // Reorg: create a heavier fork from the block at height 1. The previously mainchain blocks + // at heights 2..3 are removed, the fork blocks at heights 2..4 are connected. + // ----------------------------------------------------------------------------------------- + let removed_block_ids: Vec> = [2u64, 3] + .iter() + .map(|height| { + let id = tf + .chainstate + .get_block_id_from_height(BlockHeight::new(*height)) + .unwrap() + .unwrap(); + tf.to_chain_block_id(&id) + }) + .collect(); + + let fork_parent_id = + tf.chainstate.get_block_id_from_height(BlockHeight::new(1)).unwrap().unwrap(); + let fork_block_ids = tf.create_chain_return_ids(&fork_parent_id, 3, &mut rng).unwrap(); + let fork_blocks: Vec = + fork_block_ids.iter().map(|id| tf.block(tf.to_chain_block_id(id))).collect(); + + scanner.scan_blocks(BlockHeight::new(1), fork_blocks.clone()).await.unwrap(); + + // Exactly one reorg event must arrive, before the events of the new fork blocks. + let (name, event) = recv_event(&mut event_rx).await; + assert_eq!(name, "reorg"); + assert_eq!( + event, + StreamEvent::Reorg { + common_ancestor_height: BlockHeight::new(1), + removed_block_ids: removed_block_ids.clone(), + new_tip_height: BlockHeight::new(4), + } + ); + + // Then the block events of the new fork blocks, in height order 2..4. + for (idx, block) in fork_blocks.iter().enumerate() { + let expected_event = StreamEvent::Block { + block_id: block.get_id(), + height: BlockHeight::new(idx as u64 + 2), + timestamp: block.timestamp(), + tx_ids: block.transactions().iter().map(|tx| tx.transaction().get_id()).collect(), + }; + + let (name, event) = recv_event(&mut event_rx).await; + assert_eq!(name, "block"); + assert_eq!(event, expected_event, "unexpected fork event #{idx}"); + } + + // The removed blocks must still be fetchable: the explorer database keeps disconnected + // blocks with a null height. + for removed_block_id in &removed_block_ids { + let response = rest_client.get(block_url(*removed_block_id)).send().await.unwrap(); + assert_eq!( + response.status(), + 200, + "disconnected block must remain fetchable" + ); + } + + // ----------------------------------------------------------------------------------------- + // No duplicates: after everything settles, no further events may arrive (the expected total + // is exactly one reorg + 3 + 3 block events, all of which have been consumed above). + // ----------------------------------------------------------------------------------------- + tokio::time::sleep(Duration::from_secs(1)).await; + match tokio::time::timeout(Duration::from_secs(1), event_rx.recv()).await { + Err(_timed_out) => {} // no more events, as expected + Ok(Some((name, event))) => { + panic!("unexpected extra stream event: {name} {event:?}") + } + Ok(None) => panic!("the stream event collector has been closed"), + } + + web_task.abort(); +} diff --git a/api-server/stack-test-suite/tests/v2/address.rs b/api-server/stack-test-suite/tests/v2/address.rs index c2bb4c2531..707f20bc3b 100644 --- a/api-server/stack-test-suite/tests/v2/address.rs +++ b/api-server/stack-test-suite/tests/v2/address.rs @@ -272,6 +272,7 @@ async fn multiple_outputs_to_single_address(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -524,6 +525,7 @@ async fn test_unlocking_for_locked_utxos(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -752,6 +754,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/address_all_utxos.rs b/api-server/stack-test-suite/tests/v2/address_all_utxos.rs index 8f7f2fc1bc..05a9c4cff0 100644 --- a/api-server/stack-test-suite/tests/v2/address_all_utxos.rs +++ b/api-server/stack-test-suite/tests/v2/address_all_utxos.rs @@ -281,6 +281,7 @@ async fn multiple_utxos_to_single_address(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -537,6 +538,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/address_delegations.rs b/api-server/stack-test-suite/tests/v2/address_delegations.rs index 4400334f0b..e7acd6d953 100644 --- a/api-server/stack-test-suite/tests/v2/address_delegations.rs +++ b/api-server/stack-test-suite/tests/v2/address_delegations.rs @@ -212,6 +212,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/address_spendable_utxos.rs b/api-server/stack-test-suite/tests/v2/address_spendable_utxos.rs index 1b571bf4b9..8a79ce2c19 100644 --- a/api-server/stack-test-suite/tests/v2/address_spendable_utxos.rs +++ b/api-server/stack-test-suite/tests/v2/address_spendable_utxos.rs @@ -283,6 +283,7 @@ async fn multiple_utxos_to_single_address(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -529,6 +530,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/address_token_authority.rs b/api-server/stack-test-suite/tests/v2/address_token_authority.rs index 5779997bbc..fc7ba5b321 100644 --- a/api-server/stack-test-suite/tests/v2/address_token_authority.rs +++ b/api-server/stack-test-suite/tests/v2/address_token_authority.rs @@ -233,6 +233,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/block.rs b/api-server/stack-test-suite/tests/v2/block.rs index 8c8f4bf494..ec07c48e8d 100644 --- a/api-server/stack-test-suite/tests/v2/block.rs +++ b/api-server/stack-test-suite/tests/v2/block.rs @@ -230,6 +230,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/block_header.rs b/api-server/stack-test-suite/tests/v2/block_header.rs index c537d2022a..81dc4030e3 100644 --- a/api-server/stack-test-suite/tests/v2/block_header.rs +++ b/api-server/stack-test-suite/tests/v2/block_header.rs @@ -138,6 +138,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/block_reward.rs b/api-server/stack-test-suite/tests/v2/block_reward.rs index 0319988902..8e5520c29e 100644 --- a/api-server/stack-test-suite/tests/v2/block_reward.rs +++ b/api-server/stack-test-suite/tests/v2/block_reward.rs @@ -115,6 +115,7 @@ async fn no_reward(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -219,6 +220,7 @@ async fn has_reward(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/block_transaction_ids.rs b/api-server/stack-test-suite/tests/v2/block_transaction_ids.rs index 3b4343af16..1b05d56188 100644 --- a/api-server/stack-test-suite/tests/v2/block_transaction_ids.rs +++ b/api-server/stack-test-suite/tests/v2/block_transaction_ids.rs @@ -123,6 +123,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/chain_at_height.rs b/api-server/stack-test-suite/tests/v2/chain_at_height.rs index 8c3edef2a6..ca4f9e6195 100644 --- a/api-server/stack-test-suite/tests/v2/chain_at_height.rs +++ b/api-server/stack-test-suite/tests/v2/chain_at_height.rs @@ -133,6 +133,7 @@ async fn height_n(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/chain_tip.rs b/api-server/stack-test-suite/tests/v2/chain_tip.rs index 30fb774919..b38319e7e3 100644 --- a/api-server/stack-test-suite/tests/v2/chain_tip.rs +++ b/api-server/stack-test-suite/tests/v2/chain_tip.rs @@ -53,6 +53,7 @@ async fn at_genesis() { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -145,6 +146,7 @@ async fn height_n(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/feerate.rs b/api-server/stack-test-suite/tests/v2/feerate.rs index 9ba10d8799..6df8b7bc76 100644 --- a/api-server/stack-test-suite/tests/v2/feerate.rs +++ b/api-server/stack-test-suite/tests/v2/feerate.rs @@ -73,6 +73,7 @@ async fn ok(#[case] seed: Seed) { )), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -142,6 +143,7 @@ async fn ok_reload_feerate(#[case] seed: Seed) { )), }), time_getter, + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/htlc.rs b/api-server/stack-test-suite/tests/v2/htlc.rs index 3fb71f6557..6de1513afc 100644 --- a/api-server/stack-test-suite/tests/v2/htlc.rs +++ b/api-server/stack-test-suite/tests/v2/htlc.rs @@ -185,6 +185,7 @@ async fn spend(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -357,6 +358,7 @@ async fn refund(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/mod.rs b/api-server/stack-test-suite/tests/v2/mod.rs index 94cbb711d0..34552cd934 100644 --- a/api-server/stack-test-suite/tests/v2/mod.rs +++ b/api-server/stack-test-suite/tests/v2/mod.rs @@ -33,6 +33,7 @@ mod pool; mod pool_block_stats; mod pools; mod statistics; +mod stream; mod token; mod token_ids; mod token_ticker; @@ -124,6 +125,7 @@ async fn chain_genesis() { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/nft.rs b/api-server/stack-test-suite/tests/v2/nft.rs index cace5e1ed9..8bcb65d7af 100644 --- a/api-server/stack-test-suite/tests/v2/nft.rs +++ b/api-server/stack-test-suite/tests/v2/nft.rs @@ -197,6 +197,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/orders.rs b/api-server/stack-test-suite/tests/v2/orders.rs index d13fa0e610..a85feccd23 100644 --- a/api-server/stack-test-suite/tests/v2/orders.rs +++ b/api-server/stack-test-suite/tests/v2/orders.rs @@ -173,6 +173,7 @@ async fn create_fill_conclude_order(#[case] seed: Seed, #[case] version: OrdersV feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -279,6 +280,7 @@ async fn order_pairs(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/pool.rs b/api-server/stack-test-suite/tests/v2/pool.rs index b9242ea3c2..6de53fa528 100644 --- a/api-server/stack-test-suite/tests/v2/pool.rs +++ b/api-server/stack-test-suite/tests/v2/pool.rs @@ -181,6 +181,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/pool_block_stats.rs b/api-server/stack-test-suite/tests/v2/pool_block_stats.rs index 88564552ed..bf0044e8a8 100644 --- a/api-server/stack-test-suite/tests/v2/pool_block_stats.rs +++ b/api-server/stack-test-suite/tests/v2/pool_block_stats.rs @@ -169,6 +169,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/pools.rs b/api-server/stack-test-suite/tests/v2/pools.rs index 9021ff48f4..e54417b7c4 100644 --- a/api-server/stack-test-suite/tests/v2/pools.rs +++ b/api-server/stack-test-suite/tests/v2/pools.rs @@ -208,6 +208,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/statistics.rs b/api-server/stack-test-suite/tests/v2/statistics.rs index 8932ec3cc9..a0c9f63820 100644 --- a/api-server/stack-test-suite/tests/v2/statistics.rs +++ b/api-server/stack-test-suite/tests/v2/statistics.rs @@ -256,6 +256,7 @@ async fn ok_tokens(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -413,6 +414,7 @@ async fn ok_coins(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/stream.rs b/api-server/stack-test-suite/tests/v2/stream.rs new file mode 100644 index 0000000000..84d965acc2 --- /dev/null +++ b/api-server/stack-test-suite/tests/v2/stream.rs @@ -0,0 +1,416 @@ +// Copyright (c) 2023 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. + +//! Tests of the `/api/v2/stream` Server-Sent Events endpoint (in-memory backend). + +use std::{ + sync::{Arc, RwLock}, + time::Duration, +}; + +use api_blockchain_scanner_lib::{ + blockchain_state::BlockchainState, sync::local_state::LocalBlockchainState, +}; +use api_server_common::storage::{ + impls::in_memory::transactional::TransactionalApiServerInMemoryStorage, + storage_api::{ApiServerStorageWrite, ApiServerTransactionRw, Transactional}, +}; +use api_server_common::streaming::{StreamEvent, StreamEventsChannel, TxOrigin}; +use api_web_server::{ + ApiServerWebServerState, CachedValues, StreamEventsHandle, StreamingConfig, api::web_server, +}; +use chainstate_test_framework::TestFramework; +use common::{ + chain::{ + Block, Transaction, block::timestamp::BlockTimestamp, config::create_unit_test_config, + }, + primitives::{BlockHeight, H256, Id, Idable, time::get_time}, +}; +use hex::ToHex as _; +use test_utils::random::{Seed, make_seedable_rng}; + +use crate::DummyRPC; +use crate::test_common::{frame_data, frame_event_name}; + +/// The time to wait for a single expected SSE frame. +const FRAME_TIMEOUT: Duration = Duration::from_secs(5); + +/// A minimal SSE client on top of `reqwest`, able to read raw SSE frames. +/// +/// Note: `Response::chunk()` is used for reading, which does not require the `stream` feature of +/// `reqwest`. +struct SseConnection { + response: reqwest::Response, + buffer: String, +} + +impl SseConnection { + /// Connect to the given SSE endpoint and check the response contract. + async fn connect(client: &reqwest::Client, url: &str) -> Self { + let response = client.get(url).send().await.unwrap(); + + assert_eq!(response.status(), 200); + let content_type = response + .headers() + .get("content-type") + .expect("content-type header must be present") + .to_str() + .unwrap(); + assert!( + content_type.starts_with("text/event-stream"), + "unexpected content-type: {content_type}" + ); + assert_eq!( + response + .headers() + .get("x-accel-buffering") + .expect("x-accel-buffering header must be present"), + "no" + ); + + Self { + response, + buffer: String::new(), + } + } + + /// Read the next complete SSE frame (up to the empty-line separator), failing if it doesn't + /// arrive within `timeout`. + async fn next_frame(&mut self, timeout: Duration) -> String { + let deadline = tokio::time::Instant::now() + timeout; + loop { + if let Some(frame_end) = self.buffer.find("\n\n") { + return self.buffer.drain(..frame_end + 2).collect(); + } + + let chunk = tokio::time::timeout( + deadline.saturating_duration_since(tokio::time::Instant::now()), + self.response.chunk(), + ) + .await + .expect("timed out while waiting for an SSE frame") + .expect("SSE request failed") + .expect("SSE stream ended unexpectedly"); + + self.buffer.push_str(std::str::from_utf8(&chunk).unwrap()); + } + } + + /// Read the next complete SSE frame carrying a named event, skipping the non-event frames + /// the endpoint is expected to send (the initial `retry:` directive and the keepalive + /// comments), failing if it doesn't arrive within `timeout`. + async fn next_event_frame(&mut self, timeout: Duration) -> String { + let deadline = tokio::time::Instant::now() + timeout; + loop { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + let frame = self.next_frame(remaining).await; + if frame_event_name(&frame).is_some() { + return frame; + } + assert!( + is_non_event_frame(&frame), + "unexpected SSE frame: {frame:?}" + ); + } + } +} + +/// Whether the frame is a non-event frame the endpoint is allowed to send: a keepalive comment +/// or an SSE `retry:` directive (the latter as the first frame of the stream, per the SSE spec). +/// Note: empty lines are frame separators rather than fields, so they are ignored. +fn is_non_event_frame(frame: &str) -> bool { + frame + .lines() + .filter(|line| !line.is_empty()) + .all(|line| line.starts_with(':') || line.starts_with("retry:")) +} + +/// Connect to the stream endpoint of a web server spawned on the given listener with the given +/// handle, using the default event filter. +async fn connect_to_stream(addr: std::net::SocketAddr, query: &str) -> SseConnection { + let client = reqwest::Client::new(); + let url = format!("http://{}:{}/api/v2/stream{query}", addr.ip(), addr.port()); + SseConnection::connect(&client, &url).await +} + +/// Spawn a web server on the given listener, backed by the in-memory storage and sharing the +/// given stream events handle. +fn spawn_stream_webserver( + listener: tokio::net::TcpListener, + stream_events: StreamEventsHandle, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let chain_config = Arc::new(create_unit_test_config()); + let storage = TransactionalApiServerInMemoryStorage::new(&chain_config); + + let web_server_state = ApiServerWebServerState { + db: Arc::new(storage), + chain_config: Arc::clone(&chain_config), + rpc: Arc::new(DummyRPC {}), + cached_values: Arc::new(CachedValues { + feerate_points: RwLock::new((get_time(), vec![])), + }), + time_getter: Default::default(), + stream_events, + }; + + web_server(listener, web_server_state, true).await.unwrap(); + }) +} + +#[tokio::test] +async fn stream_endpoint_contract() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + // Note: the channel is cloned before being moved into the handle, so that the test can send + // events into it. + let channel = StreamEventsChannel::new(16); + let handle = StreamEventsHandle::new( + channel.clone(), + StreamingConfig { + keepalive_interval: Duration::from_millis(200), + }, + ); + + let task = spawn_stream_webserver(listener, handle); + + // Note: the subscription must exist before the events are sent, so the connection is opened + // first; the response headers arriving means the endpoint has subscribed to the channel. + let mut sse = connect_to_stream(addr, "").await; + + let tx_seen = StreamEvent::TxSeen { + tx_id: Id::new(H256::from_low_u64_be(1)), + origin: TxOrigin::Local, + }; + let block = StreamEvent::Block { + block_id: Id::new(H256::from_low_u64_be(2)), + height: BlockHeight::new(1), + timestamp: BlockTimestamp::from_int_seconds(1_000), + tx_ids: vec![Id::new(H256::from_low_u64_be(3))], + }; + let reorg = StreamEvent::Reorg { + common_ancestor_height: BlockHeight::new(0), + removed_block_ids: vec![Id::new(H256::from_low_u64_be(4))], + new_tip_height: BlockHeight::new(2), + }; + + // Note: sending must succeed, since the endpoint is subscribed. + channel.send(tx_seen.clone()).unwrap(); + channel.send(block.clone()).unwrap(); + channel.send(reorg.clone()).unwrap(); + + for expected in [&tx_seen, &block, &reorg] { + let frame = sse.next_event_frame(FRAME_TIMEOUT).await; + + assert_eq!( + frame_event_name(&frame), + Some(expected.event_name()), + "unexpected SSE frame: {frame:?}" + ); + + let data = frame_data(&frame).expect("the event must carry a data field"); + assert!( + data.starts_with('{'), + "the data payload must be a JSON object: {data}" + ); + let parsed: StreamEvent = serde_json::from_str(data) + .expect("the data payload must parse back into a StreamEvent"); + assert_eq!(&parsed, expected, "event roundtrip mismatch"); + } + + // The keepalive comment must arrive when the stream is idle. + let frame = sse.next_frame(FRAME_TIMEOUT).await; + assert_eq!(frame.trim(), ": keepalive", "expected a keepalive comment"); + + task.abort(); +} + +#[tokio::test] +async fn stream_types_filter() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let channel = StreamEventsChannel::new(16); + let handle = StreamEventsHandle::new( + channel.clone(), + StreamingConfig { + keepalive_interval: Duration::from_millis(200), + }, + ); + + let task = spawn_stream_webserver(listener, handle); + + let mut sse = connect_to_stream(addr, "?types=block").await; + + let tx_seen = StreamEvent::TxSeen { + tx_id: Id::new(H256::from_low_u64_be(10)), + origin: TxOrigin::Remote, + }; + let block = StreamEvent::Block { + block_id: Id::new(H256::from_low_u64_be(11)), + height: BlockHeight::new(1), + timestamp: BlockTimestamp::from_int_seconds(2_000), + tx_ids: vec![], + }; + + channel.send(tx_seen.clone()).unwrap(); + channel.send(block.clone()).unwrap(); + + // The block event must arrive (possibly preceded by keepalive comments and the initial + // `retry:` directive). + let frame = sse.next_event_frame(FRAME_TIMEOUT).await; + assert_eq!( + frame_event_name(&frame), + Some("block"), + "unexpected event instead of the block event: {frame:?}" + ); + + // No tx_seen event must ever arrive within the window after the block event. Keepalive + // comments are allowed to arrive. + 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 { + Err(_elapsed) => break, // the observation window is over + Ok(frame) => assert_ne!( + frame_event_name(&frame), + Some("tx_seen"), + "tx_seen must not be delivered to a block-only subscription" + ), + } + } + + // An invalid event type must be rejected. + let client = reqwest::Client::new(); + let response = client + .get(format!( + "http://{}:{}/api/v2/stream?types=bogus", + addr.ip(), + addr.port() + )) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 400); + + task.abort(); +} + +/// A block event must refer to block data that is queryable through the regular REST endpoint. +#[tokio::test] +async fn stream_block_event_is_queryable() { + let mut rng = make_seedable_rng(Seed::from_entropy()); + let mut tf = TestFramework::builder(&mut rng).build(); + let chain_config = tf.chain_config().clone(); + + let block_ids = tf + .create_chain_return_ids(&chain_config.genesis_block_id(), 2, &mut rng) + .unwrap(); + let blocks: Vec = + block_ids.iter().map(|id| tf.block(tf.to_chain_block_id(id))).collect(); + + // Scan the blocks into the in-memory storage through the real scanner. Note: the in-memory + // backend drops stream events, so the event is sent through the channel manually below. + let storage = { + let mut storage = TransactionalApiServerInMemoryStorage::new(&chain_config); + let mut db_tx = storage.transaction_rw().await.unwrap(); + db_tx.reinitialize_storage(&chain_config).await.unwrap(); + db_tx.commit().await.unwrap(); + storage + }; + let mut scanner = BlockchainState::new(chain_config.clone(), storage); + scanner.scan_genesis(chain_config.genesis_block().as_ref()).await.unwrap(); + scanner.scan_blocks(BlockHeight::new(0), blocks.clone()).await.unwrap(); + + let scanned_block = blocks[0].clone(); + let block_id = scanned_block.get_id(); + let tx_ids: Vec> = scanned_block + .transactions() + .iter() + .map(|tx| tx.transaction().get_id()) + .collect(); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let channel = StreamEventsChannel::new(16); + let handle = StreamEventsHandle::new( + channel.clone(), + StreamingConfig { + keepalive_interval: Duration::from_millis(200), + }, + ); + + let web_storage = scanner.storage().clone_storage().await; + let task = tokio::spawn(async move { + let web_server_state = ApiServerWebServerState { + db: Arc::new(web_storage), + chain_config, + rpc: Arc::new(DummyRPC {}), + cached_values: Arc::new(CachedValues { + feerate_points: RwLock::new((get_time(), vec![])), + }), + time_getter: Default::default(), + stream_events: handle, + }; + + web_server(listener, web_server_state, true).await.unwrap(); + }); + + let mut sse = connect_to_stream(addr, "").await; + + // This is the event the scanner would have emitted for the scanned block. + let event = StreamEvent::Block { + block_id, + height: BlockHeight::new(1), + timestamp: scanned_block.timestamp(), + tx_ids: tx_ids.clone(), + }; + channel.send(event.clone()).unwrap(); + + let frame = sse.next_event_frame(FRAME_TIMEOUT).await; + assert_eq!(frame_event_name(&frame), Some("block")); + let data = frame_data(&frame).expect("the block event must carry a data field"); + let parsed: StreamEvent = serde_json::from_str(data).unwrap(); + assert_eq!(parsed, event); + + // The event data must be immediately queryable through the REST endpoint. + let client = reqwest::Client::new(); + let response = client + .get(format!( + "http://{}:{}/api/v2/block/{}", + addr.ip(), + addr.port(), + block_id.to_hash().encode_hex::() + )) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 200); + + let body: serde_json::Value = serde_json::from_str(&response.text().await.unwrap()).unwrap(); + let served_tx_ids: Vec<&str> = body["body"]["transactions"] + .as_array() + .expect("transactions must be an array") + .iter() + .map(|tx| tx["id"].as_str().expect("tx id must be a string")) + .collect(); + let expected_tx_ids: Vec = + tx_ids.iter().map(|tx_id| tx_id.to_hash().encode_hex::()).collect(); + assert_eq!(served_tx_ids, expected_tx_ids); + + task.abort(); +} diff --git a/api-server/stack-test-suite/tests/v2/token.rs b/api-server/stack-test-suite/tests/v2/token.rs index 8cf4322685..304d459065 100644 --- a/api-server/stack-test-suite/tests/v2/token.rs +++ b/api-server/stack-test-suite/tests/v2/token.rs @@ -188,6 +188,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/token_ids.rs b/api-server/stack-test-suite/tests/v2/token_ids.rs index 6a3b6ce947..6bfe7d35eb 100644 --- a/api-server/stack-test-suite/tests/v2/token_ids.rs +++ b/api-server/stack-test-suite/tests/v2/token_ids.rs @@ -213,6 +213,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/token_ticker.rs b/api-server/stack-test-suite/tests/v2/token_ticker.rs index 4517d30748..ceda8800e2 100644 --- a/api-server/stack-test-suite/tests/v2/token_ticker.rs +++ b/api-server/stack-test-suite/tests/v2/token_ticker.rs @@ -217,6 +217,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/token_transactions.rs b/api-server/stack-test-suite/tests/v2/token_transactions.rs index a6e4831208..8d818a970b 100644 --- a/api-server/stack-test-suite/tests/v2/token_transactions.rs +++ b/api-server/stack-test-suite/tests/v2/token_transactions.rs @@ -258,6 +258,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/transaction.rs b/api-server/stack-test-suite/tests/v2/transaction.rs index d8467e6e9e..5c0d79fc0b 100644 --- a/api-server/stack-test-suite/tests/v2/transaction.rs +++ b/api-server/stack-test-suite/tests/v2/transaction.rs @@ -240,6 +240,7 @@ async fn multiple_tx_in_same_block(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -388,6 +389,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -577,6 +579,7 @@ async fn mint_tokens(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/transaction_merkle_path.rs b/api-server/stack-test-suite/tests/v2/transaction_merkle_path.rs index d0f12d10e9..3992235590 100644 --- a/api-server/stack-test-suite/tests/v2/transaction_merkle_path.rs +++ b/api-server/stack-test-suite/tests/v2/transaction_merkle_path.rs @@ -132,6 +132,7 @@ async fn cannot_find_transaction_in_block(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -246,6 +247,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/transaction_output.rs b/api-server/stack-test-suite/tests/v2/transaction_output.rs index a21f8a746c..8717bdd24e 100644 --- a/api-server/stack-test-suite/tests/v2/transaction_output.rs +++ b/api-server/stack-test-suite/tests/v2/transaction_output.rs @@ -133,6 +133,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/transaction_submit.rs b/api-server/stack-test-suite/tests/v2/transaction_submit.rs index e20f32308f..bb02d27b46 100644 --- a/api-server/stack-test-suite/tests/v2/transaction_submit.rs +++ b/api-server/stack-test-suite/tests/v2/transaction_submit.rs @@ -40,6 +40,7 @@ async fn dissabled_post_route() { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -89,6 +90,7 @@ async fn invalid_transaction() { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; @@ -144,6 +146,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/stack-test-suite/tests/v2/transactions.rs b/api-server/stack-test-suite/tests/v2/transactions.rs index 38d5767477..88b86ae05a 100644 --- a/api-server/stack-test-suite/tests/v2/transactions.rs +++ b/api-server/stack-test-suite/tests/v2/transactions.rs @@ -203,6 +203,7 @@ async fn ok(#[case] seed: Seed) { feerate_points: RwLock::new((get_time(), vec![])), }), time_getter: Default::default(), + stream_events: Default::default(), } }; diff --git a/api-server/storage-test-suite/src/basic.rs b/api-server/storage-test-suite/src/basic.rs index 72a053a520..7ea60ddf25 100644 --- a/api-server/storage-test-suite/src/basic.rs +++ b/api-server/storage-test-suite/src/basic.rs @@ -2233,11 +2233,103 @@ where { vec![ make_test!(initialization, storage_maker.clone()), - make_test!(set_get, storage_maker), + make_test!(set_get, storage_maker.clone()), + make_test!(stream_events_append_and_read, storage_maker), ] .into_iter() } +/// Streaming events are only visible to readers after the writing transaction has been committed, +/// and are returned in ascending id order. +/// +/// Note: the backends that don't support stream events simply drop the appended events and return +/// an empty list on reads. +pub async fn stream_events_append_and_read( + storage_maker: Arc, + _seed_maker: Box Seed + Send>, +) -> Result<(), Failed> +where + S: ApiServerStorage, + Fut: Future + Send + 'static, + F: Fn() -> Fut, +{ + use api_server_common::streaming::{StreamEvent, TxOrigin}; + use common::primitives::H256; + + let tx_id = Id::::new(H256::from_low_u64_be(1)); + let events = vec![ + StreamEvent::TxSeen { + tx_id, + origin: TxOrigin::Local, + }, + StreamEvent::TxSeen { + tx_id, + origin: TxOrigin::Remote, + }, + ]; + + let mut storage = storage_maker().await; + { + let mut tx = storage.transaction_rw().await.unwrap(); + let chain_config = create_unit_test_config(); + tx.reinitialize_storage(&chain_config).await.unwrap(); + tx.commit().await.unwrap(); + } + + // Phase 1: events appended in a rolled back transaction are never visible. + { + let mut tx = storage.transaction_rw().await.unwrap(); + for event in &events { + tx.append_stream_event(event).await.unwrap(); + } + tx.rollback().await.unwrap(); + + let db_tx = storage.transaction_ro().await.unwrap(); + assert!(db_tx.read_stream_events_after(0).await.unwrap().is_empty()); + } + + // Phase 2: the committed events are visible, in ascending id order. + let mut tx = storage.transaction_rw().await.unwrap(); + + for event in &events { + tx.append_stream_event(event).await.unwrap(); + } + + tx.commit().await.unwrap(); + + let db_tx = storage.transaction_ro().await.unwrap(); + let read_events = db_tx.read_stream_events_after(0).await.unwrap(); + + // 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() + ); + + if !read_events.is_empty() { + // Note: the events are returned in ascending id order and the ids are strictly monotonic. + let ids = read_events.iter().map(|(id, _)| *id).collect::>(); + assert!( + ids.windows(2).all(|ids| ids[0] < ids[1]), + "ids not monotonic: {ids:?}" + ); + + for ((_, read_event), expected_event) in read_events.iter().zip(events.iter()) { + assert_eq!(read_event, expected_event); + } + + // Note: reading after the last seen id returns nothing new. + assert!(db_tx.read_stream_events_after(ids[1]).await.unwrap().is_empty()); + assert_eq!( + db_tx.read_stream_events_after(ids[0]).await.unwrap().len(), + 1 + ); + } + + Ok(()) +} + fn get_all_substrings(s: &str) -> Vec<&str> { let mut substrings = Vec::new(); for i in 0..s.len() { diff --git a/api-server/storage-test-suite/src/podman.rs b/api-server/storage-test-suite/src/podman.rs index e597d63f01..8b86fa3249 100644 --- a/api-server/storage-test-suite/src/podman.rs +++ b/api-server/storage-test-suite/src/podman.rs @@ -15,6 +15,20 @@ use randomness::{RngExt as _, make_pseudo_rng}; +/// The container manager command to use: `podman` if available, otherwise `docker`. +/// +/// Note: the two CLIs are compatible for the commands used here. +fn container_command() -> &'static str { + // Note: `std::process::Command::new(...).status()` would print the lookup failure to stderr + // on some systems, hence the explicit probing of the PATH. + let podman_on_path = std::env::var_os("PATH").is_some_and(|paths| { + std::env::split_paths(&paths) + .any(|dir| dir.join("podman").is_file() || dir.join("podman.exe").is_file()) + }); + + if podman_on_path { "podman" } else { "docker" } +} + pub enum Container { PostgresFromDockerHub, } @@ -75,7 +89,7 @@ impl Podman { } pub fn run(&mut self) { - let mut command = std::process::Command::new("podman"); + let mut command = std::process::Command::new(container_command()); command.arg("run"); command.arg("--detach"); command.arg("--name"); @@ -97,7 +111,7 @@ impl Podman { } pub fn get_port_mapping(&self, container_port: u16) -> Option { - let mut command = std::process::Command::new("podman"); + let mut command = std::process::Command::new(container_command()); command.arg("port"); command.arg(&self.name); command.arg(format!("{}", container_port)); @@ -116,7 +130,7 @@ impl Podman { } pub fn stop(&mut self) { - let mut command = std::process::Command::new("podman"); + let mut command = std::process::Command::new(container_command()); command.arg("stop"); command.arg(&self.name); Self::run_command(command); @@ -128,7 +142,7 @@ impl Podman { self.is_running == Some(false), "The container must have been created and stopped before it can be restarted" ); - let mut command = std::process::Command::new("podman"); + let mut command = std::process::Command::new(container_command()); command.arg("start"); command.arg(&self.name); Self::run_command(command); @@ -137,7 +151,7 @@ impl Podman { /// Uses the command `podman logs` to print the logs of the container. pub fn print_logs(&mut self) { - let mut command = std::process::Command::new("podman"); + let mut command = std::process::Command::new(container_command()); command.arg("logs"); command.arg(&self.name); let output = Self::run_command(command); @@ -173,7 +187,8 @@ impl Podman { ); assert!( output.status.success(), - "Failed to run podman command: {:?}\n{}", + "Failed to run {} command: {:?}\n{}", + container_command(), command, String::from_utf8_lossy(&output.stderr) ); @@ -181,7 +196,7 @@ impl Podman { } fn remove_container(&mut self) { - let mut command = std::process::Command::new("podman"); + let mut command = std::process::Command::new(container_command()); command.arg("rm"); command.arg(&self.name); Self::run_command(command); diff --git a/api-server/web-server/Cargo.toml b/api-server/web-server/Cargo.toml index f46c6cecef..6a7da8ea58 100644 --- a/api-server/web-server/Cargo.toml +++ b/api-server/web-server/Cargo.toml @@ -22,6 +22,7 @@ axum.workspace = true async-trait.workspace = true ctor.workspace = true clap = { workspace = true, features = ["derive"] } +futures = { workspace = true, default-features = false } hex.workspace = true serde = { workspace = true, features = ["derive"] } serde_json.workspace = true diff --git a/api-server/web-server/src/api/mod.rs b/api-server/web-server/src/api/mod.rs index 348ef808f6..6f6e332d73 100644 --- a/api-server/web-server/src/api/mod.rs +++ b/api-server/web-server/src/api/mod.rs @@ -14,6 +14,7 @@ // limitations under the License. pub mod json_helpers; +pub mod stream; pub mod v2; use std::sync::Arc; diff --git a/api-server/web-server/src/api/stream.rs b/api-server/web-server/src/api/stream.rs new file mode 100644 index 0000000000..98ab262a1c --- /dev/null +++ b/api-server/web-server/src/api/stream.rs @@ -0,0 +1,218 @@ +// 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. + +//! The real-time event stream endpoint, exposing the stream events as Server-Sent Events. + +use std::{collections::BTreeSet, convert::Infallible, str::FromStr, sync::Arc, time::Duration}; + +use api_server_common::storage::storage_api::ApiServerStorage; +use api_server_common::streaming::{StreamEvent, StreamEventType, StreamEventTypeParseError}; +use axum::{ + extract::{Query, State}, + http::HeaderValue, + response::{ + IntoResponse, Response, + sse::{Event, KeepAlive, Sse}, + }, +}; +use futures::stream::Stream; +use serde::Deserialize; +use tokio::sync::broadcast; + +use crate::{ + ApiServerWebServerState, TxSubmitClient, + error::{ApiServerWebServerClientError, ApiServerWebServerError}, +}; + +/// The value of the `x-accel-buffering` header that prevents reverse proxies from buffering the +/// event stream. +const X_ACCEL_BUFFERING_VALUE: HeaderValue = HeaderValue::from_static("no"); + +/// The reconnection hint sent as the first Server-Sent Events frame. +const SSE_RETRY_INTERVAL: Duration = Duration::from_secs(3); + +#[derive(Debug, Deserialize)] +pub struct StreamQuery { + /// The comma-separated subset of the streamed event types, e.g. `types=block,reorg`. + /// Defaults to all event types. + types: Option, +} + +/// The set of stream event types a client wants to receive. +#[derive(Debug, Clone)] +pub struct StreamEventsFilter(BTreeSet); + +impl Default for StreamEventsFilter { + fn default() -> Self { + Self(StreamEventType::ALL.iter().copied().collect()) + } +} + +impl StreamEventsFilter { + /// Parse a comma-separated list of event type names. + pub fn parse(types: &str) -> Result { + let types = types + .split(',') + .map(StreamEventType::from_str) + .collect::, _>>()?; + Ok(Self(types)) + } + + fn allows(&self, event: &StreamEvent) -> bool { + self.0.contains(&event.event_type()) + } +} + +pub async fn stream_events< + T: ApiServerStorage + Send + Sync + 'static, + R: TxSubmitClient + Send + Sync + 'static, +>( + State(state): State, Arc>>, + Query(query): Query, +) -> Result { + let filter = match query.types.as_deref() { + Some(types) => StreamEventsFilter::parse(types).map_err(|_| { + ApiServerWebServerError::ClientError(ApiServerWebServerClientError::BadRequest) + })?, + None => StreamEventsFilter::default(), + }; + + let receiver = state.stream_events.channel.subscribe(); + let event_stream = sse_event_stream(receiver, filter); + + let sse = Sse::new(event_stream).keep_alive( + KeepAlive::new() + .interval(state.stream_events.config.keepalive_interval) + .text("keepalive"), + ); + + let mut response = sse.into_response(); + response.headers_mut().insert("x-accel-buffering", X_ACCEL_BUFFERING_VALUE); + + Ok(response) +} + +fn sse_event_stream( + receiver: broadcast::Receiver, + filter: StreamEventsFilter, +) -> impl Stream> { + // Note: the reconnection hint is sent as the first frame so that conforming clients + // reconnect with the intended interval; the flag below tracks whether it was sent. + futures::stream::unfold( + (receiver, filter, false), + |(mut receiver, filter, retry_sent)| async move { + if !retry_sent { + return Some(( + Ok(Event::default().retry(SSE_RETRY_INTERVAL)), + (receiver, filter, true), + )); + } + + loop { + match receiver.recv().await { + Ok(event) => { + // Note: events the client is not interested in are silently skipped. + if filter.allows(&event) { + return Some((Ok(sse_event(&event)), (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. + return Some((Ok(lag_event(skipped)), (receiver, filter, retry_sent))); + } + Err(broadcast::error::RecvError::Closed) => return None, + } + } + }, + ) +} + +fn sse_event(event: &StreamEvent) -> Event { + Event::default() + .event(event.event_name()) + .data(serde_json::to_string(event).unwrap_or_else(|_| { + // Note: the serialization of these events cannot fail in practice; the fallback + // keeps the SSE framing intact even if the event payload ever becomes unserializable. + "{\"error\":\"event serialization failed\"}".to_owned() + })) +} + +fn lag_event(skipped: u64) -> Event { + Event::default() + .event("lag") + .data(serde_json::json!({ "skipped": skipped }).to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + use api_server_common::streaming::{StreamEvent, TxOrigin}; + use common::{ + chain::block::timestamp::BlockTimestamp, + primitives::{BlockHeight, H256, Id}, + }; + + fn all_kinds_of_events() -> Vec { + vec![ + StreamEvent::TxSeen { + tx_id: Id::new(H256::from_low_u64_be(1)), + origin: TxOrigin::Local, + }, + StreamEvent::Block { + block_id: Id::new(H256::from_low_u64_be(2)), + height: BlockHeight::new(1), + timestamp: BlockTimestamp::from_int_seconds(1000), + tx_ids: vec![Id::new(H256::from_low_u64_be(1))], + }, + StreamEvent::Reorg { + common_ancestor_height: BlockHeight::new(0), + removed_block_ids: vec![Id::new(H256::from_low_u64_be(3))], + new_tip_height: BlockHeight::new(1), + }, + ] + } + + #[test] + fn filter_parsing() { + let filter = StreamEventsFilter::parse("block, reorg").unwrap(); + assert_eq!( + filter.0, + [StreamEventType::Block, StreamEventType::Reorg].into() + ); + + assert!(StreamEventsFilter::parse("").is_err()); + assert!(StreamEventsFilter::parse("block,bogus").is_err()); + + // Note: defaults to all types. + assert_eq!( + StreamEventsFilter::default().0, + StreamEventType::ALL.iter().copied().collect() + ); + } + + #[test] + fn filter_matching() { + let events = all_kinds_of_events(); + + let all = StreamEventsFilter::default(); + assert!(events.iter().all(|event| all.allows(event))); + + let blocks_only = StreamEventsFilter::parse("block").unwrap(); + assert!(!blocks_only.allows(&events[0])); + assert!(blocks_only.allows(&events[1])); + assert!(!blocks_only.allows(&events[2])); + } +} diff --git a/api-server/web-server/src/api/v2.rs b/api-server/web-server/src/api/v2.rs index e6e5c0f62f..c8d99e7916 100644 --- a/api-server/web-server/src/api/v2.rs +++ b/api-server/web-server/src/api/v2.rs @@ -134,6 +134,9 @@ pub fn routes< .route("/order", get(orders)) .route("/order/:id", get(order)) .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)) } async fn forbidden_request() -> Result<(), ApiServerWebServerError> { diff --git a/api-server/web-server/src/config.rs b/api-server/web-server/src/config.rs index b254a9eabe..6d22facf70 100644 --- a/api-server/web-server/src/config.rs +++ b/api-server/web-server/src/config.rs @@ -22,6 +22,8 @@ use tokio::net::TcpListener; use utils::{app_version_with_git_info, clap_utils}; use utils_networking::NetworkAddressWithPort; +use crate::streaming; + const LISTEN_ADDRESS: &str = "127.0.0.1:3000"; #[derive(Debug, Parser)] @@ -64,6 +66,20 @@ pub struct ApiServerWebServerConfig { /// RPC password (either provide a username and password, or use a cookie file. You cannot use both) #[clap(long)] pub node_rpc_password: Option, + + /// The maximum number of real-time stream events buffered per connected client; a client that + /// falls further behind receives a `lag` advisory event instead of the missed events. + #[clap(long, default_value_t = streaming::DEFAULT_STREAM_EVENTS_BROADCAST_CAPACITY)] + pub stream_events_broadcast_capacity: usize, + + /// The interval in seconds between real-time stream event polls; used as a safety net for + /// missed database notifications. + #[clap(long, default_value_t = streaming::DEFAULT_STREAM_EVENTS_POLL_INTERVAL.as_secs())] + pub stream_events_poll_interval_secs: u64, + + /// The interval in seconds between keepalive comments sent to connected stream clients. + #[clap(long, default_value_t = streaming::DEFAULT_STREAM_EVENTS_KEEPALIVE_INTERVAL.as_secs())] + pub stream_events_keepalive_interval_secs: u64, } #[derive(Clone, Debug, Parser)] diff --git a/api-server/web-server/src/lib.rs b/api-server/web-server/src/lib.rs index d66113e847..cb7ba5851e 100644 --- a/api-server/web-server/src/lib.rs +++ b/api-server/web-server/src/lib.rs @@ -16,8 +16,10 @@ pub mod api; pub mod config; pub mod error; +pub mod streaming; pub use error::ApiServerWebServerError; +pub use streaming::{StreamEventsHandle, StreamingConfig}; use common::{ chain::{ChainConfig, SignedTransaction}, @@ -60,4 +62,7 @@ pub struct ApiServerWebServerState { pub rpc: R, pub cached_values: Arc, pub time_getter: TimeGetter, + /// The channel of real-time stream events, fed by the database event pump and the node + /// mempool bridge, and consumed by the `/api/v2/stream` endpoint. + pub stream_events: StreamEventsHandle, } diff --git a/api-server/web-server/src/main.rs b/api-server/web-server/src/main.rs index f89e625520..e211bf5e7a 100644 --- a/api-server/web-server/src/main.rs +++ b/api-server/web-server/src/main.rs @@ -17,10 +17,20 @@ mod api; mod config; mod error; -use api_server_common::storage::impls::postgres::TransactionalApiServerPostgresStorage; +use api_server_common::storage::impls::postgres::{ + PostgresStreamEventSource, TransactionalApiServerPostgresStorage, +}; +use api_server_common::streaming::StreamEventsChannel; use api_web_server::{ - ApiServerWebServerState, CachedValues, TxSubmitClient, api::web_server, + ApiServerWebServerState, + CachedValues, + StreamEventsHandle, + TxSubmitClient, + api::web_server, config::ApiServerWebServerConfig, + // Note: `streaming` is imported into the crate root so that the shared `config` module can + // refer to it as `crate::streaming` in both the library and the binary. + streaming, }; use clap::Parser; use common::{ @@ -48,17 +58,46 @@ async fn main() -> Result<(), ApiServerWebServerInitError> { let chain_type: ChainType = args.network.into(); let chain_config = Arc::new(Builder::new(chain_type).build()); - let storage = TransactionalApiServerPostgresStorage::new( - &args.postgres_config.postgres_host, - args.postgres_config.postgres_port, - &args.postgres_config.postgres_user, - args.postgres_config.postgres_password.as_deref(), - args.postgres_config.postgres_database.as_deref(), - args.postgres_config.postgres_max_connections, - chain_config.clone(), - ) - .await - .map_err(ApiServerWebServerInitError::PostgresConnectionError)?; + let storage = Arc::new( + TransactionalApiServerPostgresStorage::new( + &args.postgres_config.postgres_host, + args.postgres_config.postgres_port, + &args.postgres_config.postgres_user, + args.postgres_config.postgres_password.as_deref(), + args.postgres_config.postgres_database.as_deref(), + args.postgres_config.postgres_max_connections, + chain_config.clone(), + ) + .await + .map_err(ApiServerWebServerInitError::PostgresConnectionError)?, + ); + + let stream_events = { + // Note: the values are clamped so that operator error cannot crash or wedge the server. + 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), + ), + }; + StreamEventsHandle::new(channel, config) + }; + + // Note: the event pump needs its own database connection for the LISTEN/NOTIFY wakeups; the + // periodic poll is the fallback for missed notifications. + let event_listener = storage + .new_event_listener() + .await + .map_err(ApiServerWebServerInitError::PostgresConnectionError)?; + let event_source = PostgresStreamEventSource::new( + Arc::clone(&storage), + event_listener, + std::time::Duration::from_secs(args.stream_events_poll_interval_secs.max(1)), + ); + tokio::spawn(streaming::run_database_event_pump( + event_source, + stream_events.clone(), + )); let rpc_client = { let rpc_auth = match ( @@ -91,14 +130,24 @@ async fn main() -> Result<(), ApiServerWebServerInitError> { .map_err(ApiServerWebServerInitError::RpcError)? }; + let rpc_client = Arc::new(rpc_client); + + // Note: the mempool events arrive over the node's WebSocket connection and are bridged into + // the stream event channel; the subscription is re-established after connection loss. + tokio::spawn(streaming::run_mempool_bridge( + Arc::clone(&rpc_client), + stream_events.clone(), + )); + let state = ApiServerWebServerState { - db: Arc::new(storage), + db: storage, chain_config, - rpc: Arc::new(rpc_client), + rpc: rpc_client, cached_values: Arc::new(CachedValues { feerate_points: RwLock::new((Time::from_secs_since_epoch(0), vec![])), }), time_getter: Default::default(), + stream_events, }; web_server( diff --git a/api-server/web-server/src/streaming.rs b/api-server/web-server/src/streaming.rs new file mode 100644 index 0000000000..6942033596 --- /dev/null +++ b/api-server/web-server/src/streaming.rs @@ -0,0 +1,207 @@ +// 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. + +//! Real-time event streaming support: the stream event channel handle used by the web server +//! state, the database event pump, and the bridge from the node's mempool events. + +use std::sync::Arc; +use std::time::Duration; + +use api_server_common::streaming::{ + StreamEvent, StreamEventSource, StreamEventsChannel, TxOrigin, run_event_pump, +}; +use mempool::rpc::MempoolRpcClient; +use mempool::rpc_event::{RpcEvent, RpcTxOrigin}; +use node_comm::rpc_client::NodeRpcClient; + +/// How many real-time stream events may be buffered per connected client before the client +/// receives a `lag` advisory event instead of the missed events. +pub const DEFAULT_STREAM_EVENTS_BROADCAST_CAPACITY: usize = 1024; + +/// How often the database event pump polls for new events as a safety net for missed +/// notifications. +pub const DEFAULT_STREAM_EVENTS_POLL_INTERVAL: Duration = Duration::from_secs(30); + +/// How often a keepalive comment is sent to connected stream clients. +pub const DEFAULT_STREAM_EVENTS_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30); + +/// How long to wait before re-attempting the subscription to the node's mempool events after it +/// has been lost. +const MEMPOOL_RESUBSCRIBE_DELAY: Duration = Duration::from_secs(1); + +/// Streaming-related configuration of the web server. +#[derive(Debug, Clone)] +pub struct StreamingConfig { + pub keepalive_interval: Duration, +} + +impl Default for StreamingConfig { + fn default() -> Self { + Self { + keepalive_interval: DEFAULT_STREAM_EVENTS_KEEPALIVE_INTERVAL, + } + } +} + +/// The stream event channel and its configuration, as stored in the web server state. +#[derive(Clone)] +pub struct StreamEventsHandle { + pub channel: StreamEventsChannel, + pub config: StreamingConfig, +} + +impl Default for StreamEventsHandle { + fn default() -> Self { + Self { + channel: StreamEventsChannel::new(DEFAULT_STREAM_EVENTS_BROADCAST_CAPACITY), + config: StreamingConfig::default(), + } + } +} + +impl StreamEventsHandle { + pub fn new(channel: StreamEventsChannel, config: StreamingConfig) -> Self { + Self { channel, config } + } +} + +/// Run the database event pump: forward the stream events committed by the scanner into the +/// broadcast channel of the given handle. +pub async fn run_database_event_pump(source: impl StreamEventSource, handle: StreamEventsHandle) { + run_event_pump(source, handle.channel, 0).await; +} + +/// Map a node mempool event into a stream event. +/// +/// Only successfully processed transactions are of interest; the new tip events and failed +/// transactions are dropped. +fn map_rpc_event_to_tx_seen(event: RpcEvent) -> Option { + match event { + RpcEvent::NewTip { .. } => None, + RpcEvent::TransactionProcessed { + tx_id, + origin, + successful, + .. + } if successful => { + let origin = match origin { + RpcTxOrigin::Local { .. } => TxOrigin::Local, + RpcTxOrigin::Remote { .. } => TxOrigin::Remote, + }; + Some(StreamEvent::TxSeen { tx_id, origin }) + } + RpcEvent::TransactionProcessed { .. } => None, + } +} + +/// Bridge the node's mempool events into the stream event channel. +/// +/// The WebSocket subscription is re-established after connection loss; only the successfully +/// processed transactions are forwarded as `TxSeen` events. Note that the events are not +/// hydrated here: a failed hydration must never block the stream, so the stream carries only the +/// transaction ids and the clients are expected to fetch the details through the REST endpoints. +pub async fn run_mempool_bridge(rpc: Arc, handle: StreamEventsHandle) { + loop { + match MempoolRpcClient::subscribe_to_events(rpc.ws_client()).await { + Ok(subscription) => { + logging::log::info!("Subscribed to node mempool events"); + let mut subscription = subscription; + while let Some(event) = subscription.next().await { + match event { + Ok(event) => { + if let Some(stream_event) = map_rpc_event_to_tx_seen(event) { + // Note: sending fails only when there are no subscribers. + let _ = handle.channel.send(stream_event); + } + } + Err(err) => { + logging::log::warn!("Node mempool subscription error: {err}"); + break; + } + } + } + logging::log::warn!("Node mempool subscription closed; re-subscribing"); + } + Err(err) => { + logging::log::error!("Failed to subscribe to node mempool events: {err}"); + } + } + tokio::time::sleep(MEMPOOL_RESUBSCRIBE_DELAY).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use common::primitives::{H256, Id}; + + fn tx_processed_event(successful: bool, origin: RpcTxOrigin) -> RpcEvent { + RpcEvent::TransactionProcessed { + tx_id: Id::new(H256::from_low_u64_be(1)), + origin, + relay: mempool::rpc_event::RpcTxRelayPolicy::DoRelay, + successful, + } + } + + #[test] + fn successful_transaction_maps_to_tx_seen() { + let origin = RpcTxOrigin::Local { + origin: mempool::rpc_event::RpcLocalTxOrigin::Mempool, + }; + let stream_event = map_rpc_event_to_tx_seen(tx_processed_event(true, origin)) + .expect("must map to an event"); + + match stream_event { + StreamEvent::TxSeen { tx_id, origin } => { + assert_eq!(tx_id, Id::new(H256::from_low_u64_be(1))); + assert_eq!(origin, TxOrigin::Local); + } + _ => panic!("unexpected event type"), + } + } + + #[test] + fn remote_origin_is_preserved() { + let origin = RpcTxOrigin::Remote { + peer_id: node_comm::node_traits::PeerId::from_u64(1), + }; + let stream_event = map_rpc_event_to_tx_seen(tx_processed_event(true, origin)) + .expect("must map to an event"); + + match stream_event { + StreamEvent::TxSeen { + origin: TxOrigin::Remote, + .. + } => {} + _ => panic!("unexpected event type"), + } + } + + #[test] + fn failed_transactions_and_new_tips_are_dropped() { + let origin = RpcTxOrigin::Local { + origin: mempool::rpc_event::RpcLocalTxOrigin::Mempool, + }; + assert!(map_rpc_event_to_tx_seen(tx_processed_event(false, origin)).is_none()); + assert!( + map_rpc_event_to_tx_seen(RpcEvent::NewTip { + id: Id::new(H256::from_low_u64_be(2)), + height: common::primitives::BlockHeight::new(1), + }) + .is_none() + ); + } +} diff --git a/blockprod/src/detail/tests/produce_block/tx_selection_by_deps.rs b/blockprod/src/detail/tests/produce_block/tx_selection_by_deps.rs index dfca2c03d8..947966ae5b 100644 --- a/blockprod/src/detail/tests/produce_block/tx_selection_by_deps.rs +++ b/blockprod/src/detail/tests/produce_block/tx_selection_by_deps.rs @@ -1339,8 +1339,7 @@ async fn assert_fees( .chainstate .call(|cs| { let tip = cs.get_best_block_id().unwrap(); - let tip_index = cs.get_gen_block_index_for_persisted_block(&tip).unwrap().unwrap(); - tip_index + cs.get_gen_block_index_for_persisted_block(&tip).unwrap().unwrap() }) .await .unwrap(); diff --git a/chainstate/constraints-value-accumulator/src/accumulated_fee.rs b/chainstate/constraints-value-accumulator/src/accumulated_fee.rs index cdd7264e76..8be4192a4b 100644 --- a/chainstate/constraints-value-accumulator/src/accumulated_fee.rs +++ b/chainstate/constraints-value-accumulator/src/accumulated_fee.rs @@ -28,6 +28,12 @@ pub struct AccumulatedFee { timelock_constrained: BTreeMap, } +impl Default for AccumulatedFee { + fn default() -> Self { + Self::new() + } +} + impl AccumulatedFee { pub fn new() -> Self { Self { diff --git a/chainstate/constraints-value-accumulator/src/constraints_accumulator.rs b/chainstate/constraints-value-accumulator/src/constraints_accumulator.rs index dace67c6bf..605e4f6ed0 100644 --- a/chainstate/constraints-value-accumulator/src/constraints_accumulator.rs +++ b/chainstate/constraints-value-accumulator/src/constraints_accumulator.rs @@ -40,6 +40,12 @@ pub struct ConstrainedValueAccumulator { timelock_constrained: BTreeMap, } +impl Default for ConstrainedValueAccumulator { + fn default() -> Self { + Self::new() + } +} + impl ConstrainedValueAccumulator { pub fn new() -> Self { Self { diff --git a/chainstate/src/detail/orphan_blocks/pool.rs b/chainstate/src/detail/orphan_blocks/pool.rs index ba165d0998..9855cda9cd 100644 --- a/chainstate/src/detail/orphan_blocks/pool.rs +++ b/chainstate/src/detail/orphan_blocks/pool.rs @@ -139,14 +139,12 @@ impl OrphanBlocksPool { // after we get all the blocks that have the same prev, we drop them from the pool res.iter().for_each(|blk| self.drop_block(&blk.get_id())); // after dropping everything, this is expected to be the only Rc left - let res = res - .drain(..) + res.drain(..) .map(|blk| { Rc::try_unwrap(blk) .expect("There cannot be more than one copy of the Rc. This is unexpected.") }) - .collect(); - res + .collect() } } diff --git a/chainstate/test-framework/src/transaction_builder.rs b/chainstate/test-framework/src/transaction_builder.rs index 44ca5876b4..692a2698af 100644 --- a/chainstate/test-framework/src/transaction_builder.rs +++ b/chainstate/test-framework/src/transaction_builder.rs @@ -31,6 +31,12 @@ pub struct TransactionBuilder { witnesses: Vec, } +impl Default for TransactionBuilder { + fn default() -> Self { + Self::new() + } +} + impl TransactionBuilder { pub fn new() -> Self { Self { diff --git a/chainstate/test-framework/src/utils.rs b/chainstate/test-framework/src/utils.rs index 86a9fa7a6f..55e96482e1 100644 --- a/chainstate/test-framework/src/utils.rs +++ b/chainstate/test-framework/src/utils.rs @@ -413,8 +413,7 @@ pub fn sign_witnesses( ) .unwrap(); - let witnesses = tx - .inputs() + tx.inputs() .iter() .enumerate() .map(|(idx, input)| { @@ -431,9 +430,7 @@ pub fn sign_witnesses( ) .unwrap() }) - .collect(); - - witnesses + .collect() } pub fn find_create_pool_tx_in_genesis(genesis: &Genesis, pool_id: &PoolId) -> Option { diff --git a/chainstate/test-suite/src/tests/orders_tests.rs b/chainstate/test-suite/src/tests/orders_tests.rs index 6ead66f4d1..bd78017c55 100644 --- a/chainstate/test-suite/src/tests/orders_tests.rs +++ b/chainstate/test-suite/src/tests/orders_tests.rs @@ -1049,10 +1049,7 @@ fn fill_order_check_storage(#[case] seed: Seed, #[case] version: OrdersVersion) / ask_amount.into_atoms(); let filled2 = (give_amount.into_atoms() * left_to_fill.into_atoms()) / ask_amount.into_atoms(); - let remainder = (give_amount - Amount::from_atoms(filled1 + filled2)) - .unwrap() - .as_non_zero(); - remainder + (give_amount - Amount::from_atoms(filled1 + filled2)).unwrap().as_non_zero() } }; diff --git a/chainstate/test-suite/src/tests/processing_tests.rs b/chainstate/test-suite/src/tests/processing_tests.rs index df6c3a56d4..8210abff72 100644 --- a/chainstate/test-suite/src/tests/processing_tests.rs +++ b/chainstate/test-suite/src/tests/processing_tests.rs @@ -1008,8 +1008,7 @@ fn read_block_reward_from_storage(#[case] seed: Seed) { .expect("Unexpected conversion error"), consensus::MiningResult::Success ); - let valid_block = Block::new_from_header(block_header, valid_block.body().clone()).unwrap(); - valid_block + Block::new_from_header(block_header, valid_block.body().clone()).unwrap() }; tf.process_block(block, BlockSource::Local).unwrap(); diff --git a/chainstate/types/src/block_status.rs b/chainstate/types/src/block_status.rs index 97cfd456f7..686063c991 100644 --- a/chainstate/types/src/block_status.rs +++ b/chainstate/types/src/block_status.rs @@ -46,8 +46,7 @@ impl BlockStatus { Self(0) } - /// Advance the last successful validation stage to the specified value. - /// Note that the stage can only be advanced one step at a time. + /// Advance the last successful validation stage to the specified value. /// Note that the stage can only be advanced one step at a time. pub fn advance_validation_stage_to(&mut self, new_stage: BlockValidationStage) { assert!(self.last_valid_stage().next() == Some(new_stage)); self.set_last_valid_stage(new_stage); @@ -149,6 +148,12 @@ impl BlockStatus { } } +impl Default for BlockStatus { + fn default() -> Self { + Self::new() + } +} + impl std::fmt::Display for BlockStatus { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "BlockStatus({:#b})", self.0) diff --git a/common/src/chain/partially_signed_transaction/additional_info.rs b/common/src/chain/partially_signed_transaction/additional_info.rs index 303be49f14..f8aaa10579 100644 --- a/common/src/chain/partially_signed_transaction/additional_info.rs +++ b/common/src/chain/partially_signed_transaction/additional_info.rs @@ -77,7 +77,6 @@ impl TxAdditionalInfo { order_info: BTreeMap::new(), } } - pub fn with_pool_info(mut self, pool_id: PoolId, info: PoolAdditionalInfo) -> Self { self.pool_info.insert(pool_id, info); self @@ -161,3 +160,9 @@ impl OutputValuesHolder for TxAdditionalInfo { .flat_map(|(_, order_info)| order_info.output_values_iter()) } } + +impl Default for TxAdditionalInfo { + fn default() -> Self { + Self::new() + } +} diff --git a/mempool/src/pool/orphans/test.rs b/mempool/src/pool/orphans/test.rs index c26ecf83e6..84375fbf8f 100644 --- a/mempool/src/pool/orphans/test.rs +++ b/mempool/src/pool/orphans/test.rs @@ -24,7 +24,7 @@ use common::{ }, primitives::{Amount, H256}, }; -use test_utils::random::{Rng, RngExt as _, Seed, make_seedable_rng}; +use test_utils::random::{Rng, Seed, make_seedable_rng}; use super::*; diff --git a/networking/src/transport/impls/channel.rs b/networking/src/transport/impls/channel.rs index e1b7d15d81..7831479723 100644 --- a/networking/src/transport/impls/channel.rs +++ b/networking/src/transport/impls/channel.rs @@ -328,6 +328,12 @@ pub enum MpscChannelTransportError { }, } +impl Default for MpscChannelTransport { + fn default() -> Self { + Self::new() + } +} + #[cfg(test)] mod tests { use std::net::SocketAddrV4; diff --git a/networking/src/transport/impls/stream_adapter/identity.rs b/networking/src/transport/impls/stream_adapter/identity.rs index 5782e74d54..f933e9eccb 100644 --- a/networking/src/transport/impls/stream_adapter/identity.rs +++ b/networking/src/transport/impls/stream_adapter/identity.rs @@ -43,3 +43,9 @@ impl StreamAdapter for Identit Box::pin(ready(Ok(base))) } } + +impl Default for IdentityStreamAdapter { + fn default() -> Self { + Self::new() + } +} diff --git a/networking/src/transport/impls/tcp.rs b/networking/src/transport/impls/tcp.rs index f803638981..9481a81790 100644 --- a/networking/src/transport/impls/tcp.rs +++ b/networking/src/transport/impls/tcp.rs @@ -149,6 +149,12 @@ impl ConnectedSocketInfo for TcpTransportStream { } } +impl Default for TcpTransportSocket { + fn default() -> Self { + Self::new() + } +} + #[cfg(test)] mod tests { use serialization::Encode; diff --git a/node-gui/backend/src/messages.rs b/node-gui/backend/src/messages.rs index d6c08d1c0d..66f8546d0c 100644 --- a/node-gui/backend/src/messages.rs +++ b/node-gui/backend/src/messages.rs @@ -44,6 +44,12 @@ pub struct WalletId(u64); static NEXT_WALLET_ID: AtomicU64 = AtomicU64::new(0); +impl Default for WalletId { + fn default() -> Self { + Self::new() + } +} + impl WalletId { pub fn new() -> Self { Self(NEXT_WALLET_ID.fetch_add(1, Ordering::Relaxed)) diff --git a/node-gui/src/main_window/main_menu.rs b/node-gui/src/main_window/main_menu.rs index b77ba76e14..213a4abb9e 100644 --- a/node-gui/src/main_window/main_menu.rs +++ b/node-gui/src/main_window/main_menu.rs @@ -100,7 +100,7 @@ fn menu_item(label: &str, msg: MenuMessage) -> Item<'_, MenuMessage, Theme, iced } fn make_menu_file<'a>(wallet_mode: WalletMode) -> Item<'a, MenuMessage, Theme, iced::Renderer> { - let root = Item::with_menu( + Item::with_menu( labeled_button("File", MenuMessage::NoOp), Menu::new(match wallet_mode { WalletMode::Hot => { @@ -199,7 +199,5 @@ fn make_menu_file<'a>(wallet_mode: WalletMode) -> Item<'a, MenuMessage, Theme, i } }) .width(300), - ); - - root + ) } diff --git a/orders-accounting/src/data.rs b/orders-accounting/src/data.rs index e050fd4daa..d785881d9d 100644 --- a/orders-accounting/src/data.rs +++ b/orders-accounting/src/data.rs @@ -98,6 +98,12 @@ pub struct OrdersAccountingData { pub give_balances: BTreeMap, } +impl Default for OrdersAccountingData { + fn default() -> Self { + Self::new() + } +} + impl OrdersAccountingData { pub fn new() -> Self { Self { @@ -115,6 +121,12 @@ pub struct OrdersAccountingDeltaData { pub(crate) give_balances: DeltaAmountCollection, } +impl Default for OrdersAccountingDeltaData { + fn default() -> Self { + Self::new() + } +} + impl OrdersAccountingDeltaData { pub fn merge_with_delta( &mut self, @@ -153,6 +165,12 @@ pub struct OrdersAccountingDeltaUndoData { pub(crate) give_balances: DeltaAmountCollection, } +impl Default for OrdersAccountingDeltaUndoData { + fn default() -> Self { + Self::new() + } +} + impl OrdersAccountingDeltaUndoData { pub fn new() -> Self { Self { diff --git a/orders-accounting/src/storage/in_memory.rs b/orders-accounting/src/storage/in_memory.rs index 9d1abf749a..aad8fc0b23 100644 --- a/orders-accounting/src/storage/in_memory.rs +++ b/orders-accounting/src/storage/in_memory.rs @@ -29,6 +29,12 @@ pub struct InMemoryOrdersAccounting { give_balances: BTreeMap, } +impl Default for InMemoryOrdersAccounting { + fn default() -> Self { + Self::new() + } +} + impl InMemoryOrdersAccounting { pub fn new() -> Self { Self { diff --git a/p2p/src/peer_manager/mod.rs b/p2p/src/peer_manager/mod.rs index b5de828e92..4b12c69905 100644 --- a/p2p/src/peer_manager/mod.rs +++ b/p2p/src/peer_manager/mod.rs @@ -2273,9 +2273,7 @@ where // should be loaded. // Note: the check for reachability is a protection against a misconfigured dns seed, // which may return bogus addresses. - let peerdb_has_no_reachable_addresses = self.peerdb.reachable_addresses().next().is_none(); - - peerdb_has_no_reachable_addresses + self.peerdb.reachable_addresses().next().is_none() } fn load_predefined_addresses(&mut self) { diff --git a/p2p/src/sync/sync_status.rs b/p2p/src/sync/sync_status.rs index 37e5168eda..a5d962e775 100644 --- a/p2p/src/sync/sync_status.rs +++ b/p2p/src/sync/sync_status.rs @@ -22,6 +22,12 @@ pub struct PeerBlockSyncStatus { pub expecting_blocks_since: Option