chore(api): Add testing connect endpoint (#2574)

Cargo.lock

@@ -4211,6 +4211,7 @@ dependencies = [
  "logos-blockchain-chain-service",
  "logos-blockchain-core",
  "logos-blockchain-ledger",
+ "logos-blockchain-libp2p",
  "logos-blockchain-network-service",
  "logos-blockchain-sdp-service",
  "logos-blockchain-storage-service",

nodes/api-common/src/paths.rs

@@ -29,3 +29,4 @@ pub mod wallet {
 
 // testing paths
 pub const UPDATE_MEMBERSHIP: &str = "/test/membership/update";
+pub const DIAL_PEER: &str = "/test/network/dial_peer";

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

@@ -6,11 +6,11 @@ use axum::{
         HeaderValue,
         header::{CONTENT_TYPE, USER_AGENT},
     },
-    routing::get,
+    routing::{get, post},
 };
 use lb_api_service::Backend;
-use lb_http_api_common::paths::MANTLE_SDP_DECLARATIONS;
-pub use lb_network_service::backends::libp2p::Libp2p as NetworkBackend;
+use lb_http_api_common::paths::{DIAL_PEER, MANTLE_SDP_DECLARATIONS};
+use lb_network_service::{NetworkService, backends::libp2p::Libp2p as NetworkBackend};
 use overwatch::{overwatch::handle::OverwatchHandle, services::AsServiceId};
 use tokio::net::TcpListener;
 use tower::limit::ConcurrencyLimitLayer;
@@ -23,7 +23,10 @@ use tower_http::{
 use tracing::Level as TracingLevel;
 
 use crate::{
-    api::{backend::AxumBackendSettings, testing::handlers::get_sdp_declarations},
+    api::{
+        backend::AxumBackendSettings,
+        testing::handlers::{dial_peer, get_sdp_declarations},
+    },
     generic_services::{self, SdpService},
 };
 pub struct TestAxumBackend {
@@ -45,6 +48,7 @@ where
         + Debug
         + Clone
         + 'static
+        + AsServiceId<NetworkService<NetworkBackend, RuntimeServiceId>>
         + AsServiceId<TestCryptarchiaService<RuntimeServiceId>>
         + AsServiceId<TestHttpCryptarchiaService<RuntimeServiceId>>
         + AsServiceId<SdpService<RuntimeServiceId>>
@@ -81,6 +85,7 @@ where
                 MANTLE_SDP_DECLARATIONS,
                 get(get_sdp_declarations::<RuntimeServiceId>),
             )
+            .route(DIAL_PEER, post(dial_peer::<RuntimeServiceId>))
             .with_state(handle)
             .layer(axum::extract::DefaultBodyLimit::max(
                 self.settings.max_body_size,

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

@@ -1,8 +1,11 @@
 use std::fmt::{Debug, Display};
 
-use axum::{extract::State, response::Response};
-use lb_api_service::http::mantle;
+use axum::{Json, extract::State, response::Response};
+use lb_api_service::http::{libp2p, mantle};
+use lb_libp2p::{Multiaddr, PeerId};
+use lb_network_service::{NetworkService, backends::libp2p::Libp2p as NetworkBackend};
 use overwatch::{overwatch::OverwatchHandle, services::AsServiceId};
+use serde::{Deserialize, Serialize};
 
 use super::backend::TestHttpCryptarchiaService;
 use crate::{
@@ -10,6 +13,11 @@ use crate::{
     make_request_and_return_response,
 };
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DialPeerRequestBody {
+    pub addr: Multiaddr,
+}
+
 pub async fn get_sdp_declarations<RuntimeServiceId>(
     State(handle): State<OverwatchHandle<RuntimeServiceId>>,
 ) -> Response
@@ -25,3 +33,24 @@ where
 {
     make_request_and_return_response!(mantle::get_sdp_declarations::<RuntimeServiceId>(&handle))
 }
+
+pub async fn dial_peer<RuntimeServiceId>(
+    State(handle): State<OverwatchHandle<RuntimeServiceId>>,
+    Json(req): Json<DialPeerRequestBody>,
+) -> Response
+where
+    RuntimeServiceId: Debug
+        + Send
+        + Sync
+        + Display
+        + 'static
+        + AsServiceId<NetworkService<NetworkBackend, RuntimeServiceId>>
+        + AsServiceId<TestHttpCryptarchiaService<RuntimeServiceId>>
+        + AsServiceId<SdpService<RuntimeServiceId>>
+        + AsServiceId<TxMempoolService<RuntimeServiceId>>,
+{
+    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)
+    })
+}

services/api/Cargo.toml

@@ -30,6 +30,7 @@ lb-chain-leader-service    = { workspace = true }
 lb-chain-service           = { features = ["libp2p"], workspace = true }
 lb-core                    = { workspace = true }
 lb-ledger                  = { workspace = true }
+lb-libp2p                  = { workspace = true }
 lb-network-service         = { workspace = true }
 lb-sdp-service             = { workspace = true }
 lb-storage-service         = { features = ["rocksdb-backend"], workspace = true }

services/api/src/http/libp2p.rs

@@ -1,8 +1,12 @@
 use std::fmt::{Debug, Display};
 
+use lb_libp2p::{Multiaddr, PeerId};
 use lb_network_service::{
     NetworkService,
-    backends::libp2p::{Command, Libp2p, Libp2pInfo, NetworkCommand::Info},
+    backends::libp2p::{
+        Command, Dial, Libp2p, Libp2pInfo,
+        NetworkCommand::{Connect, Info},
+    },
     message::NetworkMsg,
 };
 use overwatch::services::AsServiceId;
@@ -29,3 +33,30 @@ where
         .await
         .map_err(|e| Box::new(e) as overwatch::DynError)
 }
+
+pub async fn connect_peer<RuntimeServiceId>(
+    handle: &overwatch::overwatch::handle::OverwatchHandle<RuntimeServiceId>,
+    addr: Multiaddr,
+) -> Result<PeerId, overwatch::DynError>
+where
+    RuntimeServiceId:
+        AsServiceId<NetworkService<Libp2p, RuntimeServiceId>> + Debug + Sync + Display + 'static,
+{
+    let relay = handle.relay().await?;
+    let (sender, receiver) = oneshot::channel();
+
+    relay
+        .send(NetworkMsg::Process(Command::Network(Connect(Dial {
+            addr,
+            retry_count: 0,
+            result_sender: sender,
+        }))))
+        .await
+        .map_err(|(e, _)| e)?;
+
+    let dial_result = receiver
+        .await
+        .map_err(|e| Box::new(e) as overwatch::DynError)?;
+
+    dial_result.map_err(|e| Box::new(e) as overwatch::DynError)
+}

services/network/src/backends/libp2p/command.rs

@@ -39,6 +39,7 @@ pub struct Dial {
 pub struct Libp2pInfo {
     pub listen_addresses: Vec<Multiaddr>,
     pub peer_id: PeerId,
+    #[serde(default)]
     pub connected_peers: Vec<PeerId>,
     pub n_peers: usize,
     pub n_connections: u32,

tests/cucumber_tests/features/test_harness.feature

@@ -0,0 +1,11 @@
+Feature: Test harness
+
+  @smoke_ci
+  Scenario: Two nodes connect at runtime
+    Given I have a cluster with capacity of 2 nodes
+    And I start node "NODE_1"
+    And I start node "NODE_2"
+    When I connect node "NODE_2" to node "NODE_1" at runtime
+    Then node "NODE_1" has at least 1 peers within 15 seconds
+    And node "NODE_2" has at least 1 peers within 15 seconds
+    And I stop all nodes

tests/src/cucumber/steps/manual_cluster.rs

@@ -1,5 +1,6 @@
 use std::{collections::HashMap, hash::BuildHasher, time::Duration};
 
+use lb_libp2p::{Multiaddr, PeerId, Protocol};
 use lb_testing_framework::{
     DeploymentBuilder, LbcEnv, LbcLocalDeployer, NodeHttpClient, TopologyConfig,
     configs::wallet::WalletAccount, internal::DeploymentPlan,
@@ -286,6 +287,46 @@ pub async fn assert_manual_node_has_peers(
     }
 }
 
+pub async fn connect_manual_node_to_node(
+    world: &CucumberWorld,
+    source_node_name: &str,
+    target_node_name: &str,
+) -> StepResult {
+    let source_client = world.resolve_node_http_client(source_node_name)?;
+    let target_client = world.resolve_node_http_client(target_node_name)?;
+    let target_network = target_client.network_info().await?;
+    let target_addr = compose_dial_addr(
+        target_network.listen_addresses.first(),
+        target_network.peer_id,
+    )
+    .ok_or_else(|| StepError::LogicalError {
+        message: format!("node '{target_node_name}' has no listen address to dial"),
+    })?;
+
+    source_client
+        .dial_peer(target_addr)
+        .await
+        .map(|_| ())
+        .map_err(|error| StepError::LogicalError {
+            message: format!(
+                "failed to connect node '{source_node_name}' to node '{target_node_name}': {error}"
+            ),
+        })
+}
+
+fn compose_dial_addr(listen_addr: Option<&Multiaddr>, peer_id: PeerId) -> Option<Multiaddr> {
+    let mut addr = listen_addr?.clone();
+    let has_peer_id = addr
+        .iter()
+        .any(|protocol| matches!(protocol, Protocol::P2p(_)));
+
+    if !has_peer_id {
+        addr.push(Protocol::P2p(peer_id));
+    }
+
+    Some(addr)
+}
+
 pub fn build_user_wallets(
     world: &CucumberWorld,
     node_name: &str,

tests/src/cucumber/steps/manual_nodes/steps.rs

@@ -11,6 +11,7 @@ use crate::{
         steps::{
             TARGET,
             manual_cluster::{
+                assert_manual_node_has_peers, connect_manual_node_to_node,
                 install_local_manual_cluster, rebuild_pending_local_manual_cluster,
                 stop_active_manual_cluster,
             },
@@ -183,6 +184,33 @@ async fn step_start_manual_stand_alone_node(
     start_node(world, &step.value, &node_name, &Vec::new(), &Vec::new()).await
 }
 
+#[when(expr = "I connect node {string} to node {string} at runtime")]
+#[expect(
+    clippy::needless_pass_by_ref_mut,
+    reason = "Cucumber step entrypoints must take `&mut World`"
+)]
+async fn step_connect_nodes_at_runtime(
+    world: &mut CucumberWorld,
+    source_node_name: String,
+    target_node_name: String,
+) -> StepResult {
+    connect_manual_node_to_node(world, &source_node_name, &target_node_name).await
+}
+
+#[then(expr = "node {string} has at least {int} peers within {int} seconds")]
+#[expect(
+    clippy::needless_pass_by_ref_mut,
+    reason = "Cucumber step entrypoints must take `&mut World`"
+)]
+async fn step_node_has_peers(
+    world: &mut CucumberWorld,
+    node_name: String,
+    min_peers: usize,
+    timeout_secs: u64,
+) -> StepResult {
+    assert_manual_node_has_peers(world, &node_name, min_peers, timeout_secs).await
+}
+
 #[when(expr = "I restart node {string}")]
 #[expect(
     clippy::needless_pass_by_ref_mut,

tests/testing_framework/src/node/http_client.rs

@@ -10,10 +10,12 @@ use lb_http_api_common::{
     bodies::wallet::transfer_funds::{
         WalletTransferFundsRequestBody, WalletTransferFundsResponseBody,
     },
-    paths::{MANTLE_SDP_DECLARATIONS, NETWORK_INFO},
+    paths::{DIAL_PEER, MANTLE_SDP_DECLARATIONS, NETWORK_INFO},
 };
+use lb_libp2p::{Multiaddr, PeerId};
 use lb_network_service::backends::libp2p::Libp2pInfo;
 use reqwest::Url;
+use serde::{Deserialize, Serialize};
 
 #[derive(Clone)]
 pub struct NodeHttpClient {
@@ -114,6 +116,18 @@ impl NodeHttpClient {
         self.get_sdp_declarations_at(self.base_url.clone()).await
     }
 
+    pub async fn dial_peer(&self, addr: Multiaddr) -> Result<PeerId, 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, DIAL_PEER)?;
+
+        self.http_client
+            .post::<_, PeerId>(request_url, &DialPeerRequestBody { addr })
+            .await
+    }
+
     #[must_use]
     pub const fn base_url(&self) -> &Url {
         &self.base_url
@@ -149,3 +163,8 @@ impl NodeHttpClient {
             .map_err(Error::Url)
     }
 }
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+struct DialPeerRequestBody {
+    addr: Multiaddr,
+}