Serve pending transactions through the v2 REST API

api-server/CHANGELOG.md

@@ -7,6 +7,8 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/
 ## [Unreleased]
 
 ### Added
+- Pending transactions are now served through the regular REST endpoints: `GET /v2/transaction/{id}` falls back to the mempool of the connected node when the transaction is not confirmed yet, and a new `GET /v2/mempool/transactions` endpoint lists the pending transactions (paginated with `offset`/`items`, with an optional `order=dependency` parameter that lists transactions after the transactions they depend on).\
+  The responses have the same shape as the confirmed ones, except that for a pending transaction the `block_id`/`timestamp`/`confirmations` fields are `null` (the values are not applicable until the transaction is confirmed), the `fee` field is omitted, and the spent utxos of the inputs are not populated. The decimals of a transferred token are resolved from the api-server storage; if the token is not indexed yet, i.e. its issuance may be pending in the mempool itself, the decimals are taken from the issuing transaction of the mempool listing (fetched only in that case), and a token that cannot be resolved at all is rendered with zero decimals. The endpoints proxy the mempool of the connected node, so the data reflects the view of that node and disappears once the transactions are confirmed, evicted or reorganized away. Unlike the other GET endpoints, the cost of `GET /v2/mempool/transactions` is proportional to the size of the node's mempool (which is capped by the node configuration) rather than the size of the requested page, and `GET /v2/transaction/{id}` pays that cost too when it has to resolve the decimals of a token that is not indexed yet.
 - 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-max-subscribers`, `--stream-events-poll-interval-secs`, `--stream-events-keepalive-interval-secs`.

api-server/stack-test-suite/tests/common/mod.rs

@@ -20,10 +20,25 @@
 
 #![allow(dead_code)]
 
-use api_web_server::TxSubmitClient;
-use common::chain::SignedTransaction;
+use api_server_common::storage::impls::in_memory::transactional::TransactionalApiServerInMemoryStorage;
+use api_web_server::{
+    ApiServerWebServerState, CachedValues, MempoolQueryClient, TxSubmitClient, api::web_server,
+};
+use common::{
+    chain::{SignedTransaction, Transaction, config::create_unit_test_config},
+    primitives::{Id, Idable, time::get_time},
+};
+use hex::ToHex;
 use mempool::FeeRate;
 use node_comm::rpc_client::NodeRpcError;
+use serialization::hex_encoded::HexEncoded;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::Duration;
+use tokio::sync::RwLock;
+
+/// The time to wait for the web server to respond to the barrier request.
+const BARRIER_TIMEOUT: Duration = Duration::from_secs(30);
 
 /// A no-op RPC client for the web server state under test.
 pub struct DummyRPC {}
@@ -39,6 +54,221 @@ impl TxSubmitClient for DummyRPC {
     }
 }
 
+#[async_trait::async_trait]
+impl MempoolQueryClient for DummyRPC {
+    async fn mempool_transaction(
+        &self,
+        _: Id<Transaction>,
+    ) -> Result<Option<SignedTransaction>, NodeRpcError> {
+        Ok(None)
+    }
+
+    async fn mempool_transactions(&self) -> Result<Vec<SignedTransaction>, NodeRpcError> {
+        Ok(vec![])
+    }
+}
+
+/// An RPC client mock with an in-memory mempool.
+///
+/// Transactions submitted through [`TxSubmitClient::submit_tx`] are added to the mock
+/// mempool in the order of submission, imitating the insertion order of the mempool
+/// of a node. The mock does not validate the transactions (e.g. it does not check
+/// that the spent outputs exist), just like a node mempool accepts chain of unconfirmed
+/// transactions.
+#[derive(Default)]
+pub struct MempoolRPC {
+    mempool: RwLock<Vec<SignedTransaction>>,
+    fetch_count: AtomicUsize,
+}
+
+impl MempoolRPC {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// The number of times the mempool listing has been fetched from this client.
+    pub fn fetch_count(&self) -> usize {
+        self.fetch_count.load(Ordering::Relaxed)
+    }
+}
+
+#[async_trait::async_trait]
+impl TxSubmitClient for MempoolRPC {
+    async fn submit_tx(&self, tx: SignedTransaction) -> Result<(), NodeRpcError> {
+        self.mempool.write().await.push(tx);
+        Ok(())
+    }
+
+    async fn get_feerate_points(&self) -> Result<Vec<(usize, FeeRate)>, NodeRpcError> {
+        Ok(vec![])
+    }
+}
+
+#[async_trait::async_trait]
+impl MempoolQueryClient for MempoolRPC {
+    async fn mempool_transaction(
+        &self,
+        tx_id: Id<Transaction>,
+    ) -> Result<Option<SignedTransaction>, NodeRpcError> {
+        Ok(self
+            .mempool
+            .read()
+            .await
+            .iter()
+            .find(|tx| tx.transaction().get_id() == tx_id)
+            .cloned())
+    }
+
+    async fn mempool_transactions(&self) -> Result<Vec<SignedTransaction>, NodeRpcError> {
+        self.fetch_count.fetch_add(1, Ordering::Relaxed);
+        Ok(self.mempool.read().await.clone())
+    }
+}
+
+/// Extract the panic message from a panic payload, falling back to a
+/// placeholder if the payload is not a string.
+fn panic_payload_message(payload: Box<dyn std::any::Any + Send>) -> String {
+    payload
+        .downcast_ref::<&str>()
+        .map(|s| (*s).to_string())
+        .or_else(|| payload.downcast_ref::<String>().cloned())
+        .unwrap_or_else(|| "non-string panic payload".to_string())
+}
+
+/// The barrier request ensuring that the spawned web server is up: given that the
+/// listener port is open, the request to the `url` blocks until a response is made
+/// (by the web server, which takes the listener over), and the response is returned
+/// to the caller. The request is bounded by a timeout, so that a hung server task
+/// does not hang the test.
+///
+/// On any failure, the `task` running the web server is aborted and awaited, and the
+/// test panics with the failure context, including the outcome of the task (with the
+/// actual panic message, if the task panicked).
+pub async fn wait_for_web_server(
+    task: &mut tokio::task::JoinHandle<()>,
+    addr: std::net::SocketAddr,
+    url: &str,
+) -> reqwest::Response {
+    let request = reqwest::get(format!("http://{}:{}{url}", addr.ip(), addr.port()));
+
+    let err = match tokio::time::timeout(BARRIER_TIMEOUT, request).await {
+        Ok(Ok(response)) => return response,
+        Ok(Err(err)) => format!("request failed: {err}"),
+        Err(_timed_out) => format!("the request timed out after {BARRIER_TIMEOUT:?}"),
+    };
+
+    task.abort();
+    let join_result = match tokio::time::timeout(BARRIER_TIMEOUT, task).await {
+        Ok(join_result) => join_result,
+        Err(_timed_out) => {
+            panic!("the aborted server task did not terminate within {BARRIER_TIMEOUT:?}")
+        }
+    };
+    let outcome = match join_result {
+        Ok(()) => "the task finished".to_string(),
+        Err(join_err) if join_err.is_cancelled() => "the task was aborted".to_string(),
+        Err(join_err) => {
+            let message = panic_payload_message(join_err.into_panic());
+            format!("the task panicked: {message}")
+        }
+    };
+    panic!("the web server died before responding on {addr}: {err}; {outcome}");
+}
+
+/// Spawn the web server backed by the [`MempoolRPC`] client and an empty in-memory
+/// api-server storage.
+///
+/// Imitating the `spawn_webserver` helper of the test binaries, the returned response
+/// is the response to the `url` request, which doubles as the barrier ensuring that
+/// the server is up before the test proceeds.
+pub async fn spawn_webserver_with_mempool(
+    url: &str,
+) -> (
+    tokio::task::JoinHandle<()>,
+    reqwest::Response,
+    Arc<MempoolRPC>,
+    std::net::SocketAddr,
+) {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    let rpc = Arc::new(MempoolRPC::new());
+
+    let mut task = tokio::spawn({
+        let rpc = std::sync::Arc::clone(&rpc);
+        async move {
+            let web_server_state = {
+                let chain_config = Arc::new(create_unit_test_config());
+                let storage = TransactionalApiServerInMemoryStorage::new(&chain_config);
+
+                ApiServerWebServerState {
+                    db: Arc::new(storage),
+                    chain_config: Arc::clone(&chain_config),
+                    rpc,
+                    cached_values: Arc::new(CachedValues {
+                        feerate_points: std::sync::RwLock::new((get_time(), vec![])),
+                    }),
+                    time_getter: Default::default(),
+                    stream_events: Default::default(),
+                }
+            };
+
+            web_server(listener, web_server_state, true).await.unwrap();
+        }
+    });
+
+    let response = wait_for_web_server(&mut task, addr, url).await;
+
+    (task, response, rpc, addr)
+}
+
+/// Abort the task and observe its outcome: tolerated if cancelled or completed, panics
+/// with the actual panic message otherwise.
+///
+/// Generic over the task output.
+pub async fn shutdown_task<T: Send + 'static>(handle: tokio::task::JoinHandle<T>) {
+    handle.abort();
+    match handle.await {
+        Ok(_) => {}
+        Err(err) if err.is_cancelled() => {}
+        Err(err) => {
+            let message = panic_payload_message(err.into_panic());
+            panic!("task panicked: {message}");
+        }
+    }
+}
+
+/// Submit the transaction through the POST endpoint, imitating a user of the
+/// api-server, and return the hex-encoded id of the submitted transaction.
+pub async fn submit_transaction(addr: std::net::SocketAddr, tx: SignedTransaction) -> String {
+    let tx_id = tx.transaction().get_id().to_hash().encode_hex::<String>();
+
+    let hex_tx: HexEncoded<SignedTransaction> = tx.into();
+    let response = tokio::time::timeout(
+        BARRIER_TIMEOUT,
+        reqwest::Client::new()
+            .post(format!(
+                "http://{}:{}/api/v2/transaction",
+                addr.ip(),
+                addr.port()
+            ))
+            .body(hex_tx.to_string())
+            .send(),
+    )
+    .await
+    .expect("transaction submission timed out")
+    .unwrap();
+
+    let status = response.status();
+    let body = tokio::time::timeout(BARRIER_TIMEOUT, response.text())
+        .await
+        .expect("reading the submission response timed out")
+        .unwrap();
+    assert_eq!(status, 200, "transaction submission failed: {body}");
+
+    tx_id
+}
+
 /// 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: "))

api-server/stack-test-suite/tests/in_memory.rs

@@ -25,13 +25,15 @@ use common::{chain::config::create_unit_test_config, primitives::time::get_time}
 use std::sync::{Arc, RwLock};
 use tokio::net::TcpListener;
 
-pub use test_common::DummyRPC;
+pub use test_common::{
+    DummyRPC, shutdown_task, spawn_webserver_with_mempool, submit_transaction, wait_for_web_server,
+};
 
 pub async fn spawn_webserver(url: &str) -> (tokio::task::JoinHandle<()>, reqwest::Response) {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
     let addr = listener.local_addr().unwrap();
 
-    let task = tokio::spawn(async move {
+    let mut task = tokio::spawn(async move {
         let web_server_state = {
             let chain_config = Arc::new(create_unit_test_config());
             let storage = TransactionalApiServerInMemoryStorage::new(&chain_config);
@@ -51,12 +53,7 @@ pub async fn spawn_webserver(url: &str) -> (tokio::task::JoinHandle<()>, reqwest
         web_server(listener, web_server_state, true).await.unwrap();
     });
 
-    // Given that the listener port is open, this will block until a
-    // response is made (by the web server, which takes the listener
-    // over)
-    let response = reqwest::get(format!("http://{}:{}{url}", addr.ip(), addr.port()))
-        .await
-        .unwrap();
+    let response = wait_for_web_server(&mut task, addr, url).await;
 
     (task, response)
 }
@@ -68,7 +65,7 @@ async fn server_status() {
     assert_eq!(response.status(), 200);
     assert_eq!(response.text().await.unwrap(), r#"{"versions":["2.0.0"]}"#);
 
-    task.abort();
+    shutdown_task(task).await;
 }
 
 #[tokio::test]
@@ -78,5 +75,5 @@ async fn bad_request() {
     assert_eq!(response.status(), 400);
     assert_eq!(response.text().await.unwrap(), r#"{"error":"Bad request"}"#);
 
-    task.abort();
+    shutdown_task(task).await;
 }

api-server/stack-test-suite/tests/postgres_stream.rs

@@ -49,7 +49,7 @@ use common::{
     primitives::{BlockHeight, Id, Idable, time::get_time},
 };
 use hex::ToHex as _;
-use test_common::{DummyRPC, frame_data, frame_event_name};
+use test_common::{DummyRPC, frame_data, frame_event_name, shutdown_task};
 use test_utils::random::{Seed, make_seedable_rng};
 
 #[ctor::ctor]
@@ -406,23 +406,16 @@ async fn stream_events_postgres_end_to_end() {
     }
 
     // -----------------------------------------------------------------------------------------
-    // Shutdown: stop the web server and join the collector. The collector is aborted as well,
-    // since it would otherwise keep running indefinitely (the server keepalives prevent its
-    // internal chunk timeout from firing); aborting it does not swallow a panic that has
-    // already happened, which the join below propagates with the actual panic message.
+    // Shutdown: stop the web server and the collector. Both tasks are aborted before either is
+    // awaited, so that a failure in one cannot leak the other (if, say, the web server task
+    // panicked, awaiting it first would panic here before the collector is ever aborted). The
+    // collector is aborted as well, since it would otherwise keep running indefinitely (the
+    // server keepalives prevent its internal chunk timeout from firing; the terminated server
+    // connection also ends its read loop); the abort does not swallow a panic that has already
+    // happened, which the join inside `shutdown_task` propagates with the actual panic message.
     // -----------------------------------------------------------------------------------------
     web_task.abort();
     collector_task.abort();
-    if let Err(join_error) = collector_task.await {
-        assert!(
-            join_error.is_cancelled(),
-            "the SSE collector task failed: {join_error}"
-        );
-    }
-    if let Err(join_error) = web_task.await {
-        assert!(
-            join_error.is_cancelled(),
-            "the web server task failed: {join_error}"
-        );
-    }
+    shutdown_task(web_task).await;
+    shutdown_task(collector_task).await;
 }