test(e2e): add mempool coverage (#2799)

.github/workflows/end-to-end-with-cucumber.yml

@@ -191,6 +191,17 @@ jobs:
           verbose_console: "true"
           max_concurrent_scenarios: "2"
 
+      # Feature: Mempool (@mempool_ci)
+      - name: Run cucumber suite - mempool
+        if: ${{ !cancelled() && steps.verify_local_ntp.conclusion == 'success' }}
+        timeout-minutes: 60
+        uses: ./.github/actions/run-cucumber-suite
+        with:
+          suite_tag: "@mempool_ci"
+          suite_name: mempool
+          artifact_subdir: Mempool lifecycle
+          verbose_console: "true"
+
       # Feature: Zone SDK (@zone_ci)
       - name: Run cucumber suite - zone
         if: ${{ !cancelled() && steps.verify_local_ntp.conclusion == 'success' }}

nodes/api-common/src/paths.rs

@@ -39,3 +39,4 @@ pub mod admin {
 // testing paths
 pub const UPDATE_MEMBERSHIP: &str = "/test/membership/update";
 pub const DIAL_PEER: &str = "/test/network/dial_peer";
+pub const TEST_MEMPOOL_VIEW: &str = "/test/mempool/view";

nodes/node/binary/src/api/testing/backend.rs

@@ -10,7 +10,7 @@ use axum::{
 };
 use http::StatusCode;
 use lb_api_service::Backend;
-use lb_http_api_common::paths::{DIAL_PEER, MANTLE_SDP_DECLARATIONS};
+use lb_http_api_common::paths::{DIAL_PEER, MANTLE_SDP_DECLARATIONS, TEST_MEMPOOL_VIEW};
 use lb_network_service::{NetworkService, backends::libp2p::Libp2p as NetworkBackend};
 use overwatch::{overwatch::handle::OverwatchHandle, services::AsServiceId};
 use tokio::net::TcpListener;
@@ -26,7 +26,7 @@ use tracing::Level as TracingLevel;
 use crate::{
     api::{
         backend::AxumBackendSettings,
-        testing::handlers::{dial_peer, get_sdp_declarations},
+        testing::handlers::{dial_peer, get_sdp_declarations, test_mempool_view},
     },
     generic_services::{self, SdpService},
 };
@@ -86,6 +86,10 @@ where
                 MANTLE_SDP_DECLARATIONS,
                 get(get_sdp_declarations::<RuntimeServiceId>),
             )
+            .route(
+                TEST_MEMPOOL_VIEW,
+                get(test_mempool_view::<RuntimeServiceId>),
+            )
             .route(DIAL_PEER, post(dial_peer::<RuntimeServiceId>))
             .with_state(handle)
             .layer(axum::extract::DefaultBodyLimit::max(

nodes/node/binary/src/api/testing/handlers.rs

@@ -1,11 +1,18 @@
 use std::fmt::{Debug, Display};
 
 use axum::{Json, extract::State, response::Response};
-use lb_api_service::http::{libp2p, mantle};
+use futures::StreamExt as _;
+use lb_api_service::http::{consensus, libp2p, mantle};
+use lb_core::{
+    header::HeaderId,
+    mantle::{SignedMantleTx, Transaction as _, TxHash},
+};
 use lb_libp2p::{Multiaddr, PeerId};
 use lb_network_service::{NetworkService, backends::libp2p::Libp2p as NetworkBackend};
-use overwatch::{overwatch::OverwatchHandle, services::AsServiceId};
+use lb_tx_service::MempoolMsg;
+use overwatch::{DynError, overwatch::OverwatchHandle, services::AsServiceId};
 use serde::{Deserialize, Serialize};
+use tokio::sync::oneshot;
 
 use super::backend::TestHttpCryptarchiaService;
 use crate::{
@@ -34,6 +41,64 @@ where
     make_request_and_return_response!(mantle::get_sdp_declarations::<RuntimeServiceId>(&handle))
 }
 
+pub async fn test_mempool_view<RuntimeServiceId>(
+    State(handle): State<OverwatchHandle<RuntimeServiceId>>,
+) -> Response
+where
+    RuntimeServiceId: Debug
+        + Send
+        + Sync
+        + Display
+        + 'static
+        + AsServiceId<TestHttpCryptarchiaService<RuntimeServiceId>>
+        + AsServiceId<SdpService<RuntimeServiceId>>
+        + AsServiceId<TxMempoolService<RuntimeServiceId>>,
+{
+    make_request_and_return_response!(current_tip_mempool_view::<RuntimeServiceId>(&handle))
+}
+
+async fn current_tip_mempool_view<RuntimeServiceId>(
+    handle: &OverwatchHandle<RuntimeServiceId>,
+) -> Result<Vec<TxHash>, DynError>
+where
+    RuntimeServiceId: Debug
+        + Send
+        + Sync
+        + Display
+        + 'static
+        + AsServiceId<TestHttpCryptarchiaService<RuntimeServiceId>>
+        + AsServiceId<SdpService<RuntimeServiceId>>
+        + AsServiceId<TxMempoolService<RuntimeServiceId>>,
+{
+    let consensus = consensus::cryptarchia_info::<RuntimeServiceId>(handle).await?;
+
+    mempool_view_at::<RuntimeServiceId>(handle, consensus.cryptarchia_info.tip).await
+}
+
+async fn mempool_view_at<RuntimeServiceId>(
+    handle: &OverwatchHandle<RuntimeServiceId>,
+    ancestor_hint: HeaderId,
+) -> Result<Vec<TxHash>, DynError>
+where
+    RuntimeServiceId:
+        Debug + Send + Sync + Display + 'static + AsServiceId<TxMempoolService<RuntimeServiceId>>,
+{
+    let relay = handle.relay::<TxMempoolService<RuntimeServiceId>>().await?;
+    let (sender, receiver) = oneshot::channel();
+
+    relay
+        .send(MempoolMsg::View {
+            ancestor_hint,
+            reply_channel: sender,
+        })
+        .await
+        .map_err(|(error, _)| error)?;
+
+    let txs = receiver.await?;
+
+    Ok(txs.map(|tx: SignedMantleTx| tx.hash()).collect().await)
+}
+
 pub async fn dial_peer<RuntimeServiceId>(
     State(handle): State<OverwatchHandle<RuntimeServiceId>>,
     Json(req): Json<DialPeerRequestBody>,
@@ -51,6 +116,6 @@ where
 {
     make_request_and_return_response!(async move {
         let peer_id: PeerId = libp2p::connect_peer::<RuntimeServiceId>(&handle, req.addr).await?;
-        Ok::<PeerId, overwatch::DynError>(peer_id)
+        Ok::<PeerId, DynError>(peer_id)
     })
 }

tests/cucumber_tests/features/mempool.feature

@@ -0,0 +1,145 @@
+Feature: Mempool lifecycle
+
+  @mempool_ci @mempool_manual
+  Scenario: Submitted transaction becomes visible in current-tip mempool view
+    Given the genesis block has the following wallet resources:
+      | account_index | token_count | token_amount |
+      | 1             | 2           | 1000         |
+      | 2             | 0           | 0            |
+    And I have a cluster with capacity of 1 nodes
+    And no nodes are declared as blend providers
+    And I start nodes with wallet resources:
+      | node_name | account_index | wallet_name | connected_to |
+      | NODE_1    | 1             | WALLET_A    |              |
+      | NODE_1    | 2             | WALLET_B    |              |
+    When node "NODE_1" is at height 2 in 240 seconds
+    And I prepare transfer transaction "TX_VISIBLE" of 100 LGO from wallet "WALLET_A" to wallet "WALLET_B"
+    And I submit prepared transaction "TX_VISIBLE" to nodes:
+      | node_name |
+      | NODE_1    |
+    Then transaction "TX_VISIBLE" is pending in mempool of nodes in 30 seconds:
+      | node_name |
+      | NODE_1    |
+    Then I stop all nodes
+
+  @mempool_ci @mempool_manual
+  Scenario: Included transaction is removed from mempool view
+    Given the genesis block has the following wallet resources:
+      | account_index | token_count | token_amount |
+      | 1             | 2           | 1000         |
+      | 2             | 0           | 0            |
+    And I have a cluster with capacity of 1 nodes
+    And no nodes are declared as blend providers
+    And I start nodes with wallet resources:
+      | node_name | account_index | wallet_name | connected_to |
+      | NODE_1    | 1             | WALLET_A    |              |
+      | NODE_1    | 2             | WALLET_B    |              |
+    When node "NODE_1" is at height 2 in 240 seconds
+    And I submit funded transfer transaction "TX_INCLUDED" of 100 LGO from wallet "WALLET_A" to wallet "WALLET_B"
+    Then transaction "TX_INCLUDED" is included on node "NODE_1" in 240 seconds
+    And transaction "TX_INCLUDED" is not pending in mempool of all nodes in 30 seconds
+    And transaction "TX_INCLUDED" remains not pending in mempool of all nodes for 2 blocks in 180 seconds
+    Then I stop all nodes
+
+  @mempool_ci @mempool_manual
+  Scenario: Pending transaction survives restart before inclusion
+    Given the genesis block has the following wallet resources:
+      | account_index | token_count | token_amount |
+      | 1             | 2           | 1000         |
+      | 2             | 0           | 0            |
+    # Keep block production slow enough that the tx stays pending across restart.
+    And I have deployment config override "time.slot_duration" as "seconds(60)"
+    And I have deployment config override "cryptarchia.slot_activation_coeff.numerator" as "9"
+    And I have user config override "cryptarchia.service.bootstrap.prolonged_bootstrap_period" as "seconds(0)"
+    And I have a cluster with capacity of 1 nodes
+    And no nodes are declared as blend providers
+    And I start nodes with wallet resources:
+      | node_name | account_index | wallet_name | connected_to |
+      | NODE_1    | 1             | WALLET_A    |              |
+      | NODE_1    | 2             | WALLET_B    |              |
+    When node "NODE_1" is at height 2 in 360 seconds
+    And I prepare transfer transaction "TX_PENDING_RESTART" of 100 LGO from wallet "WALLET_A" to wallet "WALLET_B"
+    And I submit prepared transaction "TX_PENDING_RESTART" to nodes:
+      | node_name |
+      | NODE_1    |
+    Then transaction "TX_PENDING_RESTART" is pending in mempool of nodes in 30 seconds:
+      | node_name |
+      | NODE_1    |
+    When I restart node "NODE_1"
+    # TF only keeps the tx hash here; the node must recover the pending mempool entry.
+    Then transaction "TX_PENDING_RESTART" is pending in mempool of nodes in 30 seconds:
+      | node_name |
+      | NODE_1    |
+    Then I stop all nodes
+
+  @mempool_ci @mempool_manual
+  Scenario: Included transaction stays absent from mempool after restart
+    Given the genesis block has the following wallet resources:
+      | account_index | token_count | token_amount |
+      | 1             | 2           | 1000         |
+      | 2             | 0           | 0            |
+    And I have a cluster with capacity of 1 nodes
+    And no nodes are declared as blend providers
+    And I start nodes with wallet resources:
+      | node_name | account_index | wallet_name | connected_to |
+      | NODE_1    | 1             | WALLET_A    |              |
+      | NODE_1    | 2             | WALLET_B    |              |
+    When node "NODE_1" is at height 2 in 240 seconds
+    And I submit funded transfer transaction "TX_RESTART" of 100 LGO from wallet "WALLET_A" to wallet "WALLET_B"
+    Then transaction "TX_RESTART" is included on node "NODE_1" in 240 seconds
+    When I restart node "NODE_1"
+    And node "NODE_1" is at height 3 in 240 seconds
+    # Included txs must not be recovered back into the pending mempool after restart.
+    Then transaction "TX_RESTART" is not pending in mempool of all nodes in 30 seconds
+    Then I stop all nodes
+
+  @mempool_ci @mempool_manual
+  Scenario: Invalid transaction is not retained in mempool
+    Given I have a cluster with capacity of 1 nodes
+    And no nodes are declared as blend providers
+    And I start node "NODE_1"
+    When node "NODE_1" is at height 2 in 240 seconds
+    And I try to submit invalid transaction "TX_INVALID" to node "NODE_1"
+    Then transaction "TX_INVALID" is not included in 30 seconds
+    And transaction "TX_INVALID" is not pending in mempool of all nodes in 30 seconds
+    Then I stop all nodes
+
+  @mempool_ci @mempool_manual
+  Scenario: Same transaction from competing forks is not included again after convergence
+    Given the genesis block has the following wallet resources:
+      | account_index | token_count | token_amount |
+      | 1             | 2           | 1000         |
+      | 2             | 0           | 0            |
+    And I have a cluster with capacity of 3 nodes
+    And no nodes are declared as blend providers
+    And we use IBD peers
+    And all peers must be mode online after startup in 30 seconds
+    And we will have distinct node groups to query wallet balances:
+      | group_name | node_name |
+      | FORK_A     | NODE_A1   |
+      | FORK_B     | NODE_B1   |
+    And I start nodes with wallet resources:
+      | node_name | account_index | wallet_name | connected_to |
+      | NODE_A1   | 1             | WALLET_A    |              |
+      | NODE_A1   | 2             | WALLET_B    |              |
+    And I start node "NODE_B1"
+    When node "NODE_A1" is at height 2 in 240 seconds
+    And node "NODE_B1" is at height 2 in 240 seconds
+    And I prepare transfer transaction "TX_SHARED" of 100 LGO from wallet "WALLET_A" to wallet "WALLET_B"
+    And I submit prepared transaction "TX_SHARED" to nodes:
+      | node_name |
+      | NODE_A1   |
+      | NODE_B1   |
+    Then transaction "TX_SHARED" is pending in mempool of nodes in 60 seconds:
+      | node_name |
+      | NODE_A1   |
+      | NODE_B1   |
+    Then transaction "TX_SHARED" is included on node "NODE_A1" in 240 seconds
+    And transaction "TX_SHARED" is included on node "NODE_B1" in 240 seconds
+    When node "NODE_A1" is at height 5 in 180 seconds
+    And node "NODE_B1" is at height 5 in 180 seconds
+    And I start peer node "NODE_JOIN" connected to node "NODE_A1" and node "NODE_B1"
+    Then all nodes have at least 8 blocks and converged to within 1 blocks in 240 seconds
+    And transaction "TX_SHARED" is not pending in mempool of all nodes in 180 seconds
+    And transaction "TX_SHARED" remains not pending in mempool of all nodes for 3 blocks in 240 seconds
+    Then I stop all nodes

tests/src/cucumber/steps/manual_mempool/actions.rs

@@ -0,0 +1,185 @@
+use lb_core::mantle::{SignedMantleTx, Transaction as _, TxHash};
+use lb_key_management_system_service::keys::ZkPublicKey;
+use tracing::{info, warn};
+
+use crate::{
+    common::wallet::{WalletTransactionError, WalletTransactionIntent},
+    cucumber::{
+        error::StepError,
+        fee_reserve::DEFAULT_STORAGE_GAS_PRICE,
+        steps::{TARGET, manual_transactions::tracked_transactions::create_invalid_transaction},
+        wallet::submissions::{
+            SignedUserWalletSubmission, prepare_user_wallet_transaction_submission,
+            record_signed_user_wallet_submission, sign_prepared_user_wallet_transaction,
+        },
+        world::CucumberWorld,
+    },
+};
+
+pub async fn prepare_transfer_transaction(
+    world: &mut CucumberWorld,
+    step: &str,
+    transaction_alias: String,
+    amount: u64,
+    sender_wallet_name: String,
+    receiver_wallet_name: String,
+) -> Result<(), StepError> {
+    let receiver_pk = receiver_public_key(world, step, &receiver_wallet_name)?;
+    let intent = transfer_intent(receiver_pk, amount)?;
+    let prepared =
+        prepare_user_wallet_transaction_submission(world, step, &sender_wallet_name, intent, None)
+            .await?;
+    let signed = sign_prepared_user_wallet_transaction(step, prepared, Vec::new())?;
+    let tx_hash = record_prepared_transaction(world, transaction_alias.clone(), &signed);
+
+    report_prepared_transaction(
+        &transaction_alias,
+        tx_hash,
+        &sender_wallet_name,
+        &receiver_wallet_name,
+    );
+
+    Ok(())
+}
+
+pub async fn submit_prepared_transaction_to_nodes(
+    world: &CucumberWorld,
+    step: &str,
+    transaction_alias: String,
+    node_names: Vec<String>,
+) -> Result<(), StepError> {
+    let signed_tx = world.resolve_prepared_transaction(&transaction_alias)?;
+    let tx_hash = signed_tx.hash();
+
+    for node_name in node_names {
+        submit_prepared_transaction_to_node(
+            world,
+            step,
+            &transaction_alias,
+            &signed_tx,
+            tx_hash,
+            &node_name,
+        )
+        .await?;
+    }
+
+    Ok(())
+}
+
+pub async fn try_submit_invalid_transaction(
+    world: &mut CucumberWorld,
+    step: &str,
+    transaction_alias: String,
+    node_name: String,
+) -> Result<(), StepError> {
+    let node = world
+        .resolve_node_http_client(&node_name)
+        .inspect_err(|e| {
+            warn!(target: TARGET, "Step `{step}` error: {e}");
+        })?;
+    let signed_tx = create_invalid_transaction();
+    let tx_hash = signed_tx.hash();
+
+    world.remember_submitted_transaction(transaction_alias.clone(), tx_hash);
+
+    match node.submit_transaction(&signed_tx).await {
+        Ok(()) => {
+            info!(
+                target: TARGET,
+                "Submitted invalid transaction `{transaction_alias}` ({:?}) to `{node_name}`",
+                tx_hash
+            );
+        }
+        Err(error) => {
+            info!(
+                target: TARGET,
+                "Invalid transaction `{transaction_alias}` ({:?}) was rejected by `{node_name}`: {error}",
+                tx_hash
+            );
+        }
+    }
+
+    Ok(())
+}
+
+fn wallet_transaction_error(error: &WalletTransactionError) -> StepError {
+    StepError::LogicalError {
+        message: error.to_string(),
+    }
+}
+
+fn receiver_public_key(
+    world: &CucumberWorld,
+    step: &str,
+    receiver_wallet_name: &str,
+) -> Result<ZkPublicKey, StepError> {
+    let receiver = world
+        .resolve_wallet(receiver_wallet_name)
+        .inspect_err(|e| {
+            warn!(target: TARGET, "Step `{step}` error: {e}");
+        })?;
+
+    receiver.public_key().inspect_err(|e| {
+        warn!(target: TARGET, "Step `{step}` error: {e}");
+    })
+}
+
+fn transfer_intent(
+    receiver_pk: ZkPublicKey,
+    amount: u64,
+) -> Result<WalletTransactionIntent, StepError> {
+    WalletTransactionIntent::transfer(&[(receiver_pk, amount)], DEFAULT_STORAGE_GAS_PRICE)
+        .map_err(|error| wallet_transaction_error(&error))
+}
+
+fn record_prepared_transaction(
+    world: &mut CucumberWorld,
+    transaction_alias: String,
+    signed: &SignedUserWalletSubmission,
+) -> TxHash {
+    let tx_hash = signed.tx_hash();
+
+    record_signed_user_wallet_submission(world, signed);
+    world.remember_submitted_transaction(transaction_alias.clone(), tx_hash);
+    world.remember_prepared_transaction(transaction_alias, signed.signed_tx().clone());
+
+    tx_hash
+}
+
+async fn submit_prepared_transaction_to_node(
+    world: &CucumberWorld,
+    step: &str,
+    transaction_alias: &str,
+    signed_tx: &SignedMantleTx,
+    tx_hash: TxHash,
+    node_name: &str,
+) -> Result<(), StepError> {
+    let node = world.resolve_node_http_client(node_name).inspect_err(|e| {
+        warn!(target: TARGET, "Step `{step}` error: {e}");
+    })?;
+
+    node.submit_transaction(signed_tx).await.inspect_err(|e| {
+        warn!(target: TARGET, "Step `{step}` error: {e}");
+    })?;
+
+    info!(
+        target: TARGET,
+        "Submitted prepared transaction `{transaction_alias}` ({:?}) to `{node_name}`",
+        tx_hash
+    );
+
+    Ok(())
+}
+
+fn report_prepared_transaction(
+    transaction_alias: &str,
+    tx_hash: TxHash,
+    sender_wallet_name: &str,
+    receiver_wallet_name: &str,
+) {
+    info!(
+        target: TARGET,
+        "Prepared transfer transaction `{transaction_alias}` ({:?}) from `{sender_wallet_name}` to `{receiver_wallet_name}`",
+        tx_hash
+    );
+}

tests/src/cucumber/steps/manual_mempool/assertions.rs

@@ -0,0 +1,228 @@
+use std::time::Duration;
+
+use lb_core::mantle::TxHash;
+use tokio::time::{sleep, timeout};
+
+use crate::{
+    cucumber::{error::StepError, world::CucumberWorld},
+    non_zero,
+};
+
+pub async fn assert_transaction_not_pending_on_all_nodes(
+    world: &CucumberWorld,
+    transaction_alias: String,
+    timeout_seconds: u64,
+) -> Result<(), StepError> {
+    let tx_hash = world.resolve_submitted_transaction(&transaction_alias)?;
+    let node_names = all_node_names(world);
+
+    let wait_result = timeout(Duration::from_secs(timeout_seconds), async {
+        loop {
+            let views = collect_mempool_views(world, &node_names).await?;
+
+            if views.all_absent(tx_hash) {
+                return Ok(());
+            }
+
+            sleep(Duration::from_millis(500)).await;
+        }
+    })
+    .await;
+
+    if let Ok(result) = wait_result {
+        return result;
+    }
+
+    let observed = describe_mempool_views(world, &node_names).await;
+
+    Err(StepError::Timeout {
+        message: format!(
+            "Timed out waiting for transaction `{transaction_alias}` ({tx_hash:?}) to be absent from every node mempool. Observed views: {observed}"
+        ),
+    })
+}
+
+pub async fn assert_transaction_pending_on_nodes(
+    world: &CucumberWorld,
+    transaction_alias: String,
+    node_names: Vec<String>,
+    timeout_seconds: u64,
+) -> Result<(), StepError> {
+    let tx_hash = world.resolve_submitted_transaction(&transaction_alias)?;
+
+    let wait_result = timeout(Duration::from_secs(timeout_seconds), async {
+        loop {
+            let views = collect_mempool_views(world, &node_names).await?;
+
+            if views.all_contain(tx_hash) {
+                return Ok(());
+            }
+
+            sleep(Duration::from_millis(500)).await;
+        }
+    })
+    .await;
+
+    if let Ok(result) = wait_result {
+        return result;
+    }
+
+    let observed = describe_mempool_views(world, &node_names).await;
+
+    Err(StepError::Timeout {
+        message: format!(
+            "Timed out waiting for transaction `{transaction_alias}` ({tx_hash:?}) to be pending in mempools of nodes: {}. Observed views: {observed}",
+            node_names.join(", ")
+        ),
+    })
+}
+
+pub async fn assert_transaction_remains_not_pending_on_all_nodes(
+    world: &CucumberWorld,
+    transaction_alias: String,
+    blocks: u64,
+    timeout_seconds: u64,
+) -> Result<(), StepError> {
+    let tx_hash = world.resolve_submitted_transaction(&transaction_alias)?;
+    let required_blocks = non_zero!("blocks", blocks)?.get();
+    let node_names = all_node_names(world);
+
+    let wait_result = timeout(Duration::from_secs(timeout_seconds), async {
+        let start_height = loop {
+            let views = collect_mempool_views(world, &node_names).await?;
+
+            if views.all_absent(tx_hash) {
+                break minimum_node_height(world).await?;
+            }
+
+            sleep(Duration::from_millis(500)).await;
+        };
+        let target_height = start_height.saturating_add(required_blocks);
+
+        loop {
+            let views = collect_mempool_views(world, &node_names).await?;
+
+            if !views.all_absent(tx_hash) {
+                return Err(StepError::StepFail {
+                    message: format!(
+                        "Transaction `{transaction_alias}` became pending again in at least one node mempool. Observed views: {}",
+                        views.describe()
+                    ),
+                });
+            }
+
+            if minimum_node_height(world).await? >= target_height {
+                return Ok(());
+            }
+
+            sleep(Duration::from_millis(500)).await;
+        }
+    })
+    .await;
+
+    if let Ok(result) = wait_result {
+        return result;
+    }
+
+    let observed = describe_mempool_views(world, &node_names).await;
+
+    Err(StepError::Timeout {
+        message: format!(
+            "Timed out waiting for transaction `{transaction_alias}` to remain absent from every node mempool for {required_blocks} blocks. Observed views: {observed}"
+        ),
+    })
+}
+
+async fn collect_mempool_views(
+    world: &CucumberWorld,
+    node_names: &[String],
+) -> Result<MempoolViews, StepError> {
+    let mut nodes = Vec::with_capacity(node_names.len());
+
+    for node_name in node_names {
+        let client = world.resolve_node_http_client(node_name)?;
+        let pending = client.test_mempool_view().await?;
+
+        nodes.push(NodeMempoolView {
+            node_name: node_name.clone(),
+            pending,
+        });
+    }
+
+    Ok(MempoolViews { nodes })
+}
+
+async fn describe_mempool_views(world: &CucumberWorld, node_names: &[String]) -> String {
+    let mut views = Vec::new();
+
+    for node_name in node_names {
+        let view = match world.resolve_node_http_client(node_name) {
+            Ok(client) => match client.test_mempool_view().await {
+                Ok(items) => format!("{items:?}"),
+                Err(error) => format!("error: {error}"),
+            },
+            Err(error) => format!("error: {error}"),
+        };
+
+        views.push(format!("{node_name}={view}"));
+    }
+
+    views.join("; ")
+}
+
+fn all_node_names(world: &CucumberWorld) -> Vec<String> {
+    let mut node_names = world.all_node_names();
+    node_names.sort();
+    node_names
+}
+
+async fn minimum_node_height(world: &CucumberWorld) -> Result<u64, StepError> {
+    let mut min_height = None;
+
+    for node_name in world.all_node_names() {
+        let client = world.resolve_node_http_client(&node_name)?;
+        let height = client.consensus_info().await?.cryptarchia_info.height;
+        min_height = Some(min_height.map_or(height, |current: u64| current.min(height)));
+    }
+
+    min_height.ok_or_else(|| StepError::StepFail {
+        message: "No nodes are available to inspect mempool status".to_owned(),
+    })
+}
+
+struct MempoolViews {
+    nodes: Vec<NodeMempoolView>,
+}
+
+impl MempoolViews {
+    fn all_absent(&self, tx_hash: TxHash) -> bool {
+        self.nodes.iter().all(|node| !node.contains(tx_hash))
+    }
+
+    fn all_contain(&self, tx_hash: TxHash) -> bool {
+        self.nodes.iter().all(|node| node.contains(tx_hash))
+    }
+
+    fn describe(&self) -> String {
+        self.nodes
+            .iter()
+            .map(NodeMempoolView::describe)
+            .collect::<Vec<_>>()
+            .join("; ")
+    }
+}
+
+struct NodeMempoolView {
+    node_name: String,
+    pending: Vec<TxHash>,
+}
+
+impl NodeMempoolView {
+    fn contains(&self, tx_hash: TxHash) -> bool {
+        self.pending.contains(&tx_hash)
+    }
+
+    fn describe(&self) -> String {
+        format!("{}={:?}", self.node_name, self.pending)
+    }
+}

tests/src/cucumber/steps/manual_mempool/mod.rs

@@ -0,0 +1,3 @@
+pub mod actions;
+pub mod assertions;
+pub mod steps;

tests/src/cucumber/steps/manual_mempool/steps.rs

@@ -0,0 +1,145 @@
+use cucumber::{gherkin::Step, then, when};
+
+use crate::cucumber::{
+    error::{StepError, StepResult},
+    steps::manual_mempool::{
+        actions::{
+            prepare_transfer_transaction, submit_prepared_transaction_to_nodes,
+            try_submit_invalid_transaction,
+        },
+        assertions::{
+            assert_transaction_not_pending_on_all_nodes, assert_transaction_pending_on_nodes,
+            assert_transaction_remains_not_pending_on_all_nodes,
+        },
+    },
+    world::CucumberWorld,
+};
+
+#[when(
+    expr = "I prepare transfer transaction {string} of {int} LGO from wallet {string} to wallet {string}"
+)]
+async fn step_prepare_transfer_transaction(
+    world: &mut CucumberWorld,
+    step: &Step,
+    transaction_alias: String,
+    amount: u64,
+    sender_wallet_name: String,
+    receiver_wallet_name: String,
+) -> StepResult {
+    prepare_transfer_transaction(
+        world,
+        &step.value,
+        transaction_alias,
+        amount,
+        sender_wallet_name,
+        receiver_wallet_name,
+    )
+    .await
+}
+
+#[when(expr = "I submit prepared transaction {string} to nodes:")]
+#[expect(
+    clippy::needless_pass_by_ref_mut,
+    reason = "Cucumber step functions require `&mut World` as the first parameter"
+)]
+async fn step_submit_prepared_transaction_to_nodes(
+    world: &mut CucumberWorld,
+    step: &Step,
+    transaction_alias: String,
+) -> StepResult {
+    let node_names = parse_node_names_table(step)?;
+
+    submit_prepared_transaction_to_nodes(world, &step.value, transaction_alias, node_names).await
+}
+
+#[when(expr = "I try to submit invalid transaction {string} to node {string}")]
+async fn step_try_submit_invalid_transaction(
+    world: &mut CucumberWorld,
+    step: &Step,
+    transaction_alias: String,
+    node_name: String,
+) -> StepResult {
+    try_submit_invalid_transaction(world, &step.value, transaction_alias, node_name).await
+}
+
+#[then(expr = "transaction {string} is pending in mempool of nodes in {int} seconds:")]
+#[expect(
+    clippy::needless_pass_by_ref_mut,
+    reason = "Cucumber step functions require `&mut World` as the first parameter"
+)]
+async fn step_transaction_pending_in_node_mempools(
+    world: &mut CucumberWorld,
+    step: &Step,
+    transaction_alias: String,
+    timeout_seconds: u64,
+) -> StepResult {
+    let node_names = parse_node_names_table(step)?;
+
+    assert_transaction_pending_on_nodes(world, transaction_alias, node_names, timeout_seconds).await
+}
+
+#[then(expr = "transaction {string} is not pending in mempool of all nodes in {int} seconds")]
+#[expect(
+    clippy::needless_pass_by_ref_mut,
+    reason = "Cucumber step functions require `&mut World` as the first parameter"
+)]
+async fn step_transaction_not_pending_in_all_mempools(
+    world: &mut CucumberWorld,
+    step: &Step,
+    transaction_alias: String,
+    timeout_seconds: u64,
+) -> StepResult {
+    let _ = step;
+
+    assert_transaction_not_pending_on_all_nodes(world, transaction_alias, timeout_seconds).await
+}
+
+#[then(
+    expr = "transaction {string} remains not pending in mempool of all nodes for {int} blocks in {int} seconds"
+)]
+#[expect(
+    clippy::needless_pass_by_ref_mut,
+    reason = "Cucumber step functions require `&mut World` as the first parameter"
+)]
+async fn step_transaction_remains_not_pending_in_all_mempools(
+    world: &mut CucumberWorld,
+    step: &Step,
+    transaction_alias: String,
+    blocks: u64,
+    timeout_seconds: u64,
+) -> StepResult {
+    let _ = step;
+
+    assert_transaction_remains_not_pending_on_all_nodes(
+        world,
+        transaction_alias,
+        blocks,
+        timeout_seconds,
+    )
+    .await
+}
+
+fn parse_node_names_table(step: &Step) -> Result<Vec<String>, StepError> {
+    let table = step.table.as_ref().ok_or(StepError::MissingTable)?;
+
+    if table.rows.is_empty() || table.rows[0].len() != 1 || table.rows[0][0].trim() != "node_name" {
+        return Err(StepError::InvalidArgument {
+            message: "Expected table columns: | node_name |".to_owned(),
+        });
+    }
+
+    table
+        .rows
+        .iter()
+        .skip(1)
+        .map(|row| {
+            if row.len() != 1 {
+                return Err(StepError::InvalidArgument {
+                    message: "Each node row must have exactly one column".to_owned(),
+                });
+            }
+
+            Ok(row[0].trim().to_owned())
+        })
+        .collect()
+}

tests/src/cucumber/steps/manual_transactions/mod.rs

@@ -3,5 +3,5 @@ pub mod command_file_utils;
 mod faucet;
 pub mod inscriptions;
 pub mod steps;
-mod tracked_transactions;
+pub(crate) mod tracked_transactions;
 pub mod utils;

tests/src/cucumber/steps/manual_transactions/tracked_transactions.rs

@@ -179,7 +179,7 @@ async fn transaction_is_in_chain(
     .is_some()
 }
 
-fn create_invalid_transaction() -> SignedMantleTx {
+pub fn create_invalid_transaction() -> SignedMantleTx {
     let output_note = Note::new(1000, ZkPublicKey::new(1u8.into()));
     let transfer_op = TransferOp::new(Inputs::empty(), Outputs::new(vec![output_note]));
 

tests/src/cucumber/steps/mod.rs

@@ -4,6 +4,7 @@ pub mod workloads;
 
 pub mod manual_cluster;
 pub mod manual_k8s;
+pub mod manual_mempool;
 pub mod manual_nodes;
 pub mod manual_transactions;
 pub mod manual_zone;

tests/src/cucumber/wallet/submissions.rs

@@ -8,7 +8,7 @@ use std::time::Duration;
 use hex::ToHex as _;
 use lb_core::{
     codec::SerializeOp as _,
-    mantle::{OpProof, TxHash, Utxo},
+    mantle::{OpProof, SignedMantleTx, TxHash, Utxo},
 };
 use lb_http_api_common::bodies::wallet::transfer_funds::WalletTransferFundsRequestBody;
 use lb_key_management_system_service::keys::ZkPublicKey;
@@ -41,12 +41,27 @@ pub struct PreparedUserWalletSubmission {
     submission: PreparedWalletTransaction,
 }
 
+pub(crate) struct SignedUserWalletSubmission {
+    wallet: WalletInfo,
+    submission: SignedWalletTransaction,
+}
+
 impl PreparedUserWalletSubmission {
     pub(crate) const fn tx_hash(&self) -> TxHash {
         self.submission.tx_hash()
     }
 }
 
+impl SignedUserWalletSubmission {
+    pub(crate) const fn tx_hash(&self) -> TxHash {
+        self.submission.tx_hash()
+    }
+
+    pub(crate) const fn signed_tx(&self) -> &SignedMantleTx {
+        self.submission.signed_tx()
+    }
+}
+
 pub async fn create_and_submit_transaction(
     world: &mut CucumberWorld,
     step: &str,
@@ -186,6 +201,36 @@ pub async fn submit_prepared_user_wallet_transaction(
     Ok(tx_hash)
 }
 
+pub(crate) fn sign_prepared_user_wallet_transaction(
+    step: &str,
+    prepared: PreparedUserWalletSubmission,
+    extra_op_proofs: Vec<OpProof>,
+) -> Result<SignedUserWalletSubmission, StepError> {
+    let PreparedUserWalletSubmission { wallet, submission } = prepared;
+    let signed_submission = submission
+        .sign_with_leading_proofs(extra_op_proofs)
+        .map_err(wallet_transaction_error)
+        .inspect_err(|e| {
+            warn!(target: TARGET, "Step `{}` error: {e}", step);
+        })?;
+
+    Ok(SignedUserWalletSubmission {
+        wallet,
+        submission: signed_submission,
+    })
+}
+
+pub(crate) fn record_signed_user_wallet_submission(
+    world: &mut CucumberWorld,
+    signed_submission: &SignedUserWalletSubmission,
+) {
+    record_wallet_submission(
+        world,
+        &signed_submission.wallet,
+        &signed_submission.submission,
+    );
+}
+
 pub(crate) async fn prepare_user_wallet_transaction_submission(
     world: &mut CucumberWorld,
     step: &str,

tests/src/cucumber/world.rs

@@ -690,6 +690,8 @@ pub struct CucumberWorld {
     pub wallets: TrackedWallets,
     /// Manual: Mapping of scenario transaction aliases to submitted hashes.
     pub submitted_transactions: HashMap<String, TxHash>,
+    /// Manual: Exact signed transactions prepared for later submission.
+    pub prepared_transactions: HashMap<String, SignedMantleTx>,
     /// Manual: Mapping of logical node names to their corresponding libp2p peer
     /// IDs.
     pub node_peer_ids: HashMap<String, PeerId>,
@@ -835,6 +837,7 @@ impl Debug for CucumberWorld {
             .field("wallet_accounts", &self.wallet_accounts.len())
             .field("scenario_fee_state", &fee_state_summary(&self.fee_state))
             .field("submitted_transactions", &self.submitted_transactions.len())
+            .field("prepared_transactions", &self.prepared_transactions.len())
             .field(
                 "wallet_utxos_by_block",
                 &wallet_diagnostics.utxo_snapshot_count,
@@ -1403,6 +1406,19 @@ impl CucumberWorld {
             })
     }
 
+    pub fn remember_prepared_transaction(&mut self, alias: String, signed_tx: SignedMantleTx) {
+        self.prepared_transactions.insert(alias, signed_tx);
+    }
+
+    pub fn resolve_prepared_transaction(&self, alias: &str) -> Result<SignedMantleTx, StepError> {
+        self.prepared_transactions
+            .get(alias)
+            .cloned()
+            .ok_or(StepError::LogicalError {
+                message: format!("Prepared transaction alias '{alias}' not found in world state"),
+            })
+    }
+
     /// Helper to resolve multiple wallet names to their actual wallet
     /// information.
     pub fn resolve_wallets(&self, wallet_names: &[String]) -> Result<Vec<WalletInfo>, StepError> {

tests/testing_framework/src/node/http_client.rs

@@ -6,12 +6,19 @@ use common_http_client::{
 use futures::Stream;
 use lb_blend_service::message::NetworkInfo as BlendNetworkInfo;
 use lb_chain_service::ChainServiceInfo;
-use lb_core::{header::HeaderId, mantle::SignedMantleTx, sdp::Declaration};
+use lb_core::{
+    header::HeaderId,
+    mantle::{SignedMantleTx, TxHash},
+    sdp::Declaration,
+};
 use lb_http_api_common::{
     bodies::wallet::transfer_funds::{
         WalletTransferFundsRequestBody, WalletTransferFundsResponseBody,
     },
-    paths::{BLEND_NETWORK_INFO, DIAL_PEER, MANTLE_METRICS, MANTLE_SDP_DECLARATIONS, NETWORK_INFO},
+    paths::{
+        BLEND_NETWORK_INFO, DIAL_PEER, MANTLE_METRICS, MANTLE_SDP_DECLARATIONS, NETWORK_INFO,
+        TEST_MEMPOOL_VIEW,
+    },
     queries::BlocksStreamQuery,
 };
 use lb_libp2p::{Multiaddr, PeerId};
@@ -148,6 +155,18 @@ impl NodeHttpClient {
         self.get_sdp_declarations_at(self.base_url.clone()).await
     }
 
+    pub async fn test_mempool_view(&self) -> Result<Vec<TxHash>, Error> {
+        let testing_url = self
+            .testing_url
+            .clone()
+            .ok_or_else(|| Error::Client("testing api unavailable".to_owned()))?;
+        let request_url = Self::join_path(&testing_url, TEST_MEMPOOL_VIEW)?;
+
+        self.http_client
+            .get::<(), Vec<TxHash>>(request_url, None)
+            .await
+    }
+
     pub async fn dial_peer(&self, addr: Multiaddr) -> Result<PeerId, Error> {
         let testing_url = self
             .testing_url