RPC: add list_mempool

app/rpc_server.rs

@@ -254,6 +254,22 @@ impl<const ENABLE_PRIVATE_API: bool> rpc_api::node::RpcServer
         Ok(peers)
     }
 
+    async fn list_mempool(&self) -> RpcResult<Vec<rpc_api::node::MempoolTx>> {
+        let txs = self.app.node.get_all_transactions().map_err(custom_err)?;
+        let res = txs
+            .into_iter()
+            .map(|authorized| {
+                let tx = authorized.transaction;
+                rpc_api::node::MempoolTx {
+                    txid: tx.txid(),
+                    size: tx.canonical_size(),
+                    tx,
+                }
+            })
+            .collect();
+        Ok(res)
+    }
+
     async fn list_utxos(&self) -> RpcResult<Vec<PointedOutput>> {
         let utxos = self.app.node.get_all_utxos().map_err(custom_err)?;
         let res = utxos

cli/lib.rs

@@ -113,6 +113,8 @@ pub enum Command {
     },
     /// Get the height of the latest failed withdrawal bundle
     LatestFailedWithdrawalBundleHeight,
+    /// List the transactions the mempool holds
+    ListMempool,
     /// List peers
     ListPeers,
     /// List all UTXOs
@@ -298,6 +300,10 @@ where
                 rpc_client.latest_failed_withdrawal_bundle_height().await?;
             serde_json::to_string_pretty(&height)?
         }
+        Command::ListMempool => {
+            let txs = rpc_client.list_mempool().await?;
+            serde_json::to_string_pretty(&txs)?
+        }
         Command::ListPeers => {
             let peers = rpc_client.list_peers().await?;
             serde_json::to_string_pretty(&peers)?

integration_tests/integration_test.rs

@@ -14,6 +14,7 @@ use thunder_app_rpc_api::node::RpcClient as _;
 use crate::{
     block_template::block_template_trial,
     ibd::ibd_trial,
+    list_mempool::list_mempool_trial,
     setup::{Init, PostSetup},
     unknown_withdrawal::unknown_withdrawal_trial,
     util::BinPaths,
@@ -178,6 +179,11 @@ pub fn tests(
             file_registry.clone(),
             failure_collector.clone(),
         ),
+        list_mempool_trial(
+            bin_paths.clone(),
+            file_registry.clone(),
+            failure_collector.clone(),
+        ),
         unknown_withdrawal_trial(bin_paths, file_registry, failure_collector),
     ]
 }

integration_tests/list_mempool.rs

@@ -0,0 +1,149 @@
+//! Test that a node lists the transactions its mempool holds
+
+use bip300301_enforcer_integration_tests::{
+    integration_test::{
+        activate_sidechain, deposit, fund_enforcer, propose_sidechain,
+    },
+    setup::{
+        Mode, Network, PostSetup as EnforcerPostSetup,
+        PreSetup as EnforcerPreSetup, SetupOpts as EnforcerSetupOpts,
+        Sidechain as _,
+    },
+    util::{
+        AbortOnDrop, AsyncTrial, BinPaths as EnforcerBinPaths,
+        TestFailureCollector, TestFileRegistry,
+    },
+};
+use bitcoin::Amount;
+use futures::{
+    FutureExt as _, StreamExt as _, channel::mpsc, future::BoxFuture,
+};
+use thunder_app_rpc_api::{node::RpcClient as _, wallet::RpcClient as _};
+use tokio::time::sleep;
+use tracing::Instrument as _;
+
+use crate::{
+    setup::{Init, PostSetup},
+    util::BinPaths,
+};
+
+const DEPOSIT_AMOUNT: Amount = Amount::from_sat(21_000_000);
+const DEPOSIT_FEE: Amount = Amount::from_sat(1_000_000);
+const TRANSFER_AMOUNT: u64 = 1_000_000;
+const TRANSFER_FEE: u64 = 1_000;
+
+/// Initial setup for the test
+async fn setup(
+    enforcer_bin_paths: &EnforcerBinPaths,
+    res_tx: mpsc::UnboundedSender<anyhow::Result<()>>,
+) -> anyhow::Result<EnforcerPostSetup> {
+    let enforcer_pre_setup =
+        EnforcerPreSetup::new(enforcer_bin_paths, Network::Regtest)?;
+    let mut enforcer_post_setup = {
+        let setup_opts: EnforcerSetupOpts = Default::default();
+        enforcer_pre_setup
+            .setup(Mode::Mempool, setup_opts, res_tx.clone())
+            .await?
+    };
+    let () = propose_sidechain::<PostSetup>(&mut enforcer_post_setup).await?;
+    let () = activate_sidechain::<PostSetup>(&mut enforcer_post_setup).await?;
+    let () = fund_enforcer::<PostSetup>(&mut enforcer_post_setup).await?;
+    Ok(enforcer_post_setup)
+}
+
+async fn list_mempool_task(
+    bin_paths: BinPaths,
+    res_tx: mpsc::UnboundedSender<anyhow::Result<()>>,
+) -> anyhow::Result<()> {
+    let mut enforcer_post_setup =
+        setup(&bin_paths.others, res_tx.clone()).await?;
+    let mut sidechain = PostSetup::setup(
+        Init {
+            thunder_app: bin_paths.thunder()?.clone(),
+            data_dir_suffix: None,
+        },
+        &enforcer_post_setup,
+        res_tx,
+    )
+    .await?;
+    tracing::info!("Setup thunder node successfully");
+
+    tracing::debug!("Checking that a fresh mempool is empty");
+    anyhow::ensure!(sidechain.rpc_client.list_mempool().await?.is_empty());
+
+    let deposit_address = sidechain.get_deposit_address().await?;
+    let () = deposit(
+        &mut enforcer_post_setup,
+        &mut sidechain,
+        &deposit_address,
+        DEPOSIT_AMOUNT,
+        DEPOSIT_FEE,
+    )
+    .await?;
+    tracing::info!("Deposited to sidechain successfully");
+
+    tracing::debug!("Checking that a deposit alone leaves the mempool empty");
+    anyhow::ensure!(sidechain.rpc_client.list_mempool().await?.is_empty());
+
+    let dest = sidechain.rpc_client.get_new_address().await?;
+    let txid = sidechain
+        .rpc_client
+        .create_transfer(dest, TRANSFER_AMOUNT, TRANSFER_FEE)
+        .await?;
+    tracing::info!(%txid, "Created a transfer");
+
+    tracing::debug!("Checking that the mempool holds the transfer");
+    let mempool = sidechain.rpc_client.list_mempool().await?;
+    anyhow::ensure!(mempool.len() == 1);
+    let entry = &mempool[0];
+    anyhow::ensure!(entry.txid == txid);
+    // The txid hashes over the canonical encoding, so both agree with the
+    // transaction the entry carries.
+    anyhow::ensure!(entry.tx.txid() == txid);
+    anyhow::ensure!(entry.size == entry.tx.canonical_size());
+    anyhow::ensure!(entry.size > 0);
+
+    tracing::debug!("Checking that a block empties the mempool");
+    let () = sidechain.bmm_single(&mut enforcer_post_setup).await?;
+    anyhow::ensure!(sidechain.rpc_client.list_mempool().await?.is_empty());
+
+    drop(sidechain);
+    tracing::info!(
+        "Removing {}",
+        enforcer_post_setup.directories.base_dir.path().display()
+    );
+    drop(enforcer_post_setup.tasks);
+    // Wait for tasks to die
+    sleep(std::time::Duration::from_secs(1)).await;
+    enforcer_post_setup.directories.base_dir.cleanup()?;
+    Ok(())
+}
+
+async fn list_mempool(bin_paths: BinPaths) -> anyhow::Result<()> {
+    let (res_tx, mut res_rx) = mpsc::unbounded();
+    let _test_task: AbortOnDrop<()> = tokio::task::spawn({
+        let res_tx = res_tx.clone();
+        async move {
+            let res = list_mempool_task(bin_paths, res_tx.clone()).await;
+            let _send_err: Result<(), _> = res_tx.unbounded_send(res);
+        }
+        .in_current_span()
+    })
+    .into();
+    res_rx.next().await.ok_or_else(|| {
+        anyhow::anyhow!("Unexpected end of test task result stream")
+    })?
+}
+
+pub fn list_mempool_trial(
+    bin_paths: BinPaths,
+    file_registry: TestFileRegistry,
+    failure_collector: TestFailureCollector,
+) -> AsyncTrial<BoxFuture<'static, anyhow::Result<()>>> {
+    AsyncTrial::new(
+        "list_mempool",
+        list_mempool(bin_paths).boxed(),
+        file_registry,
+        failure_collector,
+    )
+}

integration_tests/main.rs

@@ -7,6 +7,7 @@ use tracing_subscriber::{filter as tracing_filter, layer::SubscriberExt};
 mod block_template;
 mod ibd;
 mod integration_test;
+mod list_mempool;
 mod setup;
 mod unknown_withdrawal;
 mod util;

rpc-api/lib.rs

@@ -77,6 +77,16 @@ pub mod node {
         pub block_hash: Option<BlockHash>,
     }
 
+    /// One transaction the mempool holds
+    #[derive(Clone, Debug, Deserialize, Serialize, ToSchema)]
+    pub struct MempoolTx {
+        /// Blake3 over the canonical encoding
+        pub txid: Txid,
+        /// Canonical size in bytes
+        pub size: u64,
+        pub tx: Transaction,
+    }
+
     #[open_api(ref_schemas[
         Address, MerkleRoot, OutPoint, Output, OutputContent, Txid,
         schema::BitcoinTxid, thunder_schema::BitcoinAddr,
@@ -164,6 +174,10 @@ pub mod node {
             &self,
         ) -> RpcResult<Option<u32>>;
 
+        /// List the transactions the mempool holds, in no particular order.
+        #[method(name = "list_mempool")]
+        async fn list_mempool(&self) -> RpcResult<Vec<MempoolTx>>;
+
         /// List peers
         #[method(name = "list_peers")]
         async fn list_peers(&self) -> RpcResult<Vec<Peer>>;