Full domain chain snap sync.

crates/sc-consensus-subspace/src/slot_worker.rs

@@ -93,6 +93,7 @@ where
 {
     force_authoring: bool,
     pause_sync: Arc<AtomicBool>,
+    domain_snap_sync_finished: Option<Arc<AtomicBool>>,
     inner: SO,
 }
 
@@ -106,6 +107,11 @@ where
         // (default state), it also accounts for DSN sync
         (!self.force_authoring && self.inner.is_major_syncing())
             || self.pause_sync.load(Ordering::Acquire)
+            || self
+                .domain_snap_sync_finished
+                .as_ref()
+                .map(|sync_finished| !sync_finished.load(Ordering::Acquire))
+                .unwrap_or_default()
     }
 
     fn is_offline(&self) -> bool {
@@ -122,10 +128,12 @@ where
         force_authoring: bool,
         pause_sync: Arc<AtomicBool>,
         substrate_sync_oracle: SO,
+        domain_snap_sync_finished: Option<Arc<AtomicBool>>,
     ) -> Self {
         Self {
             force_authoring,
             pause_sync,
+            domain_snap_sync_finished,
             inner: substrate_sync_oracle,
         }
     }

crates/subspace-malicious-operator/src/bin/subspace-malicious-operator.rs

@@ -217,12 +217,14 @@ fn main() -> Result<(), Error> {
 
             let keystore = partial_components.keystore_container.keystore();
 
-            let consensus_chain_node = subspace_service::new_full::<PosTable, _>(
+            let consensus_chain_node =
+                subspace_service::new_full::<PosTable, _>(
                 consensus_chain_config,
                 partial_components,
                 None,
                 true,
                 SlotProportion::new(3f32 / 4f32),
+                None,
             )
             .await
             .map_err(|error| {

crates/subspace-malicious-operator/src/malicious_domain_instance_starter.rs

@@ -13,7 +13,7 @@ use sc_cli::CliConfiguration;
 use sc_consensus_subspace::block_import::BlockImportingNotification;
 use sc_consensus_subspace::notification::SubspaceNotificationStream;
 use sc_consensus_subspace::slot_worker::NewSlotNotification;
-use sc_network::NetworkPeers;
+use sc_network::{NetworkPeers, NetworkRequest};
 use sc_transaction_pool_api::OffchainTransactionPoolFactory;
 use sc_utils::mpsc::{TracingUnboundedReceiver, TracingUnboundedSender};
 use sp_api::ProvideRuntimeApi;
@@ -28,6 +28,7 @@ use std::sync::Arc;
 use subspace_runtime::RuntimeApi as CRuntimeApi;
 use subspace_runtime_primitives::opaque::Block as CBlock;
 use subspace_runtime_primitives::AccountId;
+use subspace_service::domains::ConsensusChainSyncParams;
 use subspace_service::FullClient as CFullClient;
 
 /// `DomainInstanceStarter` used to start a domain instance node based on the given
@@ -101,7 +102,8 @@ impl DomainInstanceStarter {
                     block_importing_notification.block_number,
                     block_importing_notification.acknowledgement_sender,
                 )
-            });
+            })
+            .boxed();
 
         let new_slot_notification_stream = || {
             new_slot_notification_stream
@@ -161,6 +163,9 @@ impl DomainInstanceStarter {
                     // Always set it to `None` to not running the normal bundle producer
                     maybe_operator_id: None,
                     confirmation_depth_k: chain_constants.confirmation_depth_k(),
+                    consensus_chain_sync_params: None::<
+                        ConsensusChainSyncParams<_, Arc<dyn NetworkRequest + Sync + Send>>,
+                    >,
                 };
 
                 let mut domain_node = domain_service::new_full::<
@@ -173,6 +178,7 @@ impl DomainInstanceStarter {
                     evm_domain_runtime::RuntimeApi,
                     AccountId20,
                     _,
+                    _,
                 >(domain_params)
                 .await?;
 
@@ -219,6 +225,9 @@ impl DomainInstanceStarter {
                     // Always set it to `None` to not running the normal bundle producer
                     maybe_operator_id: None,
                     confirmation_depth_k: chain_constants.confirmation_depth_k(),
+                    consensus_chain_sync_params: None::<
+                        ConsensusChainSyncParams<_, Arc<dyn NetworkRequest + Sync + Send>>,
+                    >,
                 };
 
                 let mut domain_node = domain_service::new_full::<
@@ -231,6 +240,7 @@ impl DomainInstanceStarter {
                     auto_id_domain_runtime::RuntimeApi,
                     AccountId32,
                     _,
+                    _,
                 >(domain_params)
                 .await?;
 

crates/subspace-node/src/commands/run.rs

@@ -23,16 +23,18 @@ use domain_runtime_primitives::opaque::Block as DomainBlock;
 use futures::FutureExt;
 use sc_cli::Signals;
 use sc_consensus_slots::SlotProportion;
-use sc_service::{BlocksPruning, PruningMode};
 use sc_storage_monitor::StorageMonitorService;
 use sc_transaction_pool_api::OffchainTransactionPoolFactory;
 use sc_utils::mpsc::tracing_unbounded;
 use sp_core::traits::SpawnEssentialNamed;
 use sp_messenger::messages::ChainId;
 use std::env;
+use std::sync::Arc;
 use subspace_metrics::{start_prometheus_metrics_server, RegistryAdapter};
 use subspace_runtime::{Block, RuntimeApi};
 use subspace_service::config::ChainSyncMode;
+use subspace_service::domains::snap_sync_orchestrator::SnapSyncOrchestrator;
+use subspace_service::domains::ConsensusChainSyncParams;
 use tracing::{debug, error, info, info_span, warn};
 
 /// Options for running a node
@@ -118,29 +120,22 @@ pub async fn run(run_options: RunOptions) -> Result<(), Error> {
     info!("🏷  Node name: {}", subspace_configuration.network.node_name);
     info!("💾 Node path: {}", base_path.display());
 
-    if maybe_domain_configuration.is_some() && subspace_configuration.sync == ChainSyncMode::Snap {
-        return Err(Error::Other(
-            "Snap sync mode is not supported for domains, use full sync".to_string(),
-        ));
-    }
-
-    if maybe_domain_configuration.is_some()
-        && (matches!(
-            subspace_configuration.blocks_pruning,
-            BlocksPruning::Some(_)
-        ) || matches!(
-            subspace_configuration.state_pruning,
-            Some(PruningMode::Constrained(_))
-        ))
+    let fork_id = subspace_configuration
+        .base
+        .chain_spec
+        .fork_id()
+        .map(String::from);
+    let snap_sync_orchestrator = if maybe_domain_configuration.is_some()
+        && subspace_configuration.sync == ChainSyncMode::Snap
     {
-        return Err(Error::Other(
-            "Running an operator requires both `--blocks-pruning` and `--state-pruning` to be set \
-            to either `archive` or `archive-canonical`"
-                .to_string(),
-        ));
-    }
+        Some(Arc::new(SnapSyncOrchestrator::new()))
+    } else {
+        None
+    };
 
     let mut task_manager = {
+        let mut segment_headers_store = None;
+        let subspace_link;
         let consensus_chain_node = {
             let span = info_span!("Consensus");
             let _enter = span.enter();
@@ -159,6 +154,10 @@ pub async fn run(run_options: RunOptions) -> Result<(), Error> {
                 ))
             })?;
 
+            subspace_link = partial_components.other.subspace_link.clone();
+
+            segment_headers_store.replace(partial_components.other.segment_headers_store.clone());
+
             let full_node_fut = subspace_service::new_full::<PosTable, _>(
                 subspace_configuration,
                 partial_components,
@@ -169,6 +168,7 @@ pub async fn run(run_options: RunOptions) -> Result<(), Error> {
                     }),
                 true,
                 SlotProportion::new(3f32 / 4f32),
+                snap_sync_orchestrator.clone(),
             );
 
             full_node_fut.await.map_err(|error| {
@@ -278,25 +278,26 @@ pub async fn run(run_options: RunOptions) -> Result<(), Error> {
             };
 
             let domain_start_options = DomainStartOptions {
-                consensus_client: consensus_chain_node.client,
+                consensus_client: consensus_chain_node.client.clone(),
                 consensus_offchain_tx_pool_factory: OffchainTransactionPoolFactory::new(
                     consensus_chain_node.transaction_pool,
                 ),
-                consensus_network: consensus_chain_node.network_service,
+                consensus_network: consensus_chain_node.network_service.clone(),
                 block_importing_notification_stream: consensus_chain_node
                     .block_importing_notification_stream,
                 pot_slot_info_stream: consensus_chain_node.pot_slot_info_stream,
-                consensus_network_sync_oracle: consensus_chain_node.sync_service,
+                consensus_network_sync_oracle: consensus_chain_node.sync_service.clone(),
                 domain_message_receiver,
                 gossip_message_sink,
             };
 
             consensus_chain_node
                 .task_manager
                 .spawn_essential_handle()
-                .spawn_essential_blocking(
-                    "domain",
-                    Some("domains"),
+                .spawn_essential_blocking("domain", Some("domains"), {
+                    let consensus_chain_network_service =
+                        consensus_chain_node.network_service.clone();
+                    let consensus_chain_sync_service = consensus_chain_node.sync_service.clone();
                     Box::pin(async move {
                         let span = info_span!("Domain");
                         let _enter = span.enter();
@@ -305,6 +306,7 @@ pub async fn run(run_options: RunOptions) -> Result<(), Error> {
                             &*domain_start_options.consensus_client,
                             domain_configuration.domain_id,
                         );
+
                         let bootstrap_result = match bootstrap_result_fut.await {
                             Ok(bootstrap_result) => bootstrap_result,
                             Err(error) => {
@@ -313,17 +315,30 @@ pub async fn run(run_options: RunOptions) -> Result<(), Error> {
                             }
                         };
 
+                        let consensus_chain_sync_params =
+                            snap_sync_orchestrator.map(|snap_sync_orchestrator| {
+                                ConsensusChainSyncParams {
+                                    snap_sync_orchestrator,
+                                    fork_id,
+                                    network_service: consensus_chain_network_service,
+                                    sync_service: consensus_chain_sync_service,
+                                    backend: consensus_chain_node.backend.clone(),
+                                    subspace_link,
+                                }
+                            });
+
                         let start_domain = run_domain(
                             bootstrap_result,
                             domain_configuration,
                             domain_start_options,
+                            consensus_chain_sync_params,
                         );
 
                         if let Err(error) = start_domain.await {
                             error!(%error, "Domain starter exited with an error");
                         }
-                    }),
-                );
+                    })
+                });
         };
 
         consensus_chain_node.network_starter.start_network();