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_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_sync_finished
+                .as_ref()
+                .map(|sync_finished| !sync_finished.load(Ordering::SeqCst))
+                .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_sync_finished: Option<Arc<AtomicBool>>,
     ) -> Self {
         Self {
             force_authoring,
             pause_sync,
+            domain_sync_finished,
             inner: substrate_sync_oracle,
         }
     }

crates/subspace-service/src/domains/snap_sync_orchestrator.rs

@@ -3,6 +3,7 @@
 use crate::sync_from_dsn::snap_sync::{DefaultTargetBlockProvider, SnapSyncTargetBlockProvider};
 use async_trait::async_trait;
 use parking_lot::Mutex;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use subspace_core_primitives::BlockNumber;
 use tokio::sync::Notify;
@@ -23,15 +24,15 @@ pub struct SnapSyncOrchestrator {
     notify_consensus_snap_sync_unblocked: Notify,
     consensus_snap_sync_block_number: Mutex<Option<BlockNumber>>,
     notify_domain_snap_sync_finished: Notify,
-    domain_snap_sync_finished: Mutex<bool>,
+    domain_snap_sync_finished: Arc<AtomicBool>,
 }
 
 #[async_trait]
 impl SnapSyncTargetBlockProvider for SnapSyncOrchestrator {
     async fn target_block(&self) -> Option<BlockNumber> {
-        self.consensus_snap_sync_unblocked().await;
+        self.notify_consensus_snap_sync_unblocked.notified().await;
 
-        self.target_consensus_snap_sync_block_number()
+        *self.consensus_snap_sync_block_number.lock()
     }
 }
 
@@ -48,23 +49,10 @@ impl SnapSyncOrchestrator {
             notify_consensus_snap_sync_unblocked: Notify::new(),
             consensus_snap_sync_block_number: Mutex::new(None),
             notify_domain_snap_sync_finished: Notify::new(),
-            domain_snap_sync_finished: Mutex::new(false),
+            domain_snap_sync_finished: Arc::new(AtomicBool::new(false)),
         }
     }
 
-    /// Returns optional target block for consensus chain snap sync. None means target block is
-    /// not defined yet.
-    pub fn target_consensus_snap_sync_block_number(&self) -> Option<BlockNumber> {
-        *self.consensus_snap_sync_block_number.lock()
-    }
-
-    /// Wait for the allowing signal for the consensus chain snap sync.
-    pub async fn consensus_snap_sync_unblocked(&self) {
-        debug!("Waiting for a signal to start consensus chain snap sync.");
-        self.notify_consensus_snap_sync_unblocked.notified().await;
-        debug!("Finished waiting for a signal to start consensus chain snap sync.");
-    }
-
     /// Unblocks (allows) consensus chain snap sync with the given target block.
     pub fn unblock_consensus_snap_sync(&self, target_block_number: BlockNumber) {
         debug!(%target_block_number, "Allowed starting consensus chain snap sync.");
@@ -76,31 +64,24 @@ impl SnapSyncOrchestrator {
     }
 
     /// Returns true if domain snap sync finished.
-    pub fn domain_snap_sync_finished(&self) -> bool {
-        *self.domain_snap_sync_finished.lock()
+    pub fn domain_snap_sync_finished(&self) -> Arc<AtomicBool> {
+        self.domain_snap_sync_finished.clone()
     }
 
     /// Signal that domain snap sync finished.
     pub fn mark_domain_snap_sync_finished(&self) {
         debug!("Signal that domain snap sync finished.");
-        *self.domain_snap_sync_finished.lock() = true;
+        self.domain_snap_sync_finished.store(true, Ordering::SeqCst);
 
         self.notify_domain_snap_sync_finished.notify_waiters();
     }
 
-    /// Wait for a signal that domain snap sync finished.
-    pub async fn domain_snap_sync_finished_blocking(&self) {
-        debug!("Waiting for a signal that domain snap sync finished.");
-        self.notify_domain_snap_sync_finished.notified().await;
-        debug!("Finished waiting for a signal that domain snap sync finished.");
-    }
-
     /// Unblock all processes (synchronization cancelled).
     pub fn unblock_all(&self) {
         debug!("Allow all processes (synchronization cancelled).");
 
         self.notify_consensus_snap_sync_unblocked.notify_waiters();
         self.notify_domain_snap_sync_finished.notify_waiters();
-        *self.domain_snap_sync_finished.lock() = true;
+        self.domain_snap_sync_finished.store(true, Ordering::SeqCst);
     }
 }

crates/subspace-service/src/lib.rs

@@ -1038,6 +1038,9 @@ where
         config.base.force_authoring,
         Arc::clone(&pause_sync),
         sync_service.clone(),
+        snap_sync_orchestrator
+            .as_ref()
+            .map(|orchestrator| orchestrator.domain_snap_sync_finished()),
     );
 
     let subspace_archiver = tokio::task::block_in_place(|| {

domains/client/relayer/src/worker.rs

@@ -11,7 +11,6 @@ use sp_mmr_primitives::MmrApi;
 use sp_runtime::traits::{CheckedSub, NumberFor, One};
 use sp_runtime::SaturatedConversion;
 use std::sync::Arc;
-use subspace_service::domains::snap_sync_orchestrator::SnapSyncOrchestrator;
 
 pub async fn gossip_channel_updates<Client, Block, CBlock, SO>(
     chain_id: ChainId,
@@ -137,7 +136,6 @@ pub async fn start_relaying_messages<CClient, Client, CBlock, Block, SO>(
     confirmation_depth_k: NumberFor<CBlock>,
     sync_oracle: SO,
     gossip_message_sink: GossipMessageSink,
-    snap_sync_orchestrator: Option<Arc<SnapSyncOrchestrator>>,
 ) where
     Block: BlockT,
     CBlock: BlockT,
@@ -177,16 +175,6 @@ pub async fn start_relaying_messages<CClient, Client, CBlock, Block, SO>(
             continue;
         }
 
-        if let Some(ref snap_sync_orchestrator) = snap_sync_orchestrator {
-            if !snap_sync_orchestrator.domain_snap_sync_finished() {
-                tracing::debug!(
-                    target: LOG_TARGET,
-                    "Domain snap sync: skipping message processing..."
-                );
-                continue;
-            }
-        }
-
         let Some(confirmed_block_number) = imported_block
             .header
             .number()

domains/service/src/domain.rs

@@ -506,10 +506,6 @@ where
     // TODO: Implement when block tree is ready.
     let domain_confirmation_depth = 256u32;
 
-    let snap_sync_orchestrator = consensus_chain_sync_params
-        .as_ref()
-        .map(|params| params.snap_sync_orchestrator.clone());
-
     let operator = Operator::new(
         Box::new(spawn_essential.clone()),
         OperatorParams {
@@ -550,7 +546,6 @@ where
             // since domain sync oracle will always return `synced` due to force sync being set.
             consensus_network_sync_oracle.clone(),
             gossip_message_sink.clone(),
-            snap_sync_orchestrator,
         );
 
         spawn_essential.spawn_essential_blocking("domain-relayer", None, Box::pin(relayer_worker));