Merge pull request #967 from subspace/fix-block-producing-node

nazar-pc · web-flow · commit 98e87df9e468 · 2022-11-30T05:47:14.000+02:00
Publish pieces on DSN in the background
diff --git a/crates/sc-consensus-subspace/src/archiver.rs b/crates/sc-consensus-subspace/src/archiver.rs
@@ -380,7 +380,6 @@ pub fn start_subspace_archiver<Block, Backend, Client>(
     client: Arc<Client>,
     telemetry: Option<TelemetryHandle>,
     spawner: &impl sp_core::traits::SpawnEssentialNamed,
-    is_authoring_blocks: bool,
 ) where
     Block: BlockT,
     Backend: BackendT<Block>,
@@ -423,16 +422,12 @@ pub fn start_subspace_archiver<Block, Backend, Client>(
 
             async move {
                 // Farmers may have not received all previous segments, send them now.
-                if is_authoring_blocks {
-                    for archived_segment in older_archived_segments {
-                        send_archived_segment_notification(
-                            &archived_segment_notification_sender,
-                            archived_segment,
-                        )
-                        .await;
-                    }
-                } else {
-                    drop(older_archived_segments);
+                for archived_segment in older_archived_segments {
+                    send_archived_segment_notification(
+                        &archived_segment_notification_sender,
+                        archived_segment,
+                    )
+                    .await;
                 }
 
                 while let Some(ImportedBlockNotification {
@@ -509,13 +504,11 @@ pub fn start_subspace_archiver<Block, Backend, Client>(
                     {
                         let root_block = archived_segment.root_block;
 
-                        if is_authoring_blocks {
-                            send_archived_segment_notification(
-                                &archived_segment_notification_sender,
-                                archived_segment,
-                            )
-                            .await;
-                        }
+                        send_archived_segment_notification(
+                            &archived_segment_notification_sender,
+                            archived_segment,
+                        )
+                        .await;
 
                         let _ = root_block_sender.send(root_block).await;
                     }
diff --git a/crates/sc-consensus-subspace/src/tests.rs b/crates/sc-consensus-subspace/src/tests.rs
@@ -500,7 +500,6 @@ fn run_one_test(mutator: impl Fn(&mut TestHeader, Stage) + Send + Sync + 'static
             client.clone(),
             None,
             &task_manager.spawn_essential_handle(),
-            false,
         );
 
         let (archived_pieces_sender, archived_pieces_receiver) = oneshot::channel();
diff --git a/crates/subspace-node/src/bin/subspace-node.rs b/crates/subspace-node/src/bin/subspace-node.rs
@@ -205,7 +205,6 @@ fn main() -> Result<(), Error> {
                     client.clone(),
                     None,
                     &task_manager.spawn_essential_handle(),
-                    config.role.is_authority(),
                 );
 
                 Ok((
diff --git a/crates/subspace-service/src/dsn.rs b/crates/subspace-service/src/dsn.rs
@@ -1,17 +1,18 @@
 mod piece_record_store;
 
 use crate::dsn::piece_record_store::{AuxRecordStorage, SegmentIndexGetter};
-use futures::StreamExt;
-use sc_consensus_subspace::{ArchivedSegmentNotification, SubspaceLink};
+use futures::{Stream, StreamExt};
+use sc_client_api::AuxStore;
+use sc_consensus_subspace::ArchivedSegmentNotification;
 use sc_piece_cache::AuxPieceCache;
-use sp_core::traits::SpawnEssentialNamed;
+use sp_core::traits::SpawnNamed;
 use sp_runtime::traits::Block as BlockT;
 use std::sync::Arc;
 use subspace_core_primitives::{Piece, PieceIndex, PieceIndexHash, PIECES_IN_SEGMENT};
 use subspace_networking::libp2p::{identity, Multiaddr};
 use subspace_networking::{
     BootstrappedNetworkingParameters, CreationError, CustomRecordStore, MemoryProviderStorage,
-    Node, PieceByHashRequestHandler, PieceByHashResponse, PieceKey, ToMultihash,
+    Node, NodeRunner, PieceByHashRequestHandler, PieceByHashResponse, PieceKey, ToMultihash,
 };
 use tracing::{debug, info, trace, Instrument};
 
@@ -33,31 +34,28 @@ pub struct DsnConfig {
     pub allow_non_global_addresses_in_dht: bool,
 }
 
-/// Start an archiver that will listen for archived segments and send it to DSN network using
-/// pub-sub protocol.
-pub async fn start_dsn_node<Block, Spawner, AS: sc_client_api::AuxStore + Sync + Send + 'static>(
-    subspace_link: &SubspaceLink<Block>,
+pub(crate) async fn create_dsn_instance<Block, AS>(
     dsn_config: DsnConfig,
-    spawner: Spawner,
     piece_cache: AuxPieceCache<AS>,
     piece_getter: PieceGetter,
     segment_index_getter: SegmentIndexGetter,
-) -> Result<Node, CreationError>
+) -> Result<
+    (
+        Node,
+        NodeRunner<CustomRecordStore<AuxRecordStorage<AS>, MemoryProviderStorage>>,
+    ),
+    CreationError,
+>
 where
     Block: BlockT,
-    Spawner: SpawnEssentialNamed,
+    AS: AuxStore + Sync + Send + 'static,
 {
-    let span = tracing::info_span!(sc_tracing::logging::PREFIX_LOG_SPAN, name = "DSN");
-    let _enter = span.enter();
-
     // TODO: Combine `AuxPieceCache` with `AuxRecordStorage` and remove `PieceCache` abstraction
     let record_storage = AuxRecordStorage::new(piece_cache, segment_index_getter);
 
     trace!("Subspace networking starting.");
 
-    let networking_config = subspace_networking::Config::<
-        CustomRecordStore<AuxRecordStorage<AS>, MemoryProviderStorage>,
-    > {
+    let networking_config = subspace_networking::Config {
         keypair: dsn_config.keypair,
         listen_on: dsn_config.listen_on,
         allow_non_global_addresses_in_dht: dsn_config.allow_non_global_addresses_in_dht,
@@ -80,58 +78,51 @@ where
         ..subspace_networking::Config::with_generated_keypair()
     };
 
-    let (node, mut node_runner) = subspace_networking::create(networking_config).await?;
-
-    info!("Subspace networking initialized: Node ID is {}", node.id());
+    subspace_networking::create(networking_config).await
+}
 
-    spawner.spawn_essential(
-        "node-runner",
-        Some("subspace-networking"),
-        Box::pin(
-            async move {
-                node_runner.run().await;
+/// Start an archiver that will listen for archived segments and send it to DSN network using
+/// pub-sub protocol.
+pub(crate) async fn start_dsn_archiver<Spawner>(
+    mut archived_segment_notification_stream: impl Stream<Item = ArchivedSegmentNotification> + Unpin,
+    node: Node,
+    spawner: Spawner,
+) where
+    Spawner: SpawnNamed,
+{
+    trace!("Subspace DSN archiver started.");
+
+    let mut last_published_segment_index: Option<u64> = None;
+    while let Some(ArchivedSegmentNotification {
+        archived_segment, ..
+    }) = archived_segment_notification_stream.next().await
+    {
+        let segment_index = archived_segment.root_block.segment_index();
+        let first_piece_index = segment_index * u64::from(PIECES_IN_SEGMENT);
+
+        info!(%segment_index, "Processing a segment.");
+
+        // skip repeating publication
+        if let Some(last_published_segment_index) = last_published_segment_index {
+            if last_published_segment_index == segment_index {
+                info!(?segment_index, "Archived segment skipped.");
+                continue;
             }
-            .in_current_span(),
-        ),
-    );
-
-    let mut archived_segment_notification_stream = subspace_link
-        .archived_segment_notification_stream()
-        .subscribe();
-
-    spawner.spawn_essential(
-        "archiver",
-        Some("subspace-networking"),
-        Box::pin({
-            let node = node.clone();
-
-            async move {
-                trace!("Subspace DSN archiver started.");
-
-                let mut last_published_segment_index: Option<u64> = None;
-                while let Some(ArchivedSegmentNotification {
-                    archived_segment, ..
-                }) = archived_segment_notification_stream.next().await
-                {
-                    let segment_index = archived_segment.root_block.segment_index();
-                    let first_piece_index = segment_index * u64::from(PIECES_IN_SEGMENT);
-
-                    info!(%segment_index, "Processing a segment.");
-
-                    // skip repeating publication
-                    if let Some(last_published_segment_index) = last_published_segment_index {
-                        if last_published_segment_index == segment_index {
-                            info!(?segment_index, "Archived segment skipped.");
-                            continue;
-                        }
-                    }
-                    let keys_iter = (first_piece_index..)
-                        .take(archived_segment.pieces.count())
-                        .map(|idx| (idx, PieceIndexHash::from_index(idx)))
-                        .map(|(idx, hash)| (idx, hash.to_multihash()));
-
+        }
+        let keys_iter = (first_piece_index..)
+            .take(archived_segment.pieces.count())
+            .map(|idx| (idx, PieceIndexHash::from_index(idx)))
+            .map(|(idx, hash)| (idx, hash.to_multihash()));
+
+        spawner.spawn(
+            "segment-publishing",
+            Some("subspace-networking"),
+            Box::pin({
+                let node = node.clone();
+
+                async move {
                     for ((_idx, key), piece) in keys_iter.zip(archived_segment.pieces.as_pieces()) {
-                        //TODO: restore annoucing after https://github.com/libp2p/rust-libp2p/issues/3048
+                        //TODO: restore announcing after https://github.com/libp2p/rust-libp2p/issues/3048
                         // trace!(?key, ?idx, "Announcing key...");
                         //
                         // let announcing_result = node.start_announcing(key).await;
@@ -145,13 +136,12 @@ where
                         //TODO: ensure republication of failed announcements
                     }
 
-                    last_published_segment_index = Some(segment_index);
                     info!(%segment_index, "Segment processed.");
                 }
-            }
-            .in_current_span()
-        }),
-    );
+                .in_current_span()
+            }),
+        );
 
-    Ok(node)
+        last_published_segment_index = Some(segment_index);
+    }
 }
diff --git a/crates/subspace-service/src/lib.rs b/crates/subspace-service/src/lib.rs
@@ -21,10 +21,11 @@ mod dsn;
 mod pool;
 pub mod rpc;
 
+use crate::dsn::create_dsn_instance;
 pub use crate::pool::FullPool;
 use derive_more::{Deref, DerefMut, Into};
 use domain_runtime_primitives::Hash as DomainHash;
-use dsn::start_dsn_node;
+use dsn::start_dsn_archiver;
 pub use dsn::DsnConfig;
 use frame_system_rpc_runtime_api::AccountNonceApi;
 use futures::channel::oneshot;
@@ -54,6 +55,7 @@ use sp_blockchain::HeaderMetadata;
 use sp_consensus::Error as ConsensusError;
 use sp_consensus_slots::Slot;
 use sp_consensus_subspace::{FarmerPublicKey, SubspaceApi};
+use sp_core::traits::SpawnEssentialNamed;
 use sp_domains::ExecutorApi;
 use sp_objects::ObjectsApi;
 use sp_offchain::OffchainWorkerApi;
@@ -67,7 +69,7 @@ use subspace_fraud_proof::VerifyFraudProof;
 use subspace_networking::libp2p::multiaddr::Protocol;
 use subspace_runtime_primitives::opaque::Block;
 use subspace_runtime_primitives::{AccountId, Balance, Hash, Index as Nonce};
-use tracing::error;
+use tracing::{error, info, Instrument};
 
 /// Error type for Subspace service.
 #[derive(thiserror::Error, Debug)]
@@ -409,10 +411,11 @@ where
         });
 
     let dsn_bootstrap_nodes = {
-        let node = start_dsn_node(
-            &subspace_link,
+        let span = tracing::info_span!(sc_tracing::logging::PREFIX_LOG_SPAN, name = "DSN");
+        let _enter = span.enter();
+
+        let (node, mut node_runner) = create_dsn_instance::<Block, _>(
             config.dsn_config.clone(),
-            task_manager.spawn_essential_handle(),
             piece_cache.clone(),
             Arc::new({
                 let piece_cache = piece_cache.clone();
@@ -436,6 +439,38 @@ where
         )
         .await?;
 
+        info!("Subspace networking initialized: Node ID is {}", node.id());
+
+        task_manager.spawn_essential_handle().spawn_essential(
+            "node-runner",
+            Some("subspace-networking"),
+            Box::pin(
+                async move {
+                    node_runner.run().await;
+                }
+                .in_current_span(),
+            ),
+        );
+
+        let dsn_archiving_fut = start_dsn_archiver(
+            subspace_link
+                .archived_segment_notification_stream()
+                .subscribe(),
+            node.clone(),
+            task_manager.spawn_handle(),
+        );
+
+        task_manager.spawn_essential_handle().spawn_essential(
+            "archiver",
+            Some("subspace-networking"),
+            Box::pin(
+                async move {
+                    dsn_archiving_fut.await;
+                }
+                .in_current_span(),
+            ),
+        );
+
         // Fall back to node itself as bootstrap node for DSN so farmer always has someone to
         // connect to
         if config.dsn_config.bootstrap_nodes.is_empty() {
@@ -485,7 +520,6 @@ where
         client.clone(),
         telemetry.as_ref().map(|telemetry| telemetry.handle()),
         &task_manager.spawn_essential_handle(),
-        config.role.is_authority(),
     );
 
     let (network, system_rpc_tx, tx_handler_controller, network_starter) =
