diff --git a/Cargo.lock b/Cargo.lock index 0a1de389f..dbef7d51b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -366,6 +366,7 @@ dependencies = [ "async-trait", "axum 0.7.9", "build_utils", + "chainstate-test-framework", "clap", "common", "crypto", @@ -376,10 +377,14 @@ dependencies = [ "mempool", "node-comm", "node-lib", + "randomness", "rpc", + "rstest", "serde", "serde_json", "serialization", + "strum 0.26.3", + "test-utils", "thiserror 1.0.69", "tokio", "tower-http 0.5.2", diff --git a/api-server/CHANGELOG.md b/api-server/CHANGELOG.md index 4421a6353..7d56504b4 100644 --- a/api-server/CHANGELOG.md +++ b/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`. diff --git a/api-server/stack-test-suite/tests/common/mod.rs b/api-server/stack-test-suite/tests/common/mod.rs index 72b9df91d..49b1befc4 100644 --- a/api-server/stack-test-suite/tests/common/mod.rs +++ b/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, + ) -> Result, NodeRpcError> { + Ok(None) + } + + async fn mempool_transactions(&self) -> Result, 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>, + 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, NodeRpcError> { + Ok(vec![]) + } +} + +#[async_trait::async_trait] +impl MempoolQueryClient for MempoolRPC { + async fn mempool_transaction( + &self, + tx_id: Id, + ) -> Result, NodeRpcError> { + Ok(self + .mempool + .read() + .await + .iter() + .find(|tx| tx.transaction().get_id() == tx_id) + .cloned()) + } + + async fn mempool_transactions(&self) -> Result, 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) -> String { + payload + .downcast_ref::<&str>() + .map(|s| (*s).to_string()) + .or_else(|| payload.downcast_ref::().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, + 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(handle: tokio::task::JoinHandle) { + 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::(); + + let hex_tx: HexEncoded = 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: ")) diff --git a/api-server/stack-test-suite/tests/in_memory.rs b/api-server/stack-test-suite/tests/in_memory.rs index 8c976d0ba..a8f009d81 100644 --- a/api-server/stack-test-suite/tests/in_memory.rs +++ b/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; } diff --git a/api-server/stack-test-suite/tests/postgres_stream.rs b/api-server/stack-test-suite/tests/postgres_stream.rs index b6df644c7..b780f5164 100644 --- a/api-server/stack-test-suite/tests/postgres_stream.rs +++ b/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; } diff --git a/api-server/stack-test-suite/tests/v2/address.rs b/api-server/stack-test-suite/tests/v2/address.rs index 707f20bc3..2e7eb2260 100644 --- a/api-server/stack-test-suite/tests/v2/address.rs +++ b/api-server/stack-test-suite/tests/v2/address.rs @@ -33,7 +33,7 @@ async fn invalid_address() { assert_eq!(body["error"].as_str().unwrap(), "Invalid address"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -57,7 +57,7 @@ async fn address_not_found(#[case] seed: Seed) { assert_eq!(body["error"].as_str().unwrap(), "Address not found"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -301,7 +301,7 @@ async fn multiple_outputs_to_single_address(#[case] seed: Seed) { assert_eq!(body, expected_balance); } - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -554,7 +554,7 @@ async fn test_unlocking_for_locked_utxos(#[case] seed: Seed) { assert_eq!(body, expected_balance); } - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -783,7 +783,7 @@ async fn ok(#[case] seed: Seed) { assert_eq!(body, expected_values); } - task.abort(); + shutdown_task(task).await; } // TODO test address balances after a reorg 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 05a9c4cff..4f45ba3f6 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 @@ -33,7 +33,7 @@ async fn invalid_address() { assert_eq!(body["error"].as_str().unwrap(), "Invalid address"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -59,7 +59,7 @@ async fn address_not_found(#[case] seed: Seed) { assert!(utxos.is_empty()); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -311,7 +311,7 @@ async fn multiple_utxos_to_single_address(#[case] seed: Seed) { assert_eq!(body, expected); } - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -567,7 +567,7 @@ async fn ok(#[case] seed: Seed) { assert_eq!(body, expected_values); } - task.abort(); + shutdown_task(task).await; } // TODO test address balances after a reorg 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 e7acd6d95..569cd0318 100644 --- a/api-server/stack-test-suite/tests/v2/address_delegations.rs +++ b/api-server/stack-test-suite/tests/v2/address_delegations.rs @@ -39,7 +39,7 @@ async fn invalid_address() { assert_eq!(body["error"].as_str().unwrap(), "Invalid address"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -65,7 +65,7 @@ async fn address_not_found(#[case] seed: Seed) { assert!(utxos.is_empty()); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -241,5 +241,5 @@ async fn ok(#[case] seed: Seed) { assert_eq!(body, expected); } - task.abort(); + shutdown_task(task).await; } 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 8a79ce2c1..05cb49265 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 @@ -33,7 +33,7 @@ async fn invalid_address() { assert_eq!(body["error"].as_str().unwrap(), "Invalid address"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -62,7 +62,7 @@ async fn address_not_found(#[case] seed: Seed) { assert!(utxos.is_empty()); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -312,7 +312,7 @@ async fn multiple_utxos_to_single_address(#[case] seed: Seed) { assert_eq!(body, expected); } - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -559,7 +559,7 @@ async fn ok(#[case] seed: Seed) { assert_eq!(body, expected_values); } - task.abort(); + shutdown_task(task).await; } // TODO test address balances after a reorg 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 fc7ba5b32..a05eb85c1 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 @@ -266,5 +266,5 @@ async fn ok(#[case] seed: Seed) { } } - task.abort(); + shutdown_task(task).await; } diff --git a/api-server/stack-test-suite/tests/v2/block.rs b/api-server/stack-test-suite/tests/v2/block.rs index ec07c48e8..0e84ca884 100644 --- a/api-server/stack-test-suite/tests/v2/block.rs +++ b/api-server/stack-test-suite/tests/v2/block.rs @@ -41,7 +41,7 @@ async fn invalid_block_id() { assert_eq!(body["error"].as_str().unwrap(), "Invalid block Id"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -58,7 +58,7 @@ async fn block_not_found() { assert_eq!(body["error"].as_str().unwrap(), "Block not found"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -265,7 +265,7 @@ async fn ok(#[case] seed: Seed) { let body: serde_json::Value = serde_json::from_str(&body).unwrap(); assert_eq!(body, old_expected_block); - task.abort(); + shutdown_task(task).await; } async fn get_tx_additional_data( 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 81dc4030e..166ff1bcd 100644 --- a/api-server/stack-test-suite/tests/v2/block_header.rs +++ b/api-server/stack-test-suite/tests/v2/block_header.rs @@ -36,7 +36,7 @@ async fn invalid_block_id() { assert_eq!(body["error"].as_str().unwrap(), "Invalid block Id"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -53,7 +53,7 @@ async fn block_not_found() { assert_eq!(body["error"].as_str().unwrap(), "Block not found"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -162,5 +162,5 @@ async fn ok(#[case] seed: Seed) { assert_eq!(body, expected_header); - task.abort(); + shutdown_task(task).await; } 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 8e5520c29..af480f30c 100644 --- a/api-server/stack-test-suite/tests/v2/block_reward.rs +++ b/api-server/stack-test-suite/tests/v2/block_reward.rs @@ -33,7 +33,7 @@ async fn invalid_block_id() { assert_eq!(body["error"].as_str().unwrap(), "Invalid block Id"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -50,7 +50,7 @@ async fn block_not_found() { assert_eq!(body["error"].as_str().unwrap(), "Block not found"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -141,7 +141,7 @@ async fn no_reward(#[case] seed: Seed) { assert!(body.is_empty()); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -245,5 +245,5 @@ async fn has_reward(#[case] seed: Seed) { assert_eq!(body, expected_reward); - task.abort(); + shutdown_task(task).await; } 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 1b05d5618..1cc6de37c 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 @@ -33,7 +33,7 @@ async fn invalid_block_id() { assert_eq!(body["error"].as_str().unwrap(), "Invalid block Id"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -48,7 +48,7 @@ async fn block_not_found() { assert_eq!(body["error"].as_str().unwrap(), "Block not found"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -148,5 +148,5 @@ async fn ok(#[case] seed: Seed) { assert_eq!(body, expected_transaction_ids); - task.abort(); + shutdown_task(task).await; } 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 ca4f9e619..8b2fb4888 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 @@ -33,7 +33,7 @@ async fn invalid_height() { assert_eq!(body["error"].as_str().unwrap(), "Invalid block height"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -50,7 +50,7 @@ async fn height_zero() { "No block found at supplied height" ); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -67,7 +67,7 @@ async fn height_past_tip() { "No block found at supplied height" ); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -160,5 +160,5 @@ async fn height_n(#[case] seed: Seed) { expected_block_id.to_hash().encode_hex::() ); - task.abort(); + shutdown_task(task).await; } 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 b38319e7e..422690ba0 100644 --- a/api-server/stack-test-suite/tests/v2/chain_tip.rs +++ b/api-server/stack-test-suite/tests/v2/chain_tip.rs @@ -31,7 +31,7 @@ async fn at_genesis() { let (tx, rx) = tokio::sync::oneshot::channel(); - let task = tokio::spawn({ + let mut task = tokio::spawn({ async move { let web_server_state = { let chain_config = Arc::new(create_unit_test_config()); @@ -57,16 +57,13 @@ async fn at_genesis() { } }; - web_server(listener, web_server_state, true).await + web_server(listener, web_server_state, true) + .await + .expect("at genesis web server failed"); } }); - // 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; assert_eq!(response.status(), 200); @@ -77,7 +74,7 @@ async fn at_genesis() { assert_eq!(body, expected_tip); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -92,7 +89,7 @@ async fn height_n(#[case] seed: Seed) { let (tx, rx) = tokio::sync::oneshot::channel(); - let task = tokio::spawn({ + let mut task = tokio::spawn({ async move { let mut rng = make_seedable_rng(seed); let n_blocks = rng.random_range(1..100); @@ -150,16 +147,13 @@ async fn height_n(#[case] seed: Seed) { } }; - web_server(listener, web_server_state, true).await + web_server(listener, web_server_state, true) + .await + .expect("height n web server failed"); } }); - // 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; assert_eq!(response.status(), 200); @@ -170,5 +164,5 @@ async fn height_n(#[case] seed: Seed) { assert_eq!(body, expected_tip); - task.abort(); + shutdown_task(task).await; } diff --git a/api-server/stack-test-suite/tests/v2/feerate.rs b/api-server/stack-test-suite/tests/v2/feerate.rs index 6df8b7bc7..a33855721 100644 --- a/api-server/stack-test-suite/tests/v2/feerate.rs +++ b/api-server/stack-test-suite/tests/v2/feerate.rs @@ -40,7 +40,7 @@ async fn invalid_query_parameter() { "Invalid in top X MB query parameter" ); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -54,7 +54,7 @@ async fn ok(#[case] seed: Seed) { let listener = tokio::net::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); @@ -80,19 +80,15 @@ async fn ok(#[case] seed: Seed) { web_server(listener, web_server_state, true).await.unwrap(); }); - let response = reqwest::get(format!( - "http://{}:{}/api/v2/feerate?in_top_x_mb={in_top_x_mb}", - addr.ip(), - addr.port() - )) - .await - .unwrap(); + let url = format!("/api/v2/feerate?in_top_x_mb={in_top_x_mb}"); + + let response = wait_for_web_server(&mut task, addr, &url).await; assert_eq!(response.status(), 200); let body = response.text().await.unwrap(); assert_eq!(body, format!("\"{in_top_x_mb}\"")); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -115,6 +111,20 @@ async fn ok_reload_feerate(#[case] seed: Seed) { ]) } } + + #[async_trait::async_trait] + impl MempoolQueryClient for DummyRPC2 { + async fn mempool_transaction( + &self, + _: Id, + ) -> Result, NodeRpcError> { + Ok(None) + } + + async fn mempool_transactions(&self) -> Result, NodeRpcError> { + Ok(vec![]) + } + } let mut rng = make_seedable_rng(seed); let in_top_x_mb = rng.random_range(1..100); @@ -124,7 +134,7 @@ async fn ok_reload_feerate(#[case] seed: Seed) { let seconds = Arc::new(SeqCstAtomicU64::new(12345)); let time_getter = mocked_time_getter_seconds(Arc::clone(&seconds)); - 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); @@ -150,6 +160,14 @@ async fn ok_reload_feerate(#[case] seed: Seed) { web_server(listener, web_server_state, true).await.unwrap(); }); + let url = format!("/api/v2/feerate?in_top_x_mb={in_top_x_mb}"); + + let response = wait_for_web_server(&mut task, addr, &url).await; + assert_eq!(response.status(), 200); + + let body = response.text().await.unwrap(); + assert_eq!(body, format!("\"{in_top_x_mb}\"")); + const REFRESH_INTERVAL_SEC: u64 = 30; let mut time_passed = 0; @@ -185,5 +203,5 @@ async fn ok_reload_feerate(#[case] seed: Seed) { let new_feerate = in_top_x_mb * 2; assert_eq!(body, format!("\"{new_feerate}\"")); - task.abort(); + shutdown_task(task).await; } diff --git a/api-server/stack-test-suite/tests/v2/htlc.rs b/api-server/stack-test-suite/tests/v2/htlc.rs index 6de1513af..eb07a9e78 100644 --- a/api-server/stack-test-suite/tests/v2/htlc.rs +++ b/api-server/stack-test-suite/tests/v2/htlc.rs @@ -221,7 +221,7 @@ async fn spend(#[case] seed: Seed) { let body = response.text().await.unwrap(); assert!(body.contains(&format!("\"secret\":{}", to_json_string(secret.secret())))); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -387,5 +387,5 @@ async fn refund(#[case] seed: Seed) { let body = response.text().await.unwrap(); assert!(body.contains("\"secret\":null")); - task.abort(); + shutdown_task(task).await; } diff --git a/api-server/stack-test-suite/tests/v2/mempool_transactions.rs b/api-server/stack-test-suite/tests/v2/mempool_transactions.rs new file mode 100644 index 000000000..3872021ff --- /dev/null +++ b/api-server/stack-test-suite/tests/v2/mempool_transactions.rs @@ -0,0 +1,277 @@ +// 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. + +use chainstate_test_framework::empty_witness; +use common::{chain::UtxoOutPoint, primitives::H256}; + +use super::*; + +async fn get_mempool_transactions_response( + addr: std::net::SocketAddr, + query: &str, +) -> reqwest::Response { + let response = reqwest::get(format!( + "http://{}:{}/api/v2/mempool/transactions{query}", + addr.ip(), + addr.port() + )) + .await + .unwrap(); + + assert_eq!(response.status(), 200); + + response +} + +async fn get_mempool_transactions(addr: std::net::SocketAddr, query: &str) -> serde_json::Value { + let response = get_mempool_transactions_response(addr, query).await; + + let body = response.text().await.unwrap(); + let body: serde_json::Value = serde_json::from_str(&body).unwrap(); + + body +} + +/// The ids of the transactions of the given mempool listing, preserving the +/// order in which they are listed. +fn listed_transaction_ids_in(body: serde_json::Value) -> Vec { + let array = body + .as_array() + .unwrap_or_else(|| panic!("the mempool listing is not an array: {body}")); + array + .iter() + .map(|tx| { + tx.get("id") + .unwrap_or_else(|| panic!("a listed transaction is missing the id field: {tx}")) + .as_str() + .unwrap_or_else(|| panic!("the id field is not a string: {tx}")) + .to_owned() + }) + .collect() +} + +/// Return the ids of the transactions listed by the mempool transactions endpoint, +/// preserving the order in which they are listed. +async fn listed_transaction_ids(addr: std::net::SocketAddr, query: &str) -> Vec { + let body = get_mempool_transactions(addr, query).await; + + listed_transaction_ids_in(body) +} + +/// Return the value of the `x-mempool-ordering` response header and the ids of the +/// transactions listed by the mempool transactions endpoint, preserving the order in +/// which they are listed. +async fn listed_transaction_ids_with_ordering( + addr: std::net::SocketAddr, + query: &str, +) -> (String, Vec) { + let response = get_mempool_transactions_response(addr, query).await; + + let ordering = response + .headers() + .get("x-mempool-ordering") + .unwrap_or_else(|| panic!("the x-mempool-ordering header is missing")) + .to_str() + .unwrap() + .to_owned(); + + let body = response.text().await.unwrap(); + let body: serde_json::Value = serde_json::from_str(&body).unwrap(); + + (ordering, listed_transaction_ids_in(body)) +} + +#[rstest] +#[trace] +#[case(Seed::from_entropy())] +#[tokio::test] +async fn submitted_transaction_is_listed(#[case] seed: Seed) { + let (task, _response, _rpc, addr) = spawn_webserver_with_mempool("/").await; + let mut rng = make_seedable_rng(seed); + + let tx = TransactionBuilder::new() + .add_input( + TxInput::Utxo(UtxoOutPoint::new( + OutPointSourceId::Transaction(Id::::new(H256::random_using(&mut rng))), + 0, + )), + empty_witness(&mut rng), + ) + .build(); + + let tx_id = submit_transaction(addr, tx).await; + + let body = get_mempool_transactions(addr, "").await; + let tx_json = body + .as_array() + .unwrap_or_else(|| panic!("the mempool listing is not an array: {body}")) + .first() + .cloned() + .expect("the submitted transaction is not listed"); + + let ids = listed_transaction_ids_in(body); + + assert_eq!(ids.len(), 1); + assert_eq!(ids[0], tx_id); + + // The block-related fields of a pending transaction are null: the values + // are not applicable until the transaction is confirmed. + assert_eq!(tx_json.get("block_id"), Some(&serde_json::Value::Null)); + assert_eq!(tx_json.get("timestamp"), Some(&serde_json::Value::Null)); + assert_eq!(tx_json.get("confirmations"), Some(&serde_json::Value::Null)); + + shutdown_task(task).await; +} + +#[rstest] +#[trace] +#[case(Seed::from_entropy())] +#[tokio::test] +async fn dependency_ordering_lists_parents_before_children(#[case] seed: Seed) { + let (task, _response, _rpc, addr) = spawn_webserver_with_mempool("/").await; + let mut rng = make_seedable_rng(seed); + + // The parent spends an output unknown to this stack; the child spends the first + // output of the parent, imitating a chain of unconfirmed transactions in the + // mempool of the node. + let parent_tx = TransactionBuilder::new() + .add_input( + TxInput::Utxo(UtxoOutPoint::new( + OutPointSourceId::Transaction(Id::::new(H256::random_using(&mut rng))), + 0, + )), + empty_witness(&mut rng), + ) + .add_output(TxOutput::Transfer( + OutputValue::Coin(Amount::from_atoms(1000)), + Destination::AnyoneCanSpend, + )) + .build(); + let parent_id = parent_tx.transaction().get_id(); + + let child_tx = TransactionBuilder::new() + .add_input( + TxInput::Utxo(UtxoOutPoint::new( + OutPointSourceId::Transaction(parent_id), + 0, + )), + empty_witness(&mut rng), + ) + .add_output(TxOutput::Transfer( + OutputValue::Coin(Amount::from_atoms(500)), + Destination::AnyoneCanSpend, + )) + .build(); + + // Submit the child first: the mock mempool accepts chains of unconfirmed + // transactions without validation, so the child spending the unconfirmed + // output of the parent is accepted before the parent itself is submitted, + // imitating the out-of-order arrival of the transactions. + let child_tx_id = submit_transaction(addr, child_tx).await; + + let ids = listed_transaction_ids(addr, "").await; + assert_eq!(ids.len(), 1); + assert_eq!(ids[0], child_tx_id); + + let parent_id_hex = submit_transaction(addr, parent_tx).await; + + // With the default, insertion-based ordering, the child, which was submitted + // first, must be listed before the parent + let ids = listed_transaction_ids(addr, "").await; + + assert_eq!(ids.len(), 2); + let parent_position = ids.iter().position(|id| *id == parent_id_hex).unwrap(); + let child_position = ids.iter().position(|id| *id == child_tx_id).unwrap(); + + assert!(child_position < parent_position); + + // The dependency ordering must list the parent before the child + let ids = listed_transaction_ids(addr, "?order=dependency").await; + + assert_eq!(ids.len(), 2); + let parent_position = ids.iter().position(|id| *id == parent_id_hex).unwrap(); + let child_position = ids.iter().position(|id| *id == child_tx_id).unwrap(); + + assert!(parent_position < child_position); + + shutdown_task(task).await; +} + +#[rstest] +#[trace] +#[case(Seed::from_entropy())] +#[tokio::test] +async fn dependency_ordering_falls_back_without_refetching_the_mempool(#[case] seed: Seed) { + let (task, _response, rpc, addr) = spawn_webserver_with_mempool("/").await; + let mut rng = make_seedable_rng(seed); + + let tx = TransactionBuilder::new() + .add_input( + TxInput::Utxo(UtxoOutPoint::new( + OutPointSourceId::Transaction(Id::::new(H256::random_using(&mut rng))), + 0, + )), + empty_witness(&mut rng), + ) + .build(); + + // The same transaction is submitted twice, so the mempool listing contains + // duplicated ids, which makes the dependency ordering fail. + let tx_id = submit_transaction(addr, tx.clone()).await; + submit_transaction(addr, tx).await; + + let (ordering, ids) = listed_transaction_ids_with_ordering(addr, "?order=dependency").await; + + // The listing fell back to the insertion order of the same snapshot. + assert_eq!(ordering, "insertion"); + assert_eq!(ids, vec![tx_id.clone(), tx_id]); + // The mempool was fetched exactly once: the fallback did not refetch it. + assert_eq!(rpc.fetch_count(), 1); + + shutdown_task(task).await; +} + +#[tokio::test] +async fn invalid_ordering() { + let (task, response, _rpc, _addr) = + spawn_webserver_with_mempool("/api/v2/mempool/transactions?order=garbage").await; + + assert_eq!(response.status(), 400); + + let body = response.text().await.unwrap(); + let body: serde_json::Value = serde_json::from_str(&body).unwrap(); + + assert_eq!( + body["error"].as_str().unwrap(), + "Invalid transaction ordering" + ); + + shutdown_task(task).await; +} + +#[tokio::test] +async fn empty_mempool_returns_empty_list() { + let (task, response, _rpc, _addr) = + spawn_webserver_with_mempool("/api/v2/mempool/transactions").await; + + assert_eq!(response.status(), 200); + + let body = response.text().await.unwrap(); + let body: serde_json::Value = serde_json::from_str(&body).unwrap(); + + assert!(body.as_array().unwrap().is_empty()); + + shutdown_task(task).await; +} diff --git a/api-server/stack-test-suite/tests/v2/mod.rs b/api-server/stack-test-suite/tests/v2/mod.rs index 34552cd93..70d3fd91f 100644 --- a/api-server/stack-test-suite/tests/v2/mod.rs +++ b/api-server/stack-test-suite/tests/v2/mod.rs @@ -27,6 +27,7 @@ mod chain_tip; mod feerate; mod helpers; mod htlc; +mod mempool_transactions; mod nft; mod orders; mod pool; @@ -44,7 +45,10 @@ mod transaction_output; mod transaction_submit; mod transactions; -use crate::{DummyRPC, spawn_webserver}; +use crate::{ + DummyRPC, shutdown_task, spawn_webserver, spawn_webserver_with_mempool, submit_transaction, + wait_for_web_server, +}; use api_blockchain_scanner_lib::{ blockchain_state::BlockchainState, sync::local_state::LocalBlockchainState, }; @@ -53,7 +57,7 @@ use api_server_common::storage::{ storage_api::{ApiServerStorageWrite, ApiServerTransactionRw, Transactional}, }; use api_web_server::{ - ApiServerWebServerState, CachedValues, + ApiServerWebServerState, CachedValues, MempoolQueryClient, api::{ json_helpers::{TokenDecimals, txoutput_to_json}, web_server, @@ -99,7 +103,7 @@ async fn chain_genesis() { let (tx, rx) = tokio::sync::oneshot::channel(); - let task = tokio::spawn({ + let mut task = tokio::spawn({ async move { let web_server_state = { let chain_config = Arc::new(create_unit_test_config()); @@ -129,16 +133,13 @@ async fn chain_genesis() { } }; - web_server(listener, web_server_state, true).await + web_server(listener, web_server_state, true) + .await + .expect("chain genesis web server failed"); } }); - // 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; assert_eq!(response.status(), 200); @@ -149,5 +150,5 @@ async fn chain_genesis() { assert_eq!(body, expected_genesis); - task.abort(); + shutdown_task(task).await; } diff --git a/api-server/stack-test-suite/tests/v2/nft.rs b/api-server/stack-test-suite/tests/v2/nft.rs index 8bcb65d7a..f45566454 100644 --- a/api-server/stack-test-suite/tests/v2/nft.rs +++ b/api-server/stack-test-suite/tests/v2/nft.rs @@ -40,7 +40,7 @@ async fn invalid_nft_id() { assert_eq!(body["error"].as_str().unwrap(), "Invalid NFT Id"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -63,7 +63,7 @@ async fn nft_not_found(#[case] seed: Seed) { assert_eq!(body["error"].as_str().unwrap(), "NFT not found"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -224,5 +224,5 @@ async fn ok(#[case] seed: Seed) { assert_eq!(body, expected_values); } - task.abort(); + shutdown_task(task).await; } diff --git a/api-server/stack-test-suite/tests/v2/orders.rs b/api-server/stack-test-suite/tests/v2/orders.rs index a85feccd2..ddcaff70b 100644 --- a/api-server/stack-test-suite/tests/v2/orders.rs +++ b/api-server/stack-test-suite/tests/v2/orders.rs @@ -197,7 +197,7 @@ async fn create_fill_conclude_order(#[case] seed: Seed, #[case] version: OrdersV check_url(format!("/api/v2/transaction/{tx2_id}")).await; check_url(format!("/api/v2/transaction/{tx3_id}")).await; - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -367,5 +367,5 @@ async fn order_pairs(#[case] seed: Seed) { let arr_body = body.as_array().unwrap(); assert!(arr_body.is_empty()); - task.abort(); + shutdown_task(task).await; } diff --git a/api-server/stack-test-suite/tests/v2/pool.rs b/api-server/stack-test-suite/tests/v2/pool.rs index 6de53fa52..8719a22ac 100644 --- a/api-server/stack-test-suite/tests/v2/pool.rs +++ b/api-server/stack-test-suite/tests/v2/pool.rs @@ -35,7 +35,7 @@ async fn invalid_pool_id() { assert_eq!(body["error"].as_str().unwrap(), "Invalid pool Id"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -52,7 +52,7 @@ async fn pool_id_not_fund() { assert_eq!(body["error"].as_str().unwrap(), "Stake pool not found"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -318,5 +318,5 @@ async fn ok(#[case] seed: Seed) { } } - task.abort(); + shutdown_task(task).await; } 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 bf0044e8a..a1b51f761 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 @@ -33,7 +33,7 @@ async fn invalid_pool_id() { assert_eq!(body["error"].as_str().unwrap(), "Invalid pool Id"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -51,7 +51,7 @@ async fn from_to_not_specified() { "Failed to deserialize query string: missing field `from`" ); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -69,7 +69,7 @@ async fn pool_id_not_fund() { assert_eq!(body["error"].as_str().unwrap(), "Stake pool not found"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -197,5 +197,5 @@ async fn ok(#[case] seed: Seed) { assert_eq!(body.get("block_count").unwrap(), num_blocks); - task.abort(); + shutdown_task(task).await; } diff --git a/api-server/stack-test-suite/tests/v2/pools.rs b/api-server/stack-test-suite/tests/v2/pools.rs index e54417b7c..6fc5c37ad 100644 --- a/api-server/stack-test-suite/tests/v2/pools.rs +++ b/api-server/stack-test-suite/tests/v2/pools.rs @@ -34,7 +34,7 @@ async fn invalid_offset() { assert_eq!(body["error"].as_str().unwrap(), "Invalid offset"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -48,7 +48,7 @@ async fn invalid_num_items() { assert_eq!(body["error"].as_str().unwrap(), "Invalid number of items"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -67,7 +67,7 @@ async fn invalid_num_items_max(#[case] seed: Seed) { assert_eq!(body["error"].as_str().unwrap(), "Invalid number of items"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -81,7 +81,7 @@ async fn invalid_sort_order() { assert_eq!(body["error"].as_str().unwrap(), "Invalid pools sort order"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -334,5 +334,5 @@ async fn ok(#[case] seed: Seed) { } } - task.abort(); + shutdown_task(task).await; } diff --git a/api-server/stack-test-suite/tests/v2/statistics.rs b/api-server/stack-test-suite/tests/v2/statistics.rs index a0c9f6382..9df59e8d4 100644 --- a/api-server/stack-test-suite/tests/v2/statistics.rs +++ b/api-server/stack-test-suite/tests/v2/statistics.rs @@ -44,7 +44,7 @@ async fn invalid_token_id() { assert_eq!(body["error"].as_str().unwrap(), "Invalid token Id"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -68,7 +68,7 @@ async fn token_not_found(#[case] seed: Seed) { assert_eq!(body["error"].as_str().unwrap(), "Token not found"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -287,7 +287,7 @@ async fn ok_tokens(#[case] seed: Seed) { assert_eq!(body, expected_values); } - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -439,5 +439,5 @@ async fn ok_coins(#[case] seed: Seed) { assert_eq!(body, expected_values); } - task.abort(); + shutdown_task(task).await; } diff --git a/api-server/stack-test-suite/tests/v2/stream.rs b/api-server/stack-test-suite/tests/v2/stream.rs index 1b3c788c9..299a5402c 100644 --- a/api-server/stack-test-suite/tests/v2/stream.rs +++ b/api-server/stack-test-suite/tests/v2/stream.rs @@ -42,7 +42,7 @@ use hex::ToHex as _; use test_utils::random::{Seed, make_seedable_rng}; use crate::DummyRPC; -use crate::test_common::{frame_data, frame_event_name}; +use crate::test_common::{frame_data, frame_event_name, shutdown_task}; /// The time to wait for a single expected SSE frame. const FRAME_TIMEOUT: Duration = Duration::from_secs(5); @@ -236,7 +236,7 @@ async fn stream_endpoint_contract() { let frame = sse.next_frame(FRAME_TIMEOUT).await; assert_eq!(frame.trim(), ": keepalive", "expected a keepalive comment"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -308,7 +308,7 @@ async fn stream_types_filter() { .unwrap(); assert_eq!(response.status(), 400); - task.abort(); + shutdown_task(task).await; } /// A block event must refer to block data that is queryable through the regular REST endpoint. @@ -415,7 +415,7 @@ async fn stream_block_event_is_queryable() { tx_ids.iter().map(|tx_id| tx_id.to_hash().encode_hex::()).collect(); assert_eq!(served_tx_ids, expected_tx_ids); - task.abort(); + shutdown_task(task).await; } /// The subscriber limit must be enforced, and the slot must be released on client disconnect, so @@ -527,5 +527,5 @@ async fn stream_subscriber_limit() { assert_eq!(&parsed, expected, "event roundtrip mismatch"); } - task.abort(); + shutdown_task(task).await; } diff --git a/api-server/stack-test-suite/tests/v2/token.rs b/api-server/stack-test-suite/tests/v2/token.rs index 304d45906..e92204d8b 100644 --- a/api-server/stack-test-suite/tests/v2/token.rs +++ b/api-server/stack-test-suite/tests/v2/token.rs @@ -41,7 +41,7 @@ async fn invalid_token_id() { assert_eq!(body["error"].as_str().unwrap(), "Invalid token Id"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -64,7 +64,7 @@ async fn token_not_found(#[case] seed: Seed) { assert_eq!(body["error"].as_str().unwrap(), "Token not found"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -219,5 +219,5 @@ async fn ok(#[case] seed: Seed) { assert_eq!(body, expected_values); } - task.abort(); + shutdown_task(task).await; } 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 6bfe7d35e..6aa11a671 100644 --- a/api-server/stack-test-suite/tests/v2/token_ids.rs +++ b/api-server/stack-test-suite/tests/v2/token_ids.rs @@ -33,7 +33,7 @@ async fn invalid_offset() { assert_eq!(body["error"].as_str().unwrap(), "Invalid offset"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -47,7 +47,7 @@ async fn invalid_num_items() { assert_eq!(body["error"].as_str().unwrap(), "Invalid number of items"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -66,7 +66,7 @@ async fn invalid_num_items_max(#[case] seed: Seed) { assert_eq!(body["error"].as_str().unwrap(), "Invalid number of items"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -253,5 +253,5 @@ async fn ok(#[case] seed: Seed) { } } - task.abort(); + shutdown_task(task).await; } 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 ceda8800e..017fa5e7d 100644 --- a/api-server/stack-test-suite/tests/v2/token_ticker.rs +++ b/api-server/stack-test-suite/tests/v2/token_ticker.rs @@ -33,7 +33,7 @@ async fn invalid_offset() { assert_eq!(body["error"].as_str().unwrap(), "Invalid offset"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -47,7 +47,7 @@ async fn invalid_num_items() { assert_eq!(body["error"].as_str().unwrap(), "Invalid number of items"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -67,7 +67,7 @@ async fn invalid_num_items_max(#[case] seed: Seed) { assert_eq!(body["error"].as_str().unwrap(), "Invalid number of items"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -258,5 +258,5 @@ async fn ok(#[case] seed: Seed) { } } - task.abort(); + shutdown_task(task).await; } 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 8d818a970..9ccc0b3ff 100644 --- a/api-server/stack-test-suite/tests/v2/token_transactions.rs +++ b/api-server/stack-test-suite/tests/v2/token_transactions.rs @@ -36,7 +36,7 @@ async fn invalid_token_id() { assert_eq!(body["error"].as_str().unwrap(), "Invalid token Id"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -50,7 +50,7 @@ async fn invalid_offset() { assert_eq!(body["error"].as_str().unwrap(), "Invalid offset"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -69,7 +69,7 @@ async fn invalid_num_items() { assert_eq!(body["error"].as_str().unwrap(), "Invalid number of items"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -96,7 +96,7 @@ async fn invalid_num_items_max(#[case] seed: Seed) { assert_eq!(body["error"].as_str().unwrap(), "Invalid number of items"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -322,7 +322,7 @@ async fn ok(#[case] seed: Seed) { ); } - task.abort(); + shutdown_task(task).await; } #[track_caller] diff --git a/api-server/stack-test-suite/tests/v2/transaction.rs b/api-server/stack-test-suite/tests/v2/transaction.rs index 5c0d79fc0..e9c2edc63 100644 --- a/api-server/stack-test-suite/tests/v2/transaction.rs +++ b/api-server/stack-test-suite/tests/v2/transaction.rs @@ -30,7 +30,7 @@ async fn invalid_transaction_id() { assert_eq!(body["error"].as_str().unwrap(), "Invalid transaction Id"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -47,7 +47,58 @@ async fn transaction_not_found() { assert_eq!(body["error"].as_str().unwrap(), "Transaction not found"); - task.abort(); + shutdown_task(task).await; +} + +#[rstest] +#[trace] +#[case(Seed::from_entropy())] +#[tokio::test] +async fn pending_transaction_is_served_from_the_mempool(#[case] seed: Seed) { + use chainstate_test_framework::empty_witness; + use common::{chain::UtxoOutPoint, primitives::H256}; + + let (task, _response, _rpc, addr) = spawn_webserver_with_mempool("/").await; + let mut rng = make_seedable_rng(seed); + + let tx = TransactionBuilder::new() + .add_input( + TxInput::Utxo(UtxoOutPoint::new( + OutPointSourceId::Transaction(Id::::new(H256::random_using(&mut rng))), + 0, + )), + empty_witness(&mut rng), + ) + .build(); + + // Submit the transaction through the POST endpoint; it stays pending in the + // mempool of the node behind the web server. + let tx_id = submit_transaction(addr, tx).await; + + let response = reqwest::get(format!( + "http://{}:{}/api/v2/transaction/{tx_id}", + addr.ip(), + addr.port() + )) + .await + .unwrap(); + + assert_eq!(response.status(), 200); + + let body = response.text().await.unwrap(); + let body: serde_json::Value = serde_json::from_str(&body).unwrap(); + let body = body.as_object().unwrap(); + + assert_eq!(body.get("id").unwrap().as_str().unwrap(), tx_id); + // The block-related fields of a pending transaction are null: the values + // are not applicable until the transaction is confirmed. + assert_eq!(body.get("block_id"), Some(&serde_json::Value::Null)); + assert_eq!(body.get("timestamp"), Some(&serde_json::Value::Null)); + assert_eq!(body.get("confirmations"), Some(&serde_json::Value::Null)); + // The fee of a pending transaction is not known, so the key is omitted + assert!(body.get("fee").is_none()); + + shutdown_task(task).await; } #[rstest] @@ -287,7 +338,7 @@ async fn multiple_tx_in_same_block(#[case] seed: Seed) { &expected_transaction["confirmations"] ); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -436,7 +487,7 @@ async fn ok(#[case] seed: Seed) { &expected_transaction["confirmations"] ); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -624,5 +675,5 @@ async fn mint_tokens(#[case] seed: Seed) { let burn_out = outputs.first().unwrap().as_object().unwrap(); assert_eq!(burn_out.get("type").unwrap().as_str().unwrap(), "Burn",); - task.abort(); + shutdown_task(task).await; } 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 399223559..030a621c3 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 @@ -29,7 +29,7 @@ async fn get_transaction_failed() { assert_eq!(body["error"].as_str().unwrap(), "Invalid transaction Id"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -160,7 +160,7 @@ async fn cannot_find_transaction_in_block(#[case] seed: Seed) { "Cannot find transaction in block" ); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -271,5 +271,5 @@ async fn ok(#[case] seed: Seed) { assert_eq!(body, expected_path); - task.abort(); + shutdown_task(task).await; } 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 8717bdd24..59be7faac 100644 --- a/api-server/stack-test-suite/tests/v2/transaction_output.rs +++ b/api-server/stack-test-suite/tests/v2/transaction_output.rs @@ -30,7 +30,7 @@ async fn invalid_transaction_id() { assert_eq!(body["error"].as_str().unwrap(), "Invalid transaction Id"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -50,7 +50,7 @@ async fn transaction_not_found() { "Transaction output not found" ); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -185,5 +185,5 @@ async fn ok(#[case] seed: Seed) { ); assert!(body.get("spent_at_block_height").unwrap().is_null()); - task.abort(); + shutdown_task(task).await; } 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 bb02d27b4..8582ecd65 100644 --- a/api-server/stack-test-suite/tests/v2/transaction_submit.rs +++ b/api-server/stack-test-suite/tests/v2/transaction_submit.rs @@ -66,7 +66,7 @@ async fn dissabled_post_route() { assert_eq!(body["error"].as_str().unwrap(), "Forbidden endpoint"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -119,7 +119,7 @@ async fn invalid_transaction() { "Invalid signed transaction" ); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -186,5 +186,5 @@ async fn ok(#[case] seed: Seed) { let body = body.as_object().unwrap(); assert_eq!(body.get("tx_id").unwrap(), &tx_id); - task.abort(); + shutdown_task(task).await; } diff --git a/api-server/stack-test-suite/tests/v2/transactions.rs b/api-server/stack-test-suite/tests/v2/transactions.rs index 88b86ae05..90cabea40 100644 --- a/api-server/stack-test-suite/tests/v2/transactions.rs +++ b/api-server/stack-test-suite/tests/v2/transactions.rs @@ -33,7 +33,7 @@ async fn invalid_offset() { assert_eq!(body["error"].as_str().unwrap(), "Invalid offset"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -47,7 +47,7 @@ async fn invalid_before_tx_global_index() { assert_eq!(body["error"].as_str().unwrap(), "Invalid offset mode"); - task.abort(); + shutdown_task(task).await; } #[tokio::test] @@ -61,7 +61,7 @@ async fn invalid_num_items() { assert_eq!(body["error"].as_str().unwrap(), "Invalid number of items"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -81,7 +81,7 @@ async fn invalid_num_items_max(#[case] seed: Seed) { assert_eq!(body["error"].as_str().unwrap(), "Invalid number of items"); - task.abort(); + shutdown_task(task).await; } #[rstest] @@ -94,7 +94,7 @@ async fn ok(#[case] seed: Seed) { let (tx, rx) = tokio::sync::oneshot::channel(); - let task = tokio::spawn(async move { + let mut task = tokio::spawn(async move { let web_server_state = { let mut rng = make_seedable_rng(seed); let n_blocks = rng.random_range(3..100); @@ -207,7 +207,7 @@ async fn ok(#[case] seed: Seed) { } }; - web_server(listener, web_server_state, true).await + web_server(listener, web_server_state, true).await.expect("web server failed"); }); let expected_transactions = rx.await.unwrap(); @@ -215,9 +215,7 @@ async fn ok(#[case] seed: Seed) { let url = format!("/api/v2/transaction?offset=0&items={num_tx}"); - let response = reqwest::get(format!("http://{}:{}{url}", addr.ip(), addr.port())) - .await - .unwrap(); + let response = wait_for_web_server(&mut task, addr, &url).await; assert_eq!(response.status(), 200); @@ -274,7 +272,7 @@ async fn ok(#[case] seed: Seed) { compare_body(body, expected_transaction); } - task.abort(); + shutdown_task(task).await; } #[track_caller] diff --git a/api-server/web-server/Cargo.toml b/api-server/web-server/Cargo.toml index 6a7da8ea5..c9b028f97 100644 --- a/api-server/web-server/Cargo.toml +++ b/api-server/web-server/Cargo.toml @@ -30,5 +30,12 @@ thiserror.workspace = true tokio = { workspace = true } tower-http = { workspace = true, features = ["cors"] } +[dev-dependencies] +chainstate-test-framework = { path = "../../chainstate/test-framework" } +randomness = { path = "../../randomness" } +rstest.workspace = true +strum.workspace = true +test-utils = { path = "../../test-utils" } + [build-dependencies] build_utils = { path = "../../utils/build_utils" } diff --git a/api-server/web-server/src/api/mod.rs b/api-server/web-server/src/api/mod.rs index 6f6e332d7..ccf14cba8 100644 --- a/api-server/web-server/src/api/mod.rs +++ b/api-server/web-server/src/api/mod.rs @@ -27,7 +27,7 @@ use tower_http::cors::{AllowMethods, Any, CorsLayer}; use api_server_common::storage::storage_api::ApiServerStorage; use crate::{ - ApiServerWebServerState, TxSubmitClient, api, + ApiServerWebServerState, MempoolQueryClient, TxSubmitClient, api, error::{ApiServerWebServerClientError, ApiServerWebServerError}, }; @@ -47,7 +47,7 @@ async fn server_status() -> Result { #[allow(dead_code)] pub fn web_server< T: ApiServerStorage + Send + Sync + 'static, - R: TxSubmitClient + Send + Sync + 'static, + R: TxSubmitClient + MempoolQueryClient + Send + Sync + 'static, >( socket: TcpListener, state: ApiServerWebServerState, Arc>, diff --git a/api-server/web-server/src/api/v2.rs b/api-server/web-server/src/api/v2.rs index c8d99e791..b5d3e08d1 100644 --- a/api-server/web-server/src/api/v2.rs +++ b/api-server/web-server/src/api/v2.rs @@ -14,7 +14,7 @@ // limitations under the License. use crate::{ - TxSubmitClient, + MempoolQueryClient, TxSubmitClient, api::json_helpers::{ self, TokenDecimals, amount_to_json, block_header_to_json, pool_data_to_json, to_tx_json_with_block_info, tx_to_json, txoutput_to_json, utxo_outpoint_to_json, @@ -23,14 +23,16 @@ use crate::{ ApiServerWebServerClientError, ApiServerWebServerError, ApiServerWebServerForbiddenError, ApiServerWebServerNotFoundError, ApiServerWebServerServerError, }, + tx_dependency_ordering, }; use api_server_common::storage::storage_api::{ - AmountWithDecimals, ApiServerStorage, ApiServerStorageRead, BlockInfo, CoinOrTokenStatistic, - Order, TransactionInfo, block_aux_data::BlockAuxData, + AmountWithDecimals, ApiServerStorage, ApiServerStorageError, ApiServerStorageRead, BlockInfo, + CoinOrTokenStatistic, Order, TransactionInfo, TxAdditionalInfo, block_aux_data::BlockAuxData, }; use axum::{ Json, Router, extract::{DefaultBodyLimit, Path, Query, State}, + http::HeaderMap, response::IntoResponse, routing::{get, post}, }; @@ -38,8 +40,9 @@ use common::{ address::Address, chain::{ Block, ChainConfig, Destination, OutPointSourceId, SignedTransaction, Transaction, - UtxoOutPoint, + TxOutput, UtxoOutPoint, block::timestamp::BlockTimestamp, + make_token_id, tokens::{IsTokenFreezable, IsTokenFrozen, IsTokenUnfreezable, TokenId}, }, primitives::{Amount, BlockHeight, CoinOrTokenId, H256, Id, Idable}, @@ -55,6 +58,7 @@ use std::{ sync::Arc, time::Duration, }; +use tokio::sync::Semaphore; use utils::ensure; use crate::ApiServerWebServerState; @@ -65,9 +69,19 @@ pub const API_VERSION: &str = "2.0.0"; const TX_BODY_LIMIT: usize = 10240; +/// The maximum number of the concurrently served requests to the mempool-proxying +/// endpoints, whose cost is proportional to the size of the mempool of the node +/// instead of the size of the requested page. Additional requests wait for a free +/// permit (up to [`MEMPOOL_QUERY_WAIT_TIMEOUT`]) instead of loading the node in +/// parallel. +static MEMPOOL_QUERY_PERMITS: Semaphore = Semaphore::const_new(8); + +/// How long a request waits for a free mempool query permit before it is rejected. +const MEMPOOL_QUERY_WAIT_TIMEOUT: Duration = Duration::from_secs(30); + pub fn routes< T: ApiServerStorage + Send + Sync + 'static, - R: TxSubmitClient + Send + Sync + 'static, + R: TxSubmitClient + MempoolQueryClient + Send + Sync + 'static, >( enable_post_routes: bool, ) -> Router, Arc>> { @@ -101,6 +115,8 @@ pub fn routes< .route("/transaction/:id/merkle-path", get(transaction_merkle_path)) .route("/transaction/:id/output/:idx", get(transaction_output)); + let router = router.route("/mempool/transactions", get(mempool_transactions)); + let router = router .route("/address/:address", get(address)) .route("/address/:address/all-utxos", get(all_address_utxos)) @@ -469,6 +485,257 @@ impl FromStr for OffsetMode { } } +/// The order in which the mempool transactions are returned. +enum TxOrdering { + /// The order in which the transactions entered the mempool of the node. + Insertion, + /// Transactions that other returned transactions depend on come first. + Dependency, +} + +impl FromStr for TxOrdering { + type Err = ApiServerWebServerClientError; + + fn from_str(input: &str) -> Result { + match input { + "insertion" => Ok(Self::Insertion), + "dependency" => Ok(Self::Dependency), + _ => Err(ApiServerWebServerClientError::InvalidTransactionOrdering), + } + } +} + +/// Additional info of a pending (mempool) transaction. +/// +/// The fee and the utxos spent by the inputs are not known to the api-server without +/// indexing the mempool: the fee field is omitted from the response and the input +/// entries carry no utxo details; the number of the input entries still matches the +/// number of the transaction inputs. +/// +/// The decimals of the tokens transferred by the outputs are resolved from the +/// api-server storage or, if the issuance of the token is part of the pending listing +/// itself, from the issuing transaction; tokens whose decimals cannot be known are +/// rendered with zero decimals. +async fn pending_tx_additional_info( + db_tx: &S, + tx: &SignedTransaction, + pending_issuance_decimals: &BTreeMap, + decimals_cache: &mut BTreeMap, +) -> Result { + let internal_error = |e: ApiServerStorageError| { + logging::log::error!("internal error: {e}"); + ApiServerWebServerError::ServerError(ApiServerWebServerServerError::InternalServerError) + }; + + let mut token_decimals = BTreeMap::new(); + for token_id in tx_token_ids(tx) { + let decimals = match pending_issuance_decimals.get(&token_id) { + // The issuance of the token is pending as well: the storage has no decimals + // for it yet, but the issuing transaction carries them. + Some(decimals) => *decimals, + None => match decimals_cache.get(&token_id) { + Some(decimals) => *decimals, + None => { + let decimals = db_tx + .get_token_num_decimals(token_id) + .await + .map_err(internal_error)? + // The issuance of the token is neither pending in the listing + // nor indexed, so its decimals cannot be known, and the token + // is rendered with zero decimals. Note that the rendering is + // presentational: the atoms amounts of the response are + // authoritative regardless of the decimals. + .unwrap_or(0); + decimals_cache.insert(token_id, decimals); + decimals + } + }, + }; + token_decimals.insert(token_id, decimals); + } + + Ok(TxAdditionalInfo { + fee: Amount::ZERO, + input_utxos: vec![None; tx.transaction().inputs().len()], + token_decimals, + }) +} + +/// The decimals of the fungible token issuances carried by the given transactions. +/// +/// Returns the decimals by token id; the token ids are derived like the consensus +/// derives them for a block at the given height. Issuances whose id cannot be derived +/// are skipped. +/// +/// Note: the given height is the tip of the storage at the time of the call, while +/// a pending transaction is actually included at some later height. The derivation +/// only diverges from the consensus one if a consensus upgrade activating a new +/// token id generation version lands in between, in which case the derived ids +/// (and thus the resolved decimals and the dependency edges) are wrong until the +/// transactions are confirmed; the pending data is provisional by nature. +fn pending_issuance_decimals( + txs: &[SignedTransaction], + chain_config: &ChainConfig, + block_height: BlockHeight, +) -> BTreeMap { + // Note: walking the transactions is cheap (a match per output); the token id + // derivation only runs for the rare fungible token issuances of the mempool. + let mut decimals = BTreeMap::new(); + for tx in txs { + for out in tx.transaction().outputs() { + if let TxOutput::IssueFungibleToken(issuance) = out { + let common::chain::tokens::TokenIssuance::V1(issuance) = issuance.as_ref(); + if let Ok(token_id) = + make_token_id(chain_config, block_height, tx.transaction().inputs()) + { + decimals.insert(token_id, issuance.number_of_decimals); + } + } + } + } + decimals +} + +/// The ids of the version 1 tokens transferred by the outputs of the transaction. +fn tx_token_ids(tx: &SignedTransaction) -> BTreeSet { + common::chain::output_values_holder::collect_token_v1_ids_from_output_values_holder(tx) +} + +pub async fn mempool_transactions< + T: ApiServerStorage, + R: TxSubmitClient + MempoolQueryClient + Send + Sync + 'static, +>( + Query(params): Query>, + State(state): State, Arc>>, +) -> Result { + const ORDERING: &str = "order"; + let ordering = params + .get(ORDERING) + .map(|order| TxOrdering::from_str(order)) + .transpose()? + .unwrap_or(TxOrdering::Insertion); + + let offset_and_items = get_offset_and_items(¶ms)?; + + // Note: the cost of this endpoint is proportional to the size of the mempool of + // the node, so the number of the concurrently served requests is bounded, and a + // request that waits for a permit for too long is rejected. + let _query_permit = + tokio::time::timeout(MEMPOOL_QUERY_WAIT_TIMEOUT, MEMPOOL_QUERY_PERMITS.acquire()) + .await + .map_err(|_timed_out| ApiServerWebServerError::TooManyMempoolRequests)? + .map_err(|e| { + logging::log::error!("internal error: {e}"); + ApiServerWebServerError::ServerError( + ApiServerWebServerServerError::InternalServerError, + ) + })?; + + // Note: the tip of the storage is read once and used for both the token id + // derivation of the dependency ordering and of the pending issuances. + let inclusion_height = best_block(&state).await?.block_height().next_height(); + + let mut txs = state.rpc.mempool_transactions().await.map_err(|e| { + logging::log::error!("internal error: {e}"); + ApiServerWebServerError::ServerError(ApiServerWebServerServerError::InternalServerError) + })?; + + // Whether the listing is ordered by the dependencies between the transactions; + // a request for the dependency ordering can fall back to the insertion order. + let mut ordered_by_dependency = matches!(ordering, TxOrdering::Dependency); + + match ordering { + TxOrdering::Insertion => {} + TxOrdering::Dependency => { + let chain_config = Arc::clone(&state.chain_config); + // The sorting is CPU-bound and proportional to the mempool size; run it + // off the async runtime threads. A transaction that cannot be sorted (an + // id derivation failure of an invalid transaction) must not take down the + // whole listing: serve the insertion order instead. + txs = match tokio::task::spawn_blocking(move || { + tx_dependency_ordering::order_transactions_by_dependency( + txs, + &chain_config, + inclusion_height, + ) + }) + .await + { + Ok(Ok(sorted)) => sorted, + Ok(Err((err, fallback_txs))) => { + // The failed ordering returns the transactions unsorted in + // the original (insertion) order, so the listing falls back + // to it without refetching the mempool: a refetch could + // return a different snapshot, which would skew both the + // listing and the pending issuance decimals resolved below. + ordered_by_dependency = false; + logging::log::warn!("Falling back to the mempool insertion order: {err}"); + fallback_txs + } + Err(err) => { + logging::log::error!("internal error: {err}"); + return Err(ApiServerWebServerError::ServerError( + ApiServerWebServerServerError::InternalServerError, + )); + } + }; + } + } + + // The decimals of the issuances of the whole fetched mempool listing are resolved + // before the pagination: a transaction of the requested page may spend or transfer + // a token issued by a transaction outside of it. + let issuance_decimals = pending_issuance_decimals(&txs, &state.chain_config, inclusion_height); + + let txs = txs + .into_iter() + .skip(offset_and_items.offset as usize) + .take(offset_and_items.items as usize) + .collect::>(); + + let mut jsons = Vec::with_capacity(txs.len()); + if !txs.is_empty() { + let db_tx = state.db.transaction_ro().await.map_err(|e| { + logging::log::error!("internal error: {e}"); + ApiServerWebServerError::ServerError(ApiServerWebServerServerError::InternalServerError) + })?; + // The decimals of the same token are looked up only once per request. + let mut decimals_cache = BTreeMap::new(); + + for tx in &txs { + let additional_info = + pending_tx_additional_info(&db_tx, tx, &issuance_decimals, &mut decimals_cache) + .await?; + let mut json = tx_to_json(tx, &additional_info, &state.chain_config); + let obj = json.as_object_mut().expect("object"); + // The fee of a pending transaction is not known to the api-server. + obj.remove("fee"); + // The block-related fields of a pending transaction are null: the + // values are not applicable until the transaction is confirmed. + obj.insert("block_id".into(), serde_json::Value::Null); + obj.insert("timestamp".into(), serde_json::Value::Null); + obj.insert("confirmations".into(), serde_json::Value::Null); + jsons.push(json); + } + } + + // Tell the clients which ordering the listing ended up in: a request for the + // dependency ordering can be served in the insertion order as a fallback. + let mut headers = HeaderMap::new(); + headers.insert( + "x-mempool-ordering", + if ordered_by_dependency { + "dependency" + } else { + "insertion" + } + .parse() + .expect("valid header value"), + ); + + Ok((headers, Json(serde_json::Value::Array(jsons)))) +} + pub async fn transactions( Query(params): Query>, State(state): State, Arc>>, @@ -523,17 +790,139 @@ pub async fn transactions( Ok(Json(serde_json::Value::Array(txs))) } -pub async fn transaction( +pub async fn transaction< + T: ApiServerStorage, + R: TxSubmitClient + MempoolQueryClient + Send + Sync + 'static, +>( Path(transaction_id): Path, - State(state): State, Arc>>, + State(state): State, Arc>>, ) -> Result { - let ( - block, - TransactionInfo { - tx, - additional_info, - }, - ) = get_transaction(&transaction_id, &state).await?; + let (block, tx_info) = match get_transaction(&transaction_id, &state).await { + Ok(tx_info) => tx_info, + Err(ApiServerWebServerError::NotFound( + ApiServerWebServerNotFoundError::TransactionNotFound, + )) => { + // The transaction is not confirmed (yet); it may still be pending in the + // mempool of the connected node. + let transaction_id: Id = H256::from_str(&transaction_id) + .map_err(|_| { + ApiServerWebServerError::ClientError( + ApiServerWebServerClientError::InvalidTransactionId, + ) + })? + .into(); + + let tx = state + .rpc + .mempool_transaction(transaction_id) + .await + .map_err(|e| { + logging::log::error!("internal error: {e}"); + ApiServerWebServerError::ServerError( + ApiServerWebServerServerError::InternalServerError, + ) + })? + .ok_or(ApiServerWebServerError::NotFound( + ApiServerWebServerNotFoundError::TransactionNotFound, + ))?; + // If the transaction transfers tokens whose issuance is pending as + // well, the decimals are taken from the mempool listing, like in + // the listing endpoint; otherwise the storage is the only source. + // The listing (bounded by the query permits) is only fetched if + // some of the transferred tokens is not indexed yet, i.e. it may + // be an issuance pending in the mempool itself, so that the cost + // of the request does not scale with the size of the mempool for + // the transactions transferring the already known tokens. + // + // Note that a transaction referencing a token that does not exist + // at all keeps taking this path: the cost of a listing fetch per + // such request is accepted, since it is the same cost class as the + // listing endpoint itself, and the concurrency (and the wait for + // it) is bounded by the query permits. + let token_ids = tx_token_ids(&tx); + let db_tx = state.db.transaction_ro().await.map_err(|e| { + logging::log::error!("internal error: {e}"); + ApiServerWebServerError::ServerError( + ApiServerWebServerServerError::InternalServerError, + ) + })?; + // The decimals of the same token are looked up only once. + let mut decimals_cache = BTreeMap::new(); + let pending_issuance_decimals = if token_ids.is_empty() { + BTreeMap::new() + } else { + // The tokens with the decimals already indexed are resolved + // from the storage right away; only the missing ones can be + // the issuances pending in the mempool listing. + let mut pending_token_ids = BTreeSet::new(); + for token_id in token_ids { + match db_tx.get_token_num_decimals(token_id).await.map_err(|e| { + logging::log::error!("internal error: {e}"); + ApiServerWebServerError::ServerError( + ApiServerWebServerServerError::InternalServerError, + ) + })? { + Some(decimals) => { + decimals_cache.insert(token_id, decimals); + } + None => { + pending_token_ids.insert(token_id); + } + } + } + + if pending_token_ids.is_empty() { + BTreeMap::new() + } else { + // Note: like in the listing endpoint, the number of the + // concurrently served mempool queries is bounded, and a + // request that waits for a permit for too long is rejected. + let _query_permit = tokio::time::timeout( + MEMPOOL_QUERY_WAIT_TIMEOUT, + MEMPOOL_QUERY_PERMITS.acquire(), + ) + .await + .map_err(|_timed_out| ApiServerWebServerError::TooManyMempoolRequests)? + .map_err(|e| { + logging::log::error!("internal error: {e}"); + ApiServerWebServerError::ServerError( + ApiServerWebServerServerError::InternalServerError, + ) + })?; + let mempool_txs = state.rpc.mempool_transactions().await.map_err(|e| { + logging::log::error!("internal error: {e}"); + ApiServerWebServerError::ServerError( + ApiServerWebServerServerError::InternalServerError, + ) + })?; + let inclusion_height = best_block(&state).await?.block_height().next_height(); + pending_issuance_decimals(&mempool_txs, &state.chain_config, inclusion_height) + } + }; + + let additional_info = pending_tx_additional_info( + &db_tx, + &tx, + &pending_issuance_decimals, + &mut decimals_cache, + ) + .await?; + + ( + None, + TransactionInfo { + tx, + additional_info, + }, + ) + } + Err(err) => return Err(err), + }; + + let TransactionInfo { + tx, + additional_info, + } = tx_info; let confirmations = if let Some(block) = &block { let tip_height = best_block(&state).await?.block_height(); @@ -544,25 +933,27 @@ pub async fn transaction( let mut json = tx_to_json(&tx, &additional_info, &state.chain_config); let obj = json.as_object_mut().expect("object"); + if block.is_none() { + // The transaction is pending in the mempool: the fee of a pending + // transaction is not known to the api-server. + obj.remove("fee"); + } + obj.insert( "block_id".into(), - block - .as_ref() - .map_or("".to_string(), |b| { - b.block_id().to_hash().encode_hex::() - }) - .into(), + block.as_ref().map_or(serde_json::Value::Null, |b| { + b.block_id().to_hash().encode_hex::().into() + }), ); obj.insert( "timestamp".into(), - block - .as_ref() - .map_or("".to_string(), |b| b.block_timestamp().to_string()) - .into(), + block.as_ref().map_or(serde_json::Value::Null, |b| { + b.block_timestamp().to_string().into() + }), ); obj.insert( "confirmations".into(), - confirmations.map_or("".to_string(), |c| c.to_string()).into(), + confirmations.map_or(serde_json::Value::Null, |c| c.to_string().into()), ); Ok(Json(json)) @@ -1670,3 +2061,45 @@ fn get_offset_and_items( Ok(OffsetAndItems { offset, items }) } + +#[cfg(test)] +mod tests { + use super::*; + use chainstate_test_framework::TransactionBuilder; + use common::{ + chain::{TxInput, config::create_regtest, signature::inputsig::InputWitness}, + primitives::Id, + }; + + /// The pending-transaction responses are derived from the output of + /// `tx_to_json` by removing the `fee` key (the fee of a pending transaction + /// is not known to the api-server) and overwriting the block-related keys. + /// This pins the contract: `tx_to_json` must always emit the `fee` key, so + /// that its removal in the pending responses cannot silently stop working. + #[test] + fn tx_to_json_always_emits_the_fee_key_removed_by_the_pending_responses() { + let chain_config = create_regtest(); + let tx = TransactionBuilder::new() + .add_input( + TxInput::Utxo(UtxoOutPoint::new( + OutPointSourceId::Transaction(Id::::new(H256::zero())), + 0, + )), + InputWitness::NoSignature(None), + ) + .build(); + let additional_info = TxAdditionalInfo { + fee: Amount::ZERO, + input_utxos: vec![], + token_decimals: BTreeMap::new(), + }; + + let json = tx_to_json(&tx, &additional_info, &chain_config); + + let obj = json.as_object().expect("tx_to_json must produce an object"); + assert!( + obj.contains_key("fee"), + "tx_to_json must emit the `fee` key: {obj:?}" + ); + } +} diff --git a/api-server/web-server/src/error.rs b/api-server/web-server/src/error.rs index b29164718..edf807cda 100644 --- a/api-server/web-server/src/error.rs +++ b/api-server/web-server/src/error.rs @@ -45,6 +45,8 @@ pub enum ApiServerWebServerError { ServerError(#[from] ApiServerWebServerServerError), #[error("Too many concurrent stream connections")] TooManyStreamConnections, + #[error("Too many concurrent mempool requests")] + TooManyMempoolRequests, } #[derive(Debug, Error, Serialize)] @@ -97,6 +99,8 @@ pub enum ApiServerWebServerClientError { InvalidOffset, #[error("Invalid offset mode")] InvalidOffsetMode, + #[error("Invalid transaction ordering")] + InvalidTransactionOrdering, #[error("Invalid number of items")] InvalidNumItems, #[error("Invalid pools sort order")] @@ -146,6 +150,9 @@ impl IntoResponse for ApiServerWebServerError { ApiServerWebServerError::TooManyStreamConnections => { (StatusCode::TOO_MANY_REQUESTS, self.to_string()) } + ApiServerWebServerError::TooManyMempoolRequests => { + (StatusCode::TOO_MANY_REQUESTS, self.to_string()) + } }; (status, Json(json!({ "error": message }))).into_response() diff --git a/api-server/web-server/src/lib.rs b/api-server/web-server/src/lib.rs index cb7ba5851..ab547bb03 100644 --- a/api-server/web-server/src/lib.rs +++ b/api-server/web-server/src/lib.rs @@ -17,13 +17,14 @@ pub mod api; pub mod config; pub mod error; pub mod streaming; +pub mod tx_dependency_ordering; pub use error::ApiServerWebServerError; pub use streaming::{StreamEventsHandle, StreamingConfig}; use common::{ - chain::{ChainConfig, SignedTransaction}, - primitives::time::Time, + chain::{ChainConfig, SignedTransaction, Transaction}, + primitives::{Id, time::Time}, time_getter::TimeGetter, }; use mempool::FeeRate; @@ -40,6 +41,21 @@ pub trait TxSubmitClient { async fn get_feerate_points(&self) -> Result, NodeRpcError>; } +/// Queries into the mempool of the connected node. +/// +/// The returned transactions are pending: they may be included into a block later, +/// or disappear (e.g. by being evicted or by being included into a block that is +/// later reorganized away). +#[async_trait::async_trait] +pub trait MempoolQueryClient { + async fn mempool_transaction( + &self, + tx_id: Id, + ) -> Result, NodeRpcError>; + + async fn mempool_transactions(&self) -> Result, NodeRpcError>; +} + #[async_trait::async_trait] impl TxSubmitClient for NodeRpcClient { async fn submit_tx(&self, tx: SignedTransaction) -> Result<(), NodeRpcError> { @@ -51,6 +67,20 @@ impl TxSubmitClient for NodeRpcClient { } } +#[async_trait::async_trait] +impl MempoolQueryClient for NodeRpcClient { + async fn mempool_transaction( + &self, + tx_id: Id, + ) -> Result, NodeRpcError> { + NodeInterface::mempool_get_transaction(self, tx_id).await + } + + async fn mempool_transactions(&self) -> Result, NodeRpcError> { + NodeInterface::mempool_get_transactions(self).await + } +} + pub struct CachedValues { pub feerate_points: RwLock<(Time, Vec<(usize, FeeRate)>)>, } diff --git a/api-server/web-server/src/main.rs b/api-server/web-server/src/main.rs index d2e094166..a81ecb601 100644 --- a/api-server/web-server/src/main.rs +++ b/api-server/web-server/src/main.rs @@ -22,8 +22,8 @@ use api_server_common::storage::impls::postgres::{ }; use api_server_common::streaming::StreamEventsChannel; use api_web_server::{ - ApiServerWebServerState, CachedValues, StreamEventsHandle, TxSubmitClient, api::web_server, - config::ApiServerWebServerConfig, streaming, + ApiServerWebServerState, CachedValues, MempoolQueryClient, StreamEventsHandle, TxSubmitClient, + api::web_server, config::ApiServerWebServerConfig, streaming, tx_dependency_ordering, }; use clap::Parser; use common::{ diff --git a/api-server/web-server/src/tx_dependency_ordering/dependency_graph.rs b/api-server/web-server/src/tx_dependency_ordering/dependency_graph.rs new file mode 100644 index 000000000..6cd434989 --- /dev/null +++ b/api-server/web-server/src/tx_dependency_ordering/dependency_graph.rs @@ -0,0 +1,826 @@ +// 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. + +use std::collections::BTreeMap; + +use common::{ + chain::{ + AccountCommand, AccountNonce, AccountSpending, ChainConfig, DelegationId, + OrderAccountCommand, OrderId, OutPointSourceId, PoolId, SignedTransaction, Transaction, + TxInput, TxOutput, UtxoOutPoint, make_order_id, make_token_id, output_value::OutputValue, + tokens::TokenId, + }, + primitives::{BlockHeight, Id, Idable}, +}; + +/// A type of dependency that a transaction can depend on +#[derive(Eq, PartialEq, Ord, PartialOrd, Clone)] +enum Dependency { + Utxo(UtxoOutPoint), + TokenCreation(TokenId), + TokenCommand(TokenId, AccountNonce), + OrderCreation(OrderId), + OrderFill(OrderId), + OrderFreeze(OrderId), + DelegationCreation(DelegationId), + DelegationSpending(DelegationId, AccountNonce), + PoolCreation(PoolId), +} + +type TxIndex = usize; + +struct DependenciesMap { + providers: BTreeMap>, + dependents: BTreeMap>, +} + +impl DependenciesMap { + fn new() -> Self { + Self { + providers: BTreeMap::new(), + dependents: BTreeMap::new(), + } + } +} + +pub trait DependencyNode { + type Id: Eq + PartialOrd + Ord + std::fmt::Debug; + type Priority: Ord + Copy; + + fn id(&self) -> Self::Id; + fn dependencies(&self) -> &[Self::Id]; + fn priority(&self) -> Self::Priority; +} + +// Highest priority txs are first, then delegation stake, delegation withdrawal and last token freeze +#[derive(Debug, Eq, PartialEq, Ord, PartialOrd, Clone, Copy)] +pub enum TxPriorityOrder { + Highest = 3, + DelegationStake = 2, + DelegationWithdrawal = 1, + TokenFreeze = 0, +} + +/// Represents an item that can be topologically sorted. +#[derive(Eq, PartialEq)] +pub struct TxDependencyNode { + id: Id, + dependencies: Vec>, + tx: SignedTransaction, + priority: TxPriorityOrder, +} + +impl TxDependencyNode { + fn new(tx: SignedTransaction) -> Self { + let id = tx.transaction().get_id(); + let priority = tx_priority_order(&tx); + + Self { + id, + dependencies: Vec::new(), + tx, + priority, + } + } + + pub fn into_signed_transaction(self) -> SignedTransaction { + self.tx + } +} + +impl DependencyNode for TxDependencyNode { + type Id = Id; + type Priority = TxPriorityOrder; + + fn id(&self) -> Self::Id { + self.id + } + + fn priority(&self) -> Self::Priority { + self.priority + } + + fn dependencies(&self) -> &[Self::Id] { + &self.dependencies + } +} + +// Build a dependency graph from the provided transactions +pub fn build_dependency_graph( + transactions: Vec, + chain_config: &ChainConfig, + block_height: BlockHeight, +) -> Vec { + let mut dependencies = DependenciesMap::new(); + + for (tx_index, tx) in transactions.iter().enumerate() { + process_input_dependencies(tx_index, &mut dependencies, tx); + process_output_dependencies(tx_index, chain_config, block_height, &mut dependencies, tx); + } + + let mut dependency_nodes = + transactions.into_iter().map(TxDependencyNode::new).collect::>(); + + for (dep, tx_indices) in dependencies.dependents.iter() { + if let Some(providers) = dependencies.providers.get(dep) { + let providers = providers + .iter() + .map(|provider_tx_index| dependency_nodes[*provider_tx_index].id) + .collect::>(); + + for tx_index in tx_indices { + // A transaction can be both a provider and a dependent of the same + // dependency (e.g. a transaction carrying two account commands of the + // same token at consecutive nonces): such a self-dependency carries no + // ordering information and would be reported as a cycle. + let self_id = dependency_nodes[*tx_index].id; + dependency_nodes[*tx_index] + .dependencies + .extend(providers.iter().filter(|id| **id != self_id).copied()); + } + } + } + + dependency_nodes +} + +fn process_output_dependencies( + tx_index: usize, + chain_config: &ChainConfig, + block_height: BlockHeight, + dependencies: &mut DependenciesMap, + tx: &SignedTransaction, +) { + let inputs = tx.transaction().inputs(); + for (out_index, out) in tx.transaction().outputs().iter().enumerate() { + // Every output is a spendable utxo regardless of its type, so the + // transactions spending it depend on this one. + let outpoint = UtxoOutPoint::new( + OutPointSourceId::Transaction(tx.transaction().get_id()), + out_index as u32, + ); + dependencies + .providers + .entry(Dependency::Utxo(outpoint)) + .or_default() + .push(tx_index); + + match out { + TxOutput::CreateOrder(order_data) => { + let order_id = match make_order_id(inputs) { + Ok(order_id) => order_id, + Err(err) => { + // A transaction whose order id cannot be derived is + // invalid, but the node may still track it in the + // mempool: such a transaction must not take down the + // ordering of the whole listing, so only its + // dependency edges are skipped. + logging::log::warn!( + "The order id of the output {out_index} of the transaction {} cannot be derived; skipping its dependency edges: {err}", + tx.transaction().get_id(), + ); + continue; + } + }; + dependencies + .providers + .entry(Dependency::OrderCreation(order_id)) + .or_default() + .push(tx_index); + + match order_data.ask() { + OutputValue::TokenV1(token_id, _) => { + dependencies + .dependents + .entry(Dependency::TokenCreation(*token_id)) + .or_default() + .push(tx_index); + } + OutputValue::Coin(_) | OutputValue::TokenV0(_) => {} + } + match order_data.give() { + OutputValue::TokenV1(token_id, _) => { + dependencies + .dependents + .entry(Dependency::TokenCreation(*token_id)) + .or_default() + .push(tx_index); + } + OutputValue::Coin(_) | OutputValue::TokenV0(_) => {} + } + } + TxOutput::IssueFungibleToken(_) => { + let token_id = match make_token_id(chain_config, block_height, inputs) { + Ok(token_id) => token_id, + Err(err) => { + // See the order id derivation above: the dependency + // edges of an un-derivable transaction are skipped. + logging::log::warn!( + "The token id of the output {out_index} of the transaction {} cannot be derived; skipping its dependency edges: {err}", + tx.transaction().get_id(), + ); + continue; + } + }; + dependencies + .providers + .entry(Dependency::TokenCreation(token_id)) + .or_default() + .push(tx_index); + } + TxOutput::IssueNft(token_id, _, _) => { + // Note: an nft issuance mints an nft of an existing token; it does not + // create the token itself, so it provides no token creation dependency. + let _ = token_id; + } + TxOutput::DelegateStaking(_amount, _delegation_id) => { + // Note: in the mempool of the node, a delegation stake provides no + // mempool-side dependency: staking requires the delegation to be + // already known to the chain, like the first spend of it does. + } + TxOutput::CreateStakePool(pool_id, _) => { + dependencies + .providers + .entry(Dependency::PoolCreation(*pool_id)) + .or_default() + .push(tx_index); + } + TxOutput::CreateDelegationId(_, pool_id) => { + // Creating a delegation requires the stake pool to be known already. + dependencies + .dependents + .entry(Dependency::PoolCreation(*pool_id)) + .or_default() + .push(tx_index); + } + // The remaining outputs carry no dependencies beyond the utxo one. + _ => {} + } + } +} + +fn tx_priority_order(tx: &SignedTransaction) -> TxPriorityOrder { + let mut priority = TxPriorityOrder::Highest; + for inp in tx.transaction().inputs().iter() { + match inp { + TxInput::Utxo(_) => {} + TxInput::Account(acc) => match acc.account() { + AccountSpending::DelegationBalance(_, _) => { + priority = std::cmp::min(priority, TxPriorityOrder::DelegationWithdrawal); + } + }, + TxInput::AccountCommand(_, cmd) => match cmd { + AccountCommand::FreezeToken(_, _) => { + priority = std::cmp::min(priority, TxPriorityOrder::TokenFreeze); + } + AccountCommand::MintTokens(_, _) + | AccountCommand::UnmintTokens(_) + | AccountCommand::UnfreezeToken(_) + | AccountCommand::LockTokenSupply(_) + | AccountCommand::ChangeTokenMetadataUri(_, _) + | AccountCommand::ChangeTokenAuthority(_, _) + | AccountCommand::ConcludeOrder(_) + | AccountCommand::FillOrder(_, _, _) => {} + }, + TxInput::OrderAccountCommand(_) => {} + } + } + + for out in tx.transaction().outputs() { + if let TxOutput::DelegateStaking(_, _) = out { + priority = std::cmp::min(priority, TxPriorityOrder::DelegationStake); + } + } + + priority +} + +fn process_input_dependencies( + tx_index: usize, + dependencies: &mut DependenciesMap, + tx: &SignedTransaction, +) { + for inp in tx.transaction().inputs().iter() { + match inp { + TxInput::Utxo(utxo_outpoint) => { + dependencies + .dependents + .entry(Dependency::Utxo(utxo_outpoint.clone())) + .or_default() + .push(tx_index); + } + TxInput::Account(acct) => { + // The delegation spends of an account are nonce-sequenced: a spend of + // the nonce `n` has to come after the spend of the nonce `n - 1`, and + // the first spend (nonce 0) has to come after the delegation creation. + match acct.account() { + AccountSpending::DelegationBalance(delegation_id, _) => { + dependencies + .dependents + .entry(Dependency::DelegationSpending(*delegation_id, acct.nonce())) + .or_default() + .push(tx_index); + + if acct.nonce().value() == 0 { + dependencies + .dependents + .entry(Dependency::DelegationCreation(*delegation_id)) + .or_default() + .push(tx_index); + } + + // The next spend of the delegation has to come after this one. + if let Some(next_nonce) = acct.nonce().increment() { + dependencies + .providers + .entry(Dependency::DelegationSpending(*delegation_id, next_nonce)) + .or_default() + .push(tx_index); + } + } + } + } + TxInput::AccountCommand(nonce, cmd) => match cmd { + AccountCommand::MintTokens(token_id, _) + | AccountCommand::FreezeToken(token_id, _) + | AccountCommand::UnmintTokens(token_id) + | AccountCommand::UnfreezeToken(token_id) + | AccountCommand::LockTokenSupply(token_id) + | AccountCommand::ChangeTokenMetadataUri(token_id, _) + | AccountCommand::ChangeTokenAuthority(token_id, _) => { + dependencies + .providers + .entry(Dependency::TokenCommand(*token_id, *nonce)) + .or_default() + .push(tx_index); + + dependencies + .dependents + .entry(Dependency::TokenCreation(*token_id)) + .or_default() + .push(tx_index); + if let Some(previous_nonce) = nonce.decrement() { + dependencies + .dependents + .entry(Dependency::TokenCommand(*token_id, previous_nonce)) + .or_default() + .push(tx_index); + } + } + // The deprecated order commands (before the orders v1 upgrade) operate + // on the same orders as their `OrderAccountCommand` counterparts, so + // they carry the same dependencies. + AccountCommand::FillOrder(order_id, _, _) => { + dependencies + .providers + .entry(Dependency::OrderFill(*order_id)) + .or_default() + .push(tx_index); + + dependencies + .dependents + .entry(Dependency::OrderCreation(*order_id)) + .or_default() + .push(tx_index); + } + AccountCommand::ConcludeOrder(order_id) => { + dependencies + .dependents + .entry(Dependency::OrderCreation(*order_id)) + .or_default() + .push(tx_index); + dependencies + .dependents + .entry(Dependency::OrderFill(*order_id)) + .or_default() + .push(tx_index); + dependencies + .dependents + .entry(Dependency::OrderFreeze(*order_id)) + .or_default() + .push(tx_index); + } + }, + TxInput::OrderAccountCommand(cmd) => match cmd { + OrderAccountCommand::FillOrder(order_id, _) => { + dependencies + .providers + .entry(Dependency::OrderFill(*order_id)) + .or_default() + .push(tx_index); + + dependencies + .dependents + .entry(Dependency::OrderCreation(*order_id)) + .or_default() + .push(tx_index); + } + OrderAccountCommand::FreezeOrder(order_id) => { + dependencies + .providers + .entry(Dependency::OrderFreeze(*order_id)) + .or_default() + .push(tx_index); + + dependencies + .dependents + .entry(Dependency::OrderCreation(*order_id)) + .or_default() + .push(tx_index); + dependencies + .dependents + .entry(Dependency::OrderFill(*order_id)) + .or_default() + .push(tx_index); + } + OrderAccountCommand::ConcludeOrder(order_id) => { + dependencies + .dependents + .entry(Dependency::OrderCreation(*order_id)) + .or_default() + .push(tx_index); + dependencies + .dependents + .entry(Dependency::OrderFill(*order_id)) + .or_default() + .push(tx_index); + dependencies + .dependents + .entry(Dependency::OrderFreeze(*order_id)) + .or_default() + .push(tx_index); + } + }, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + use chainstate_test_framework::TransactionBuilder; + use common::{ + chain::{ + AccountCommandTag, Destination, OrderData, OutPointSourceId, TxInput, TxOutput, + UtxoOutPoint, + config::create_regtest, + output_value::OutputValue, + signature::inputsig::InputWitness, + tokens::{IsTokenUnfreezable, TokenIssuance}, + }, + primitives::{Amount, BlockHeight, H256, Id}, + }; + use randomness::RngExt as _; + use randomness::seq::IteratorRandom; + use test_utils::{ + random::{Rng, Seed, make_seedable_rng}, + token_utils::random_token_issuance_v1, + }; + + use rstest::rstest; + use strum::IntoEnumIterator; + + #[rstest] + #[trace] + #[case(Seed::from_entropy())] + fn test_utxo_dependency_chain(#[case] seed: Seed) { + let mut rng = make_seedable_rng(seed); + let chain_config = create_regtest(); + let block_height = BlockHeight::new(0); + + // simple A -> B UTXO chain + let random_tx_id = Id::new(H256::random_using(&mut rng)); + let random_utxo_outpoint = + UtxoOutPoint::new(OutPointSourceId::Transaction(random_tx_id), 0); + let txa = TransactionBuilder::new() + .add_input( + TxInput::Utxo(random_utxo_outpoint), + InputWitness::NoSignature(None), + ) + .add_anyone_can_spend_output(100) + .build(); + let txa_id = txa.transaction().get_id(); + + let output_from_txa = UtxoOutPoint::new(OutPointSourceId::Transaction(txa_id), 0); + let txb = TransactionBuilder::new() + .add_input( + TxInput::Utxo(output_from_txa), + InputWitness::NoSignature(None), + ) + .add_anyone_can_spend_output(100) + .build(); + let txb_id = txb.transaction().get_id(); + + let transactions = vec![txa, txb]; + let dependency_graph = build_dependency_graph(transactions, &chain_config, block_height); + assert_eq!(dependency_graph.len(), 2); + assert_eq!(dependency_graph[0].id, txa_id); + assert_eq!(dependency_graph[1].id, txb_id); + assert!(dependency_graph[0].dependencies.is_empty()); + assert_eq!(dependency_graph[1].dependencies, vec![txa_id]); + } + + // test txs not dependent on UTXO input/outputs but on token creation/command + #[rstest] + #[trace] + #[case(Seed::from_entropy())] + fn test_token_creation_dependency_chain(#[case] seed: Seed) { + let mut rng = make_seedable_rng(seed); + let chain_config = create_regtest(); + let block_height = BlockHeight::new(0); + + // A creates a new token, B uses a command on it + let random_tx_id = Id::new(H256::random_using(&mut rng)); + let random_utxo_outpoint = + UtxoOutPoint::new(OutPointSourceId::Transaction(random_tx_id), 0); + let token = TxOutput::IssueFungibleToken(Box::new(TokenIssuance::V1( + random_token_issuance_v1(&chain_config, Destination::AnyoneCanSpend, &mut rng), + ))); + let txa = TransactionBuilder::new() + .add_input( + TxInput::Utxo(random_utxo_outpoint), + InputWitness::NoSignature(None), + ) + .add_output(token) + .build(); + let txa_id = txa.transaction().get_id(); + let token_id = make_token_id( + &chain_config, + BlockHeight::new(0), + txa.transaction().inputs(), + ) + .unwrap(); + + let random_utxo_outpoint2 = + UtxoOutPoint::new(OutPointSourceId::Transaction(random_tx_id), 1); + let random_command1 = make_random_token_command(token_id, &mut rng); + let txb = TransactionBuilder::new() + .add_input( + TxInput::Utxo(random_utxo_outpoint2), + InputWitness::NoSignature(None), + ) + .add_input( + TxInput::AccountCommand(AccountNonce::new(0), random_command1), + InputWitness::NoSignature(None), + ) + .add_anyone_can_spend_output(100) + .build(); + let txb_id = txb.transaction().get_id(); + + let random_utxo_outpoint3 = + UtxoOutPoint::new(OutPointSourceId::Transaction(random_tx_id), 2); + let random_command2 = make_random_token_command(token_id, &mut rng); + let txc = TransactionBuilder::new() + .add_input( + TxInput::Utxo(random_utxo_outpoint3), + InputWitness::NoSignature(None), + ) + .add_input( + TxInput::AccountCommand(AccountNonce::new(1), random_command2), + InputWitness::NoSignature(None), + ) + .add_anyone_can_spend_output(100) + .build(); + let txc_id = txc.transaction().get_id(); + + let transactions = vec![txa, txb, txc]; + let dependency_graph = build_dependency_graph(transactions, &chain_config, block_height); + assert_eq!(dependency_graph.len(), 3); + assert_eq!(dependency_graph[0].id, txa_id); + assert_eq!(dependency_graph[1].id, txb_id); + assert_eq!(dependency_graph[2].id, txc_id); + assert!(dependency_graph[0].dependencies.is_empty()); + assert_eq!(dependency_graph[1].dependencies, vec![txa_id]); + assert_eq!(dependency_graph[2].dependencies, vec![txa_id, txb_id]); + } + + // The outputs of all the types are spendable utxos: a mempool transaction + // spending the output of a typed-output transaction (e.g. a token issuance) + // must depend on it, like the spends of the plain transfer outputs do; + // otherwise the ordering fails with a missing dependency. + #[rstest] + #[trace] + #[case(Seed::from_entropy())] + fn test_typed_output_utxo_dependency_chain(#[case] seed: Seed) { + let mut rng = make_seedable_rng(seed); + let chain_config = create_regtest(); + let block_height = BlockHeight::new(0); + + let random_tx_id = Id::new(H256::random_using(&mut rng)); + let random_utxo_outpoint = + UtxoOutPoint::new(OutPointSourceId::Transaction(random_tx_id), 0); + let txa = TransactionBuilder::new() + .add_input( + TxInput::Utxo(random_utxo_outpoint), + InputWitness::NoSignature(None), + ) + .add_output(TxOutput::IssueFungibleToken(Box::new(TokenIssuance::V1( + random_token_issuance_v1(&chain_config, Destination::AnyoneCanSpend, &mut rng), + )))) + .add_anyone_can_spend_output(100) + .build(); + let txa_id = txa.transaction().get_id(); + + let output_from_txa = UtxoOutPoint::new(OutPointSourceId::Transaction(txa_id), 0); + let txb = TransactionBuilder::new() + .add_input( + TxInput::Utxo(output_from_txa), + InputWitness::NoSignature(None), + ) + .add_anyone_can_spend_output(100) + .build(); + let txb_id = txb.transaction().get_id(); + + let transactions = vec![txa, txb]; + let dependency_graph = build_dependency_graph(transactions, &chain_config, block_height); + assert_eq!(dependency_graph.len(), 2); + assert_eq!(dependency_graph[0].id, txa_id); + assert_eq!(dependency_graph[1].id, txb_id); + assert!(dependency_graph[0].dependencies.is_empty()); + assert_eq!(dependency_graph[1].dependencies, vec![txa_id]); + } + + // test new order depending on new token creation + #[rstest] + #[trace] + #[case(Seed::from_entropy())] + fn test_order_depending_on_token_creation(#[case] seed: Seed) { + let mut rng = make_seedable_rng(seed); + let chain_config = create_regtest(); + let block_height = BlockHeight::new(0); + + // A creates a new token, B creates an order using it + let random_tx_id = Id::new(H256::random_using(&mut rng)); + let random_utxo_outpoint = + UtxoOutPoint::new(OutPointSourceId::Transaction(random_tx_id), 0); + let token = TxOutput::IssueFungibleToken(Box::new(TokenIssuance::V1( + random_token_issuance_v1(&chain_config, Destination::AnyoneCanSpend, &mut rng), + ))); + let txa = TransactionBuilder::new() + .add_input( + TxInput::Utxo(random_utxo_outpoint), + InputWitness::NoSignature(None), + ) + .add_output(token) + .build(); + let txa_id = txa.transaction().get_id(); + let token_id = make_token_id( + &chain_config, + BlockHeight::new(0), + txa.transaction().inputs(), + ) + .unwrap(); + + let random_utxo_outpoint2 = + UtxoOutPoint::new(OutPointSourceId::Transaction(random_tx_id), 1); + + let order_data = OrderData::new( + Destination::AnyoneCanSpend, + OutputValue::Coin(Amount::from_atoms(10)), + OutputValue::TokenV1(token_id, Amount::from_atoms(10)), + ); + let order = TxOutput::CreateOrder(Box::new(order_data)); + let txb = TransactionBuilder::new() + .add_input( + TxInput::Utxo(random_utxo_outpoint2), + InputWitness::NoSignature(None), + ) + .add_output(order) + .build(); + let txb_id = txb.transaction().get_id(); + + let transactions = vec![txa, txb]; + let dependency_graph = build_dependency_graph(transactions, &chain_config, block_height); + assert_eq!(dependency_graph.len(), 2); + assert_eq!(dependency_graph[0].id, txa_id); + assert_eq!(dependency_graph[1].id, txb_id); + assert!(dependency_graph[0].dependencies.is_empty()); + assert_eq!(dependency_graph[1].dependencies, vec![txa_id]); + } + + // test transactions dependent on orders creation fill freeze and conclude + #[rstest] + #[trace] + #[case(Seed::from_entropy())] + fn test_order_creation_fill_freeze_conclude_dependency_chain(#[case] seed: Seed) { + let mut rng = make_seedable_rng(seed); + let chain_config = create_regtest(); + let block_height = BlockHeight::new(0); + + // A creates a new order, B fills it, C freezes it, D concludes it + let random_tx_id = Id::new(H256::random_using(&mut rng)); + let random_utxo_outpoint = + UtxoOutPoint::new(OutPointSourceId::Transaction(random_tx_id), 0); + + let random_token_id = Id::new(H256::random_using(&mut rng)); + let order_data = OrderData::new( + Destination::AnyoneCanSpend, + OutputValue::Coin(Amount::from_atoms(10)), + OutputValue::TokenV1(random_token_id, Amount::from_atoms(10)), + ); + let order = TxOutput::CreateOrder(Box::new(order_data)); + let txa = TransactionBuilder::new() + .add_input( + TxInput::Utxo(random_utxo_outpoint), + InputWitness::NoSignature(None), + ) + .add_output(order) + .build(); + let txa_id = txa.transaction().get_id(); + let order_id = make_order_id(txa.transaction().inputs()).unwrap(); + + let random_utxo_outpoint2 = + UtxoOutPoint::new(OutPointSourceId::Transaction(random_tx_id), 1); + let fill_order = OrderAccountCommand::FillOrder(order_id, Amount::from_atoms(1)); + let txb = TransactionBuilder::new() + .add_input( + TxInput::Utxo(random_utxo_outpoint2), + InputWitness::NoSignature(None), + ) + .add_input( + TxInput::OrderAccountCommand(fill_order), + InputWitness::NoSignature(None), + ) + .add_anyone_can_spend_output(100) + .build(); + let txb_id = txb.transaction().get_id(); + + let random_utxo_outpoint3 = + UtxoOutPoint::new(OutPointSourceId::Transaction(random_tx_id), 2); + let freeze_order = OrderAccountCommand::FreezeOrder(order_id); + let txc = TransactionBuilder::new() + .add_input( + TxInput::Utxo(random_utxo_outpoint3), + InputWitness::NoSignature(None), + ) + .add_input( + TxInput::OrderAccountCommand(freeze_order), + InputWitness::NoSignature(None), + ) + .add_anyone_can_spend_output(100) + .build(); + let txc_id = txc.transaction().get_id(); + + let random_utxo_outpoint4 = + UtxoOutPoint::new(OutPointSourceId::Transaction(random_tx_id), 3); + let conclude_order = OrderAccountCommand::ConcludeOrder(order_id); + let txd = TransactionBuilder::new() + .add_input( + TxInput::Utxo(random_utxo_outpoint4), + InputWitness::NoSignature(None), + ) + .add_input( + TxInput::OrderAccountCommand(conclude_order), + InputWitness::NoSignature(None), + ) + .add_anyone_can_spend_output(100) + .build(); + let txd_id = txd.transaction().get_id(); + + let transactions = vec![txa, txb, txc, txd]; + let dependency_graph = build_dependency_graph(transactions, &chain_config, block_height); + assert_eq!(dependency_graph.len(), 4); + assert_eq!(dependency_graph[0].id, txa_id); + assert_eq!(dependency_graph[1].id, txb_id); + assert_eq!(dependency_graph[2].id, txc_id); + assert_eq!(dependency_graph[3].id, txd_id); + assert!(dependency_graph[0].dependencies.is_empty()); + assert_eq!(dependency_graph[1].dependencies, vec![txa_id]); + assert_eq!(dependency_graph[2].dependencies, vec![txa_id, txb_id]); + assert_eq!( + dependency_graph[3].dependencies, + vec![txa_id, txb_id, txc_id] + ); + } + + fn make_random_token_command(token_id: TokenId, rng: &mut impl Rng) -> AccountCommand { + match AccountCommandTag::iter().choose(rng).unwrap() { + AccountCommandTag::MintTokens => { + AccountCommand::MintTokens(token_id, Amount::from_atoms(rng.random_range(1..100))) + } + AccountCommandTag::UnmintTokens => AccountCommand::UnmintTokens(token_id), + AccountCommandTag::FreezeToken => { + AccountCommand::FreezeToken(token_id, IsTokenUnfreezable::Yes) + } + AccountCommandTag::UnfreezeToken => AccountCommand::UnfreezeToken(token_id), + AccountCommandTag::LockTokenSupply => AccountCommand::LockTokenSupply(token_id), + AccountCommandTag::ChangeTokenMetadataUri => { + AccountCommand::ChangeTokenMetadataUri(token_id, "URI".into()) + } + _ => AccountCommand::ChangeTokenAuthority(token_id, Destination::AnyoneCanSpend), + } + } +} diff --git a/api-server/web-server/src/tx_dependency_ordering/mod.rs b/api-server/web-server/src/tx_dependency_ordering/mod.rs new file mode 100644 index 000000000..3b443e740 --- /dev/null +++ b/api-server/web-server/src/tx_dependency_ordering/mod.rs @@ -0,0 +1,446 @@ +// 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. + +use std::collections::{BTreeMap, BinaryHeap}; + +use common::{chain::ChainConfig, chain::SignedTransaction, primitives::BlockHeight}; + +mod dependency_graph; + +use dependency_graph::{DependencyNode, build_dependency_graph}; + +// Order transactions by dependency between each other. +// Returns a Vec of transactions starting from the top-most parent transaction +// which doesn't depend on any other transaction following it, and ending with the leaves. +// +// On failure, the transactions are returned unsorted in the original (insertion) +// order along with the error, so that the caller can fall back to the insertion +// order without refetching a possibly different mempool snapshot. +pub fn order_transactions_by_dependency( + transactions: Vec, + chain_config: &ChainConfig, + block_height: BlockHeight, +) -> Result, (TopoSortError, Vec)> { + let graph = build_dependency_graph(transactions, chain_config, block_height); + + match topological_sort(graph) { + Ok(sorted_graph) => { + Ok(sorted_graph.into_iter().map(|node| node.into_signed_transaction()).collect()) + } + Err((err, graph)) => Err(( + err, + graph.into_iter().map(|node| node.into_signed_transaction()).collect(), + )), + } +} + +/// Errors that can occur during topological sorting. +#[derive(thiserror::Error, Debug, PartialEq, Eq)] +pub enum TopoSortError { + #[error("Circular dependency was detected")] + CycleDetected, + #[error("A node declared a dependency that is not present in the provided vector.")] + MissingDependency, + #[error("The same node id was provided more than once")] + DuplicateId, +} + +/// Sorts a vector of `DependencyNode`s topologically. +/// +/// Items with no dependencies (roots) will appear first in the resulting vector. +/// On failure, the nodes are returned unsorted in the original order together +/// with the error. +fn topological_sort(nodes: Vec) -> Result, (TopoSortError, Vec)> +where + T: DependencyNode, +{ + struct QueueItem { + idx: usize, + priority: P, + } + + impl PartialEq for QueueItem

{ + fn eq(&self, other: &Self) -> bool { + self.idx == other.idx + } + } + + impl Eq for QueueItem

{} + + impl Ord for QueueItem

{ + fn cmp(&self, other: &Self) -> std::cmp::Ordering { + // The tie-breaker on the index makes the pop order of the equal-priority + // items deterministic: the items earlier in the input vector come first. + self.priority.cmp(&other.priority).then_with(|| other.idx.cmp(&self.idx)) + } + } + + impl PartialOrd for QueueItem

{ + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } + } + + let n = nodes.len(); + if n <= 1 { + return Ok(nodes); + } + + // Map each node's ID to its index in the original vector. + let mut id_to_index = BTreeMap::new(); + for i in 0..n { + // A duplicated id would make the failure surface later as a bogus + // cycle (the earlier node with the same id could never be popped), + // so it is reported distinctly here. + if id_to_index.insert(nodes[i].id(), i).is_some() { + return Err((TopoSortError::DuplicateId, nodes)); + } + } + + // Adjacency list: dependents[i] contains indices of nodes that depend on node i. + let mut dependents: Vec> = vec![Vec::new(); n]; + // Indegree: indegrees[i] is the number of unresolved dependencies node i has. + let mut indegrees: Vec = vec![0; n]; + + // Build the graph + for i in 0..n { + for dep_id in nodes[i].dependencies() { + let Some(&dep_index) = id_to_index.get(dep_id) else { + return Err((TopoSortError::MissingDependency, nodes)); + }; + + dependents[dep_index].push(i); + indegrees[i] += 1; + } + } + + // Start with all nodes that have 0 dependencies (the "roots") + let mut queue = BinaryHeap::new(); + for (idx, node) in nodes.iter().enumerate() { + if indegrees[idx] == 0 { + queue.push(QueueItem { + idx, + priority: node.priority(), + }); + } + } + + let mut sorted_indices = Vec::with_capacity(n); + + while let Some(QueueItem { + idx: current_idx, .. + }) = queue.pop() + { + sorted_indices.push(current_idx); + + // For every node that depends on the current node, remove the dependency edge + for &dependent_idx in &dependents[current_idx] { + indegrees[dependent_idx] -= 1; + + // If the dependent node now has no pending dependencies, it's ready to be processed + if indegrees[dependent_idx] == 0 { + queue.push(QueueItem { + idx: dependent_idx, + priority: nodes[dependent_idx].priority(), + }); + } + } + } + + // If we haven't sorted all items, there must be a cycle + if sorted_indices.len() != n { + return Err((TopoSortError::CycleDetected, nodes)); + } + + // Reconstruct the sorted vector without cloning `T` + // We wrap the original items in Option, and `take()` them out in sorted order. + let mut wrapped_nodes: Vec> = nodes.into_iter().map(Some).collect(); + + let sorted_nodes = sorted_indices + .into_iter() + .map(|idx| wrapped_nodes[idx].take().expect("present")) + .collect(); + + Ok(sorted_nodes) +} + +#[cfg(test)] +mod tests { + use super::*; + use randomness::SliceRandom; + use test_utils::random::Seed; + + use crate::tx_dependency_ordering::dependency_graph::TxPriorityOrder; + + use rstest::rstest; + + #[derive(Debug, PartialEq, Eq, Clone)] + struct DummyNode { + priority: TxPriorityOrder, + id: u32, + dependencies: Vec, + } + + impl DependencyNode for DummyNode { + type Id = u32; + type Priority = TxPriorityOrder; + + fn id(&self) -> Self::Id { + self.id + } + + fn priority(&self) -> Self::Priority { + self.priority + } + + fn dependencies(&self) -> &[Self::Id] { + &self.dependencies + } + } + + #[rstest] + #[trace] + #[case(Seed::from_entropy())] + fn test_priority_ordering(#[case] seed: Seed) { + let mut rng = test_utils::random::make_seedable_rng(seed); + let root_node = DummyNode { + priority: TxPriorityOrder::Highest, + id: 1, + dependencies: vec![], + }; + let root_freeze_node = DummyNode { + priority: TxPriorityOrder::TokenFreeze, + id: 4, + dependencies: vec![], + }; + let highest_dependent_node = DummyNode { + priority: TxPriorityOrder::Highest, + id: 5, + dependencies: vec![1], + }; + let delegation_stake_node = DummyNode { + priority: TxPriorityOrder::DelegationStake, + id: 2, + dependencies: vec![1], + }; + let delegation_withdrawal_node = DummyNode { + priority: TxPriorityOrder::DelegationWithdrawal, + id: 3, + dependencies: vec![1], + }; + let expected_sorted_ids = vec![ + // should be first as everyone depends on it + root_node.id, + // those depend on the root but internally will be ordered highest, stake then withdrawal + highest_dependent_node.id, + delegation_stake_node.id, + delegation_withdrawal_node.id, + // even though this has no dependencies it should still be last by priority + root_freeze_node.id, + ]; + let mut nodes = vec![ + root_freeze_node, + delegation_withdrawal_node, + delegation_stake_node, + highest_dependent_node, + root_node, + ]; + nodes.shuffle(&mut rng); + let sorted_nodes = topological_sort(nodes).unwrap(); + let sorted_ids = sorted_nodes.iter().map(|node| node.id).collect::>(); + + assert_eq!(sorted_ids, expected_sorted_ids); + } + + #[rstest] + #[trace] + #[case(Seed::from_entropy())] + fn test_dependency_ordering(#[case] seed: Seed) { + let mut rng = test_utils::random::make_seedable_rng(seed); + let root_node = DummyNode { + priority: TxPriorityOrder::Highest, + id: 1, + dependencies: vec![], + }; + let dependent_node = DummyNode { + priority: TxPriorityOrder::TokenFreeze, + id: 2, + dependencies: vec![1], + }; + // even though this has higher priority than TokenFreeze it still depends on it + let dependent_node2 = DummyNode { + priority: TxPriorityOrder::Highest, + id: 3, + dependencies: vec![2], + }; + let mut nodes = vec![dependent_node, root_node, dependent_node2]; + nodes.shuffle(&mut rng); + let sorted_nodes = topological_sort(nodes).unwrap(); + let sorted_ids = sorted_nodes.iter().map(|node| node.id).collect::>(); + + assert_eq!(sorted_ids, vec![1, 2, 3]); + } + + #[test] + fn test_diamond_dependency_pattern() { + // Graph: A -> B, A -> C, B -> D, C -> D + let a = DummyNode { + priority: TxPriorityOrder::Highest, + id: 1, + dependencies: vec![], + }; + let b = DummyNode { + priority: TxPriorityOrder::Highest, + id: 2, + dependencies: vec![1], + }; + let c = DummyNode { + priority: TxPriorityOrder::Highest, + id: 3, + dependencies: vec![1], + }; + let d = DummyNode { + priority: TxPriorityOrder::Highest, + id: 4, + dependencies: vec![2, 3], + }; + + let sorted = topological_sort(vec![d, b, a, c]).unwrap(); + let sorted_ids = sorted.iter().map(|n| n.id).collect::>(); + + assert_eq!(sorted_ids.first(), Some(&1)); // A must be first + assert_eq!(sorted_ids.last(), Some(&4)); // D must be last + } + + #[test] + fn test_disconnected_components() { + // Graph: A -> B (Chain 1) and C -> D (Chain 2) + let a = DummyNode { + priority: TxPriorityOrder::Highest, + id: 1, + dependencies: vec![], + }; + let b = DummyNode { + priority: TxPriorityOrder::Highest, + id: 2, + dependencies: vec![1], + }; + let c = DummyNode { + priority: TxPriorityOrder::Highest, + id: 3, + dependencies: vec![], + }; + let d = DummyNode { + priority: TxPriorityOrder::Highest, + id: 4, + dependencies: vec![3], + }; + + let sorted = topological_sort(vec![d, b, a, c]).unwrap(); + let sorted_ids = sorted.iter().map(|n| n.id).collect::>(); + + // Dependencies must be respected + let pos_a = sorted_ids.iter().position(|&id| id == 1).unwrap(); + let pos_b = sorted_ids.iter().position(|&id| id == 2).unwrap(); + let pos_c = sorted_ids.iter().position(|&id| id == 3).unwrap(); + let pos_d = sorted_ids.iter().position(|&id| id == 4).unwrap(); + + assert!(pos_a < pos_b); + assert!(pos_c < pos_d); + } + + #[test] + fn test_cycle_detection() { + let root_node = DummyNode { + priority: TxPriorityOrder::Highest, + id: 1, + dependencies: vec![2], + }; + let dependent_node = DummyNode { + priority: TxPriorityOrder::Highest, + id: 2, + dependencies: vec![1], + }; + let nodes = vec![root_node.clone(), dependent_node.clone()]; + let (err, unsorted) = topological_sort(nodes).unwrap_err(); + + assert_eq!(err, TopoSortError::CycleDetected); + // The nodes are returned unsorted in the original order. + assert_eq!(unsorted, vec![root_node, dependent_node]); + } + + #[test] + fn test_missing_dependency() { + let node1 = DummyNode { + priority: TxPriorityOrder::Highest, + id: 1, + dependencies: vec![2], + }; + let node2 = DummyNode { + priority: TxPriorityOrder::Highest, + id: 2, + // 3 is not in the nodes list + dependencies: vec![3], + }; + let nodes = vec![node1.clone(), node2.clone()]; + let (err, unsorted) = topological_sort(nodes).unwrap_err(); + + assert_eq!(err, TopoSortError::MissingDependency); + assert_eq!(unsorted, vec![node1, node2]); + } + + #[test] + fn test_duplicate_id() { + // A duplicated id would be indistinguishable from a cycle without the + // dedicated check: the earlier node could never be popped. + let node1 = DummyNode { + priority: TxPriorityOrder::Highest, + id: 1, + dependencies: vec![], + }; + let node2 = DummyNode { + priority: TxPriorityOrder::Highest, + id: 1, + dependencies: vec![], + }; + let nodes = vec![node1.clone(), node2.clone()]; + let (err, unsorted) = topological_sort(nodes).unwrap_err(); + + assert_eq!(err, TopoSortError::DuplicateId); + assert_eq!(unsorted, vec![node1, node2]); + } + + #[test] + fn test_empty_input() { + let nodes: Vec = vec![]; + let sorted_nodes = topological_sort(nodes).unwrap(); + + assert!(sorted_nodes.is_empty()); + } + + #[test] + fn test_single_node() { + let root_node = DummyNode { + priority: TxPriorityOrder::Highest, + id: 1, + dependencies: vec![], + }; + let nodes = vec![root_node]; + let sorted_nodes = topological_sort(nodes).unwrap(); + let sorted_ids = sorted_nodes.iter().map(|node| node.id).collect::>(); + + assert_eq!(sorted_ids, vec![1]); + } +}