Merge pull request #966 from subspace/dsn-sync-for-node

Fix DSN sync for node

crates/subspace-node/Cargo.toml

@@ -38,6 +38,7 @@ sc-cli = { version = "0.10.0-dev", git = "https://github.com/subspace/substrate"
 sc-client-api = { version = "4.0.0-dev", git = "https://github.com/subspace/substrate", rev = "6d57dbc639bb3d9460dabeccb063cc6556452535" }
 sc-consensus = { version = "0.10.0-dev", git = "https://github.com/subspace/substrate", rev = "6d57dbc639bb3d9460dabeccb063cc6556452535" }
 sc-consensus-slots = { version = "0.10.0-dev", git = "https://github.com/subspace/substrate", rev = "6d57dbc639bb3d9460dabeccb063cc6556452535" }
+sc-consensus-subspace = { version = "0.1.0", path = "../sc-consensus-subspace" }
 sc-subspace-chain-specs = { version = "0.1.0", path = "../sc-subspace-chain-specs" }
 sc-executor = { version = "0.10.0-dev", git = "https://github.com/subspace/substrate", rev = "6d57dbc639bb3d9460dabeccb063cc6556452535", features = ["wasmtime"] }
 sc-service = { version = "0.10.0-dev", git = "https://github.com/subspace/substrate", rev = "6d57dbc639bb3d9460dabeccb063cc6556452535", default-features = false, features = ["wasmtime"] }
@@ -58,7 +59,6 @@ subspace-runtime-primitives = { version = "0.1.0", path = "../subspace-runtime-p
 subspace-service = { version = "0.1.0", path = "../subspace-service" }
 system-domain-runtime = { version = "0.1.0", path = "../../domains/runtime/system" }
 thiserror = "1.0.32"
-tokio = "1.21.2"
 
 [build-dependencies]
 substrate-build-script-utils = { version = "3.0.0", git = "https://github.com/subspace/substrate", rev = "6d57dbc639bb3d9460dabeccb063cc6556452535" }

crates/subspace-node/src/bin/subspace-node.rs

@@ -196,10 +196,21 @@ fn main() -> Result<(), Error> {
                     client,
                     import_queue,
                     task_manager,
+                    other: (_block_import, subspace_link, _telemetry),
                     ..
                 } = subspace_service::new_partial::<RuntimeApi, ExecutorDispatch>(&config)?;
+
+                sc_consensus_subspace::start_subspace_archiver(
+                    &subspace_link,
+                    client.clone(),
+                    None,
+                    &task_manager.spawn_essential_handle(),
+                    config.role.is_authority(),
+                );
+
                 Ok((
-                    cmd.run(client, import_queue).map_err(Error::SubstrateCli),
+                    cmd.run(client, import_queue, task_manager.spawn_essential_handle())
+                        .map_err(Error::SubstrateCli),
                     task_manager,
                 ))
             })?;

crates/subspace-node/src/import_blocks_from_dsn.rs

@@ -22,6 +22,7 @@ use sc_consensus::{BlockImportError, BlockImportStatus, IncomingBlock, Link};
 use sc_service::ImportQueue;
 use sc_tracing::tracing::{debug, info, trace};
 use sp_consensus::BlockOrigin;
+use sp_core::traits::SpawnEssentialNamed;
 use sp_runtime::generic::BlockId;
 use sp_runtime::traits::{Block as BlockT, Header, NumberFor};
 use std::sync::Arc;
@@ -58,15 +59,26 @@ pub struct ImportBlocksFromDsnCmd {
 
 impl ImportBlocksFromDsnCmd {
     /// Run the import-blocks command
-    pub async fn run<B, C, IQ>(&self, client: Arc<C>, import_queue: IQ) -> sc_cli::Result<()>
+    pub async fn run<B, C, IQ>(
+        &self,
+        client: Arc<C>,
+        import_queue: IQ,
+        spawner: impl SpawnEssentialNamed,
+    ) -> sc_cli::Result<()>
     where
         C: HeaderBackend<B> + BlockBackend<B> + Send + Sync + 'static,
         B: BlockT + for<'de> serde::Deserialize<'de>,
         IQ: sc_service::ImportQueue<B> + 'static,
     {
-        import_blocks(self.bootstrap_node.clone(), client, import_queue, false)
-            .await
-            .map_err(Into::into)
+        import_blocks(
+            self.bootstrap_node.clone(),
+            client,
+            import_queue,
+            &spawner,
+            false,
+        )
+        .await
+        .map_err(Into::into)
     }
 }
 
@@ -109,7 +121,7 @@ impl<B: BlockT> Link<B> for WaitLink<B> {
             B::Hash,
         )>,
     ) {
-        println!("Imported {imported} blocks");
+        debug!("Imported {imported} blocks");
         self.imported_blocks += imported as u64;
 
         for result in results {
@@ -126,6 +138,7 @@ async fn import_blocks<B, IQ, C>(
     bootstrap_nodes: Vec<Multiaddr>,
     client: Arc<C>,
     mut import_queue: IQ,
+    spawner: &impl SpawnEssentialNamed,
     force: bool,
 ) -> Result<(), sc_service::Error>
 where
@@ -142,9 +155,13 @@ where
     .await
     .map_err(|error| sc_service::Error::Other(error.to_string()))?;
 
-    tokio::spawn(async move {
-        node_runner.run().await;
-    });
+    spawner.spawn_essential(
+        "node-runner",
+        Some("subspace-networking"),
+        Box::pin(async move {
+            node_runner.run().await;
+        }),
+    );
 
     debug!("Waiting for connected peers...");
     let _ = node.wait_for_connected_peers().await;
@@ -263,10 +280,28 @@ where
         }
     }
 
+    while link.imported_blocks < imported_blocks {
+        futures::future::poll_fn(|ctx| {
+            import_queue.poll_actions(ctx, &mut link);
+
+            Poll::Ready(())
+        })
+        .await;
+
+        if let Some(WaitLinkError { error, hash }) = &link.error {
+            return Err(sc_service::Error::Other(format!(
+                "Stopping block import after #{} blocks on {} because of an error: {}",
+                link.imported_blocks, hash, error
+            )));
+        }
+    }
+
     info!(
-        "🎉 Imported {} blocks, best #{}, exiting...",
+        "🎉 Imported {} blocks, best #{}/#{}, check against reliable sources to make sure it is a \
+        block on canonical chain",
         imported_blocks,
-        client.info().best_number
+        client.info().best_number,
+        client.info().best_hash
     );
 
     Ok(())