Impl DHT bootnodes support for all nodes (#1344)

Latest polkadot-sdk release includes a new feature that allows parachain nodes to advertise their IP address in the relay chain DHT (kademlia). This allows other nodes to discover peers without a centralized peer list.

In this PR we implement it for all nodes, including collators and data preservers. Container chain collators do not advertise their IP address, but they do discover bootnodes for the chain they are assigned to. The reason for not advertising is that collators may rotate frequently, and also that they don't have the full block history. So collators can only be valid targets for warp sync, not for full sync.

Author: tmpolaczyk

.github/workflows/run-zombienet-tests.yml

@@ -68,7 +68,7 @@ jobs:
 
           if [ -z "$test_name" ]; then
             if [ "$test_bundle" == "weekly_bundle" ]; then
-              tests='["zombie_tanssi_warp_sync", "zombie_tanssi_parathreads", "zombie_tanssi_metrics", "zombie_tanssi_relay_finalization_recovery"]'
+              tests='["zombie_tanssi_warp_sync", "zombie_tanssi_parathreads", "zombie_tanssi_metrics", "zombie_tanssi_relay_finalization_recovery", "zombie_tanssi_relay_dht_bootnodes"]'
             elif [ "$test_bundle" == "manual_bundle" ]; then
               tests='["zombie_tanssi_relay_finalization_recovery", "zombie_simple_template_offchain", "zombie_tanssi_keep_db", "zombie_tanssi_one_node", "zombie_tanssi_relay_starlight", "zombie_tanssi_relay_unneeded_para", "zombie_tanssi_relay_keystore", "zombie_tanssi_relay_foreign_tokens"]'
             else
@@ -201,4 +201,4 @@ jobs:
             --body "Tests ${{ matrix.test_name }} failed in [${{ github.workflow }}](${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}).\n\n@moondance-labs/coredev please investigate." \
             --repo "${{ github.repository }}"
         env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

Cargo.lock

@@ -204,7 +204,7 @@ where
     let (_, import_queue) = import_queue(&parachain_config, &node_builder);
 
     // Relay chain interface
-    let (relay_chain_interface, _collator_key) = node_builder
+    let (relay_chain_interface, _collator_key, start_bootnode_params) = node_builder
         .build_relay_chain_interface(&parachain_config, polkadot_config, collator_options.clone())
         .await?;
 
@@ -295,6 +295,7 @@ where
         para_id,
         relay_chain_interface.clone(),
         relay_chain_slot_duration,
+        start_bootnode_params,
     )?;
 
     Ok((node_builder.task_manager, node_builder.client))

chains/container-chains/nodes/frontier/src/service.rs

@@ -133,7 +133,7 @@ where
     let (_, import_queue) = import_queue(&parachain_config, &node_builder);
 
     // Relay chain interface
-    let (relay_chain_interface, _collator_key) = node_builder
+    let (relay_chain_interface, _collator_key, start_bootnode_params) = node_builder
         .build_relay_chain_interface(&parachain_config, polkadot_config, collator_options.clone())
         .await?;
 
@@ -170,6 +170,7 @@ where
         para_id,
         relay_chain_interface.clone(),
         relay_chain_slot_duration,
+        start_bootnode_params,
     )?;
 
     Ok((node_builder.task_manager, node_builder.client))

chains/container-chains/nodes/simple/src/service.rs

@@ -15,9 +15,11 @@
 // along with Tanssi.  If not, see <http://www.gnu.org/licenses/>.
 
 use {
+    async_channel::Receiver,
     async_io::Timer,
     core::time::Duration,
     core_extensions::TypeIdentity,
+    cumulus_client_bootnodes::{start_bootnode_tasks, StartBootnodeTasksParams},
     cumulus_client_cli::CollatorOptions,
     cumulus_client_consensus_common::ParachainConsensus,
     cumulus_client_service::{
@@ -39,26 +41,32 @@ use {
         sp_wasm_interface::HostFunctions, HeapAllocStrategy, RuntimeVersionOf, WasmExecutor,
         DEFAULT_HEAP_ALLOC_STRATEGY,
     },
-    sc_network::{config::FullNetworkConfiguration, NetworkBlock},
+    sc_network::{
+        config::FullNetworkConfiguration, request_responses::IncomingRequest,
+        service::traits::NetworkService, NetworkBlock,
+    },
     sc_network_sync::SyncingService,
     sc_network_transactions::TransactionsHandlerController,
     sc_service::{
-        Configuration, KeystoreContainer, SpawnTaskHandle, TFullBackend, TFullClient, TaskManager,
+        config::Multiaddr, Configuration, KeystoreContainer, SpawnTaskHandle, TFullBackend,
+        TFullClient, TaskManager,
     },
     sc_telemetry::{Telemetry, TelemetryWorker, TelemetryWorkerHandle},
-    sc_transaction_pool_api::OffchainTransactionPoolFactory,
+    sc_transaction_pool_api::{OffchainTransactionPoolFactory, TransactionPool},
     sc_utils::mpsc::TracingUnboundedSender,
-    sp_api::ConstructRuntimeApi,
+    sp_api::{ConstructRuntimeApi, StorageProof},
     sp_block_builder::BlockBuilder,
     sp_consensus::SelectChain,
-    sp_core::traits::CodeExecutor,
+    sp_core::{
+        traits::{CodeExecutor, SpawnNamed},
+        H256,
+    },
     sp_inherents::CreateInherentDataProviders,
     sp_offchain::OffchainWorkerApi,
     sp_runtime::Percent,
     sp_transaction_pool::runtime_api::TaggedTransactionQueue,
     std::{str::FromStr, sync::Arc},
 };
-use {sc_transaction_pool_api::TransactionPool, sp_api::StorageProof, sp_core::traits::SpawnNamed};
 
 tp_traits::alias!(
     pub trait MinimalRuntimeApi<
@@ -299,6 +307,18 @@ where
     }
 }
 
+#[derive(Clone)]
+pub struct StartBootnodeParams {
+    pub relay_chain_fork_id: Option<String>,
+    pub parachain_fork_id: Option<String>,
+    pub advertise_non_global_ips: bool,
+    pub parachain_public_addresses: Vec<Multiaddr>,
+    pub relay_chain_network: Arc<dyn NetworkService>,
+    pub paranode_rx: Receiver<IncomingRequest>,
+    pub embedded_dht_bootnode: bool,
+    pub dht_bootnode_discovery: bool,
+}
+
 impl<T: NodeBuilderConfig, SNetwork, STxHandler, SImportQueueService>
     NodeBuilder<T, SNetwork, STxHandler, SImportQueueService>
 where
@@ -314,9 +334,20 @@ where
     ) -> sc_service::error::Result<(
         Arc<(dyn RelayChainInterface + 'static)>,
         Option<CollatorPair>,
+        StartBootnodeParams,
     )> {
-        // FIXME(MD-1374): support DHT bootnodes
-        let (relay_chain_interface, collator_key, _relay_chain_network, _paranode_rx) =
+        let relay_chain_fork_id = polkadot_config
+            .chain_spec
+            .fork_id()
+            .map(ToString::to_string);
+        let parachain_fork_id = parachain_config
+            .chain_spec
+            .fork_id()
+            .map(ToString::to_string);
+        let advertise_non_global_ips = parachain_config.network.allow_non_globals_in_dht;
+        let parachain_public_addresses = parachain_config.network.public_addresses.clone();
+
+        let (relay_chain_interface, collator_key, relay_chain_network, paranode_rx) =
             build_relay_chain_interface(
                 polkadot_config,
                 parachain_config,
@@ -328,7 +359,18 @@ where
             .await
             .map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?;
 
-        Ok((relay_chain_interface, collator_key))
+        let start_bootnode_params = StartBootnodeParams {
+            relay_chain_fork_id,
+            parachain_fork_id,
+            advertise_non_global_ips,
+            parachain_public_addresses,
+            relay_chain_network,
+            paranode_rx,
+            embedded_dht_bootnode: collator_options.embedded_dht_bootnode,
+            dht_bootnode_discovery: collator_options.dht_bootnode_discovery,
+        };
+
+        Ok((relay_chain_interface, collator_key, start_bootnode_params))
     }
 
     /// Given an import queue, calls [`cumulus_client_service::build_network`] and
@@ -716,11 +758,14 @@ where
         para_id: ParaId,
         relay_chain_interface: RCInterface,
         relay_chain_slot_duration: Duration,
+        start_bootnode_params: StartBootnodeParams,
     ) -> sc_service::error::Result<NodeBuilder<T, SNetwork, STxHandler, ()>>
     where
         SNetwork: TypeIdentity<Type = Network<BlockOf<T>>>,
         SImportQueueService: TypeIdentity<Type = ImportQueueServiceOf<T>>,
         RCInterface: RelayChainInterface + Clone + 'static,
+        RCInterface: TypeIdentity<Type = Arc<dyn RelayChainInterface + 'static>>,
+        BlockHashOf<T>: TypeIdentity<Type = H256>,
     {
         let NodeBuilder {
             client,
@@ -754,7 +799,7 @@ where
             announce_block,
             task_manager: &mut task_manager,
             para_id,
-            relay_chain_interface,
+            relay_chain_interface: relay_chain_interface.clone(),
             relay_chain_slot_duration,
             import_queue: import_queue_service,
             recovery_handle: Box::new(overseer_handle),
@@ -763,9 +808,39 @@ where
         };
 
         // TODO: change for async backing
+        // TODO: to fix deprecation warning, we only need to change
+        // `start_full_node` to `start_relay_chain_tasks`
         #[allow(deprecated)]
         cumulus_client_service::start_full_node(params)?;
 
+        let StartBootnodeParams {
+            relay_chain_fork_id,
+            parachain_fork_id,
+            advertise_non_global_ips,
+            parachain_public_addresses,
+            relay_chain_network,
+            paranode_rx,
+            embedded_dht_bootnode,
+            dht_bootnode_discovery,
+        } = start_bootnode_params;
+
+        // Advertise parachain bootnode address in relay chain DHT
+        start_bootnode_tasks(StartBootnodeTasksParams {
+            embedded_dht_bootnode,
+            dht_bootnode_discovery,
+            para_id,
+            task_manager: &mut task_manager,
+            relay_chain_interface: TypeIdentity::into_type(relay_chain_interface),
+            relay_chain_fork_id,
+            relay_chain_network,
+            request_receiver: paranode_rx,
+            parachain_network: network.network.clone(),
+            advertise_non_global_ips,
+            parachain_genesis_hash: TypeIdentity::into_type(client.chain_info().genesis_hash),
+            parachain_fork_id,
+            parachain_public_addresses,
+        });
+
         Ok(NodeBuilder {
             client,
             backend,

client/node-common/src/service/node_builder.rs

@@ -19,7 +19,6 @@ pub mod watch_assignment;
 
 use {
     cumulus_client_cli::CollatorOptions,
-    cumulus_client_service::build_relay_chain_interface,
     cumulus_primitives_core::ParaId,
     cumulus_relay_chain_interface::RelayChainInterface,
     dc_orchestrator_chain_interface::OrchestratorChainInterface,
@@ -33,7 +32,9 @@ use {
         service::MinimalContainerRuntimeApi,
         spawner::{ContainerChainSpawnParams, ContainerChainSpawner},
     },
-    tc_service_orchestrator_chain::solochain::EnableContainerChainSpawner,
+    tc_service_orchestrator_chain::solochain::{
+        build_relay_chain_interface_solochain, EnableContainerChainSpawner,
+    },
     url::Url,
 };
 
@@ -131,6 +132,7 @@ where
         let orchestrator_chain_interface: Arc<dyn OrchestratorChainInterface>;
         let relay_chain_interface: Arc<dyn RelayChainInterface>;
         let keystore;
+        let start_bootnode_params;
 
         if self.orchestrator_endpoints.is_empty() {
             // Embeded node
@@ -197,6 +199,7 @@ where
                 relay_chain_interface = started.relay_chain_interface;
                 orchestrator_chain_interface = started.orchestrator_chain_interface;
                 keystore = started.keystore;
+                start_bootnode_params = started.start_bootnode_params;
             } else {
                 log::info!("Starting embeded orchestrator solochain node ...");
 
@@ -222,6 +225,7 @@ where
                 relay_chain_interface = started.relay_chain_interface;
                 orchestrator_chain_interface = started.orchestrator_chain_interface;
                 keystore = started.keystore;
+                start_bootnode_params = started.start_bootnode_params;
             }
         } else {
             log::info!("Connecting to remote orchestrator node(s) ...");
@@ -266,20 +270,22 @@ where
 
             let telemetry_worker_handle = telemetry.as_ref().map(|(worker, _)| worker.handle());
 
-            relay_chain_interface = build_relay_chain_interface(
-                polkadot_config,
+            // This is not solochain mode but this function is just to return the start_bootnode_params
+            let relay_chain_interface_parts = build_relay_chain_interface_solochain(
                 &self.config,
+                polkadot_config,
+                collator_options,
                 telemetry_worker_handle,
                 &mut task_manager,
-                collator_options,
                 None,
             )
             .await
-            .map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?
-            .0;
+            .map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?;
+            relay_chain_interface = relay_chain_interface_parts.0;
 
             let keystore_container = KeystoreContainer::new(&self.config.keystore)?;
             keystore = keystore_container.keystore();
+            start_bootnode_params = relay_chain_interface_parts.2;
         }
 
         log::info!("Starting container chain spawner and assignment watcher ...");
@@ -300,9 +306,10 @@ where
                 sync_keystore: keystore,
                 collation_params: None,
                 spawn_handle: task_manager.spawn_handle().clone(),
-                data_preserver: true,
                 generate_rpc_builder: self.generate_rpc_builder,
                 override_sync_mode: None,
+                start_bootnode_params,
+                data_preserver: true,
                 phantom: PhantomData,
             },
             state: Default::default(),

client/service-container-chain-data-preserver/src/lib.rs

@@ -72,6 +72,7 @@ sp-timestamp = { workspace = true, features = [ "std" ] }
 polkadot-primitives = { workspace = true }
 
 # Cumulus
+cumulus-client-bootnodes = { workspace = true }
 cumulus-client-cli = { workspace = true }
 cumulus-client-collator = { workspace = true }
 cumulus-client-consensus-aura = { workspace = true }