Add fast sync algorithm.

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

@@ -210,6 +210,7 @@ fn main() -> Result<(), Error> {
                 sync_from_dsn: true,
                 is_timekeeper: false,
                 timekeeper_cpu_cores: Default::default(),
+                fast_sync_enabled: false,
             };
 
             let partial_components = subspace_service::new_partial::<PosTable, RuntimeApi>(

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

@@ -8,7 +8,7 @@ use sc_cli::{
     TransactionPoolParams, RPC_DEFAULT_PORT,
 };
 use sc_informant::OutputFormat;
-use sc_network::config::{MultiaddrWithPeerId, NonReservedPeerMode, SetConfig};
+use sc_network::config::{MultiaddrWithPeerId, NonReservedPeerMode, SetConfig, SyncMode};
 use sc_service::{BlocksPruning, Configuration, PruningMode};
 use sc_storage_monitor::StorageMonitorParams;
 use sc_telemetry::TelemetryEndpoints;
@@ -162,6 +162,8 @@ enum StatePruningMode {
     Archive,
     /// Keep only the data of finalized blocks.
     ArchiveCanonical,
+    /// Keep the data of the last number of finalized blocks.
+    Number(u32),
 }
 
 impl FromStr for StatePruningMode {
@@ -171,17 +173,21 @@ impl FromStr for StatePruningMode {
         match input {
             "archive" => Ok(Self::Archive),
             "archive-canonical" => Ok(Self::ArchiveCanonical),
-            _ => Err("Invalid state pruning mode specified".to_string()),
+            n => n
+                .parse()
+                .map_err(|_| "Invalid block pruning mode specified".to_string())
+                .map(Self::Number),
         }
     }
 }
 
 impl fmt::Display for StatePruningMode {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.write_str(match self {
-            Self::Archive => "archive",
-            Self::ArchiveCanonical => "archive-canonical",
-        })
+        match self {
+            Self::Archive => f.write_str("archive"),
+            Self::ArchiveCanonical => f.write_str("archive-canonical"),
+            Self::Number(n) => f.write_str(n.to_string().as_str()),
+        }
     }
 }
 
@@ -256,6 +262,7 @@ impl PruningOptions {
         match self.state_pruning {
             StatePruningMode::Archive => PruningMode::ArchiveAll,
             StatePruningMode::ArchiveCanonical => PruningMode::ArchiveCanonical,
+            StatePruningMode::Number(num) => PruningMode::blocks_pruning(num),
         }
     }
 
@@ -397,6 +404,10 @@ pub(super) struct ConsensusChainOptions {
 
     #[clap(flatten)]
     timekeeper_options: TimekeeperOptions,
+
+    /// Experimental support of state-only sync using DSN.
+    #[arg(long, default_value_t = false)]
+    fast_sync: bool,
 }
 
 pub(super) struct PrometheusConfiguration {
@@ -440,6 +451,7 @@ pub(super) fn create_consensus_chain_configuration(
         sync_from_dsn,
         storage_monitor,
         mut timekeeper_options,
+        fast_sync,
     } = consensus_node_options;
 
     let transaction_pool;
@@ -584,7 +596,14 @@ pub(super) fn create_consensus_chain_configuration(
         chain_spec: Box::new(chain_spec),
         informant_output_format: OutputFormat { enable_color },
     };
-    let consensus_chain_config = Configuration::from(consensus_chain_config);
+    let mut consensus_chain_config = Configuration::from(consensus_chain_config);
+    // TODO: revisit SyncMode change after https://github.com/paritytech/polkadot-sdk/issues/4407
+    if fast_sync {
+        consensus_chain_config.network.sync_mode = SyncMode::LightState {
+            skip_proofs: true,
+            storage_chain_mode: false,
+        };
+    }
 
     let pot_external_entropy =
         derive_pot_external_entropy(&consensus_chain_config, pot_external_entropy)?;
@@ -653,6 +672,7 @@ pub(super) fn create_consensus_chain_configuration(
             sync_from_dsn,
             is_timekeeper: timekeeper_options.timekeeper,
             timekeeper_cpu_cores: timekeeper_options.timekeeper_cpu_cores,
+            fast_sync_enabled: fast_sync,
         },
         dev,
         pot_external_entropy,

crates/subspace-service/Cargo.toml

@@ -29,6 +29,7 @@ pallet-transaction-payment-rpc = { git = "https://github.com/subspace/polkadot-s
 parity-scale-codec = "3.6.9"
 parking_lot = "0.12.2"
 prometheus-client = "0.22.2"
+prost = "0.12"
 sc-basic-authorship = { git = "https://github.com/subspace/polkadot-sdk", rev = "808269708cf5375526755797e8f9a9986016727d" }
 sc-chain-spec = { git = "https://github.com/subspace/polkadot-sdk", rev = "808269708cf5375526755797e8f9a9986016727d" }
 sc-client-api = { git = "https://github.com/subspace/polkadot-sdk", rev = "808269708cf5375526755797e8f9a9986016727d" }
@@ -52,6 +53,7 @@ sc-telemetry = { git = "https://github.com/subspace/polkadot-sdk", rev = "808269
 sc-tracing = { git = "https://github.com/subspace/polkadot-sdk", rev = "808269708cf5375526755797e8f9a9986016727d" }
 sc-transaction-pool = { git = "https://github.com/subspace/polkadot-sdk", rev = "808269708cf5375526755797e8f9a9986016727d" }
 sc-transaction-pool-api = { git = "https://github.com/subspace/polkadot-sdk", rev = "808269708cf5375526755797e8f9a9986016727d" }
+sc-utils = { git = "https://github.com/subspace/polkadot-sdk", rev = "808269708cf5375526755797e8f9a9986016727d" }
 schnorrkel = "0.11.4"
 sp-api = { git = "https://github.com/subspace/polkadot-sdk", rev = "808269708cf5375526755797e8f9a9986016727d" }
 sp-blockchain = { git = "https://github.com/subspace/polkadot-sdk", rev = "808269708cf5375526755797e8f9a9986016727d" }

crates/subspace-service/src/config.rs

@@ -1,5 +1,5 @@
 use crate::dsn::DsnConfig;
-use crate::sync_from_dsn::DsnSyncPieceGetter;
+use crate::sync_from_dsn::import_blocks::DsnSyncPieceGetter;
 use sc_chain_spec::ChainSpec;
 use sc_network::config::{
     MultiaddrWithPeerId, NetworkConfiguration, NodeKeyConfig, SetConfig, SyncMode, TransportConfig,
@@ -269,6 +269,8 @@ pub struct SubspaceConfiguration {
     pub is_timekeeper: bool,
     /// CPU cores that timekeeper can use
     pub timekeeper_cpu_cores: HashSet<usize>,
+    /// Enables state-only sync using DSN.
+    pub fast_sync_enabled: bool,
 }
 
 impl Deref for SubspaceConfiguration {

crates/subspace-service/src/lib.rs

@@ -74,7 +74,7 @@ use sc_proof_of_time::source::gossip::pot_gossip_peers_set_config;
 use sc_proof_of_time::source::{PotSlotInfo, PotSourceWorker};
 use sc_proof_of_time::verifier::PotVerifier;
 use sc_service::error::Error as ServiceError;
-use sc_service::{Configuration, NetworkStarter, SpawnTasksParams, TaskManager};
+use sc_service::{ClientExt, Configuration, NetworkStarter, SpawnTasksParams, TaskManager};
 use sc_subspace_block_relay::{
     build_consensus_relay, BlockRelayConfigurationError, NetworkWrapper,
 };
@@ -228,18 +228,19 @@ pub type FullClient<RuntimeApi> = sc_service::TFullClient<Block, RuntimeApi, Run
 pub type FullBackend = sc_service::TFullBackend<Block>;
 pub type FullSelectChain = sc_consensus::LongestChain<FullBackend, Block>;
 
-struct SubspaceExtensionsFactory<PosTable, Client, DomainBlock> {
+struct SubspaceExtensionsFactory<PosTable, Block: BlockT, Client, DomainBlock> {
     kzg: Kzg,
     client: Arc<Client>,
     backend: Arc<FullBackend>,
     pot_verifier: PotVerifier,
     executor: Arc<RuntimeExecutor>,
     domains_executor: Arc<sc_domains::RuntimeExecutor>,
+    fast_sync_state: Arc<Mutex<Option<NumberFor<Block>>>>,
     _pos_table: PhantomData<(PosTable, DomainBlock)>,
 }
 
 impl<PosTable, Block, Client, DomainBlock> ExtensionsFactory<Block>
-    for SubspaceExtensionsFactory<PosTable, Client, DomainBlock>
+    for SubspaceExtensionsFactory<PosTable, Block, Client, DomainBlock>
 where
     PosTable: Table,
     Block: BlockT,
@@ -269,6 +270,7 @@ where
         exts.register(PotExtension::new({
             let client = Arc::clone(&self.client);
             let pot_verifier = self.pot_verifier.clone();
+            let fast_sync_state = self.fast_sync_state.clone();
 
             Box::new(
                 move |parent_hash, slot, proof_of_time, quick_verification| {
@@ -298,6 +300,7 @@ where
                             return false;
                         }
                     };
+
                     let parent_pre_digest = match extract_pre_digest(&parent_header) {
                         Ok(parent_pre_digest) => parent_pre_digest,
                         Err(error) => {
@@ -318,6 +321,22 @@ where
                         return false;
                     }
 
+                    // Check for fast-sync state
+                    {
+                        if let Some(state_block_number) = fast_sync_state.lock().as_ref() {
+                            let parent_block_number = *parent_header.number();
+                            if parent_block_number < *state_block_number {
+                                debug!(
+                                    %parent_block_number,
+                                    %state_block_number,
+                                    "Skipped PoT verification because of the fast sync state block."
+                                );
+
+                                return true;
+                            }
+                        }
+                    }
+
                     let pot_parameters = match client.runtime_api().pot_parameters(parent_hash) {
                         Ok(pot_parameters) => pot_parameters,
                         Err(error) => {
@@ -423,6 +442,8 @@ where
     pub sync_target_block_number: Arc<AtomicU32>,
     /// Telemetry
     pub telemetry: Option<Telemetry>,
+    /// The first block with state enabled by fast-sync.
+    pub fast_sync_state: Arc<Mutex<Option<NumberFor<Block>>>>,
 }
 
 type PartialComponents<RuntimeApi> = sc_service::PartialComponents<
@@ -490,17 +511,21 @@ where
 
     let executor = Arc::new(executor);
 
+    let fast_sync_state = Arc::<Mutex<Option<NumberFor<Block>>>>::default();
     client
         .execution_extensions()
-        .set_extensions_factory(SubspaceExtensionsFactory::<PosTable, _, DomainBlock> {
-            kzg: kzg.clone(),
-            client: Arc::clone(&client),
-            pot_verifier: pot_verifier.clone(),
-            executor: executor.clone(),
-            domains_executor: Arc::new(domains_executor),
-            backend: backend.clone(),
-            _pos_table: PhantomData,
-        });
+        .set_extensions_factory(
+            SubspaceExtensionsFactory::<PosTable, Block, _, DomainBlock> {
+                kzg: kzg.clone(),
+                client: Arc::clone(&client),
+                pot_verifier: pot_verifier.clone(),
+                executor: executor.clone(),
+                domains_executor: Arc::new(domains_executor),
+                backend: backend.clone(),
+                fast_sync_state: fast_sync_state.clone(),
+                _pos_table: PhantomData,
+            },
+        );
 
     let telemetry = telemetry.map(|(worker, telemetry)| {
         task_manager
@@ -598,6 +623,7 @@ where
         pot_verifier,
         sync_target_block_number,
         telemetry,
+        fast_sync_state,
     };
 
     Ok(PartialComponents {
@@ -709,8 +735,19 @@ where
         pot_verifier,
         sync_target_block_number,
         mut telemetry,
+        fast_sync_state,
     } = other;
 
+    // Clear block gap after fast sync on reruns.
+    // Substrate detects a gap and inserts on each sync.
+    if config.fast_sync_enabled {
+        let finalized_hash_existed = client.info().finalized_hash != client.info().genesis_hash;
+        if finalized_hash_existed {
+            debug!(client_info=?client.info(), "Clear block gap after fast-sync.");
+            client.clear_block_gap();
+        }
+    }
+
     let offchain_indexing_enabled = config.offchain_worker.indexing_enabled;
     let (node, bootstrap_nodes) = match config.subspace_networking {
         SubspaceNetworking::Reuse {
@@ -806,7 +843,7 @@ where
         }
     };
 
-    let import_queue_service = import_queue.service();
+    let import_queue_service = Arc::new(tokio::sync::Mutex::new(import_queue.service()));
     let network_wrapper = Arc::new(NetworkWrapper::default());
     let block_relay = Some(
         build_consensus_relay(
@@ -890,6 +927,7 @@ where
         );
 
     network_wrapper.set(network_service.clone());
+
     if config.sync_from_dsn {
         let dsn_sync_piece_getter = config.dsn_piece_getter.unwrap_or_else(|| {
             Arc::new(PieceProvider::new(
@@ -916,6 +954,9 @@ where
             sync_target_block_number,
             pause_sync,
             dsn_sync_piece_getter,
+            config.fast_sync_enabled,
+            fast_sync_state,
+            sync_service.clone(),
         );
         task_manager
             .spawn_handle()