feat(sync): implement execution layer snap sync

Cargo.toml

@@ -10,6 +10,7 @@ members = [
     "crates/mempool",
     "crates/rpc",
     "crates/metrics",
+    "crates/sync",
 ]
 resolver = "2"
 
@@ -148,9 +149,11 @@ informalsystems-malachitebft-metrics = { version = "0.5", default-features = fal
 #   when sync requests fail, enabling retries instead of silent timeouts
 # - sync: Adds SkipToHeight effect to allow nodes to skip ahead when peers
 #   have pruned old blocks and cannot serve the required heights
+# - app: Updated to use new sync config fields (tip_first_sync, tip_first_buffer)
 [patch.crates-io]
 informalsystems-malachitebft-network = { path = "vendor/malachitebft-network" }
 informalsystems-malachitebft-sync = { path = "vendor/malachitebft-sync" }
+informalsystems-malachitebft-app = { path = "vendor/malachitebft-app" }
 
 [profile.release]
 opt-level = 3

crates/execution/src/engine.rs

@@ -329,6 +329,14 @@ impl<P: Provider + Clone> ExecutionEngine<P> {
         self.database.provider()
     }
 
+    /// Get the current block number.
+    ///
+    /// Returns the block number of the most recently executed block, or 0 if
+    /// no blocks have been executed yet.
+    pub fn current_block_number(&self) -> u64 {
+        self.current_block
+    }
+
     /// Process all transactions in a block.
     ///
     /// Returns a tuple containing receipts, cumulative gas used, logs, and total fees.

crates/node/Cargo.toml

@@ -21,6 +21,7 @@ cipherbft-consensus = { path = "../consensus", features = ["malachite"] }
 cipherbft-rpc = { path = "../rpc" }
 cipherbft-mempool = { path = "../mempool" }
 cipherbft-metrics = { path = "../metrics" }
+cipherbft-sync = { path = "../sync" }
 
 # Ethereum primitives (for genesis types in tests and transaction parsing)
 alloy-primitives = { version = "1", features = ["serde"] }

crates/node/src/config.rs

@@ -77,6 +77,38 @@ pub const DEFAULT_RPC_WS_PORT: u16 = 8546;
 /// Default metrics port (Prometheus standard)
 pub const DEFAULT_METRICS_PORT: u16 = 9100;
 
+/// Default snap sync enabled setting
+pub const DEFAULT_SNAP_SYNC_ENABLED: bool = true;
+
+/// Default minimum peers required to start sync
+pub const DEFAULT_MIN_SYNC_PEERS: usize = 3;
+
+/// Default sync request timeout in seconds
+pub const DEFAULT_SYNC_TIMEOUT_SECS: u64 = 30;
+
+/// Default block gap threshold to trigger snap sync (vs block-by-block)
+pub const DEFAULT_SNAP_SYNC_THRESHOLD: u64 = 1024;
+
+/// Serde default function for snap_sync_enabled
+fn default_snap_sync_enabled() -> bool {
+    DEFAULT_SNAP_SYNC_ENABLED
+}
+
+/// Serde default function for min_sync_peers
+fn default_min_sync_peers() -> usize {
+    DEFAULT_MIN_SYNC_PEERS
+}
+
+/// Serde default function for sync_timeout
+fn default_sync_timeout() -> u64 {
+    DEFAULT_SYNC_TIMEOUT_SECS
+}
+
+/// Serde default function for snap_sync_threshold
+fn default_snap_sync_threshold() -> u64 {
+    DEFAULT_SNAP_SYNC_THRESHOLD
+}
+
 /// Serde default function for rpc_http_port
 fn default_rpc_http_port() -> u16 {
     DEFAULT_RPC_HTTP_PORT
@@ -109,6 +141,52 @@ pub struct PeerConfig {
     pub worker_addrs: Vec<SocketAddr>,
 }
 
+/// Sync configuration for state synchronization
+///
+/// Controls how the node performs snap sync when joining the network
+/// or recovering from being behind.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct SyncConfig {
+    /// Enable snap sync for fast bootstrap.
+    ///
+    /// When enabled, nodes that are significantly behind will use snap sync
+    /// to download state directly rather than replaying all blocks.
+    #[serde(default = "default_snap_sync_enabled")]
+    pub snap_sync_enabled: bool,
+
+    /// Minimum peers required to start sync.
+    ///
+    /// The node will wait until it has at least this many peers with
+    /// consistent state before beginning snap sync.
+    #[serde(default = "default_min_sync_peers")]
+    pub min_sync_peers: usize,
+
+    /// Sync request timeout in seconds.
+    ///
+    /// Maximum time to wait for a response to a sync request before
+    /// retrying with a different peer.
+    #[serde(default = "default_sync_timeout")]
+    pub sync_timeout_secs: u64,
+
+    /// Block gap threshold to trigger snap sync (vs block-by-block).
+    ///
+    /// If the node is behind by more than this many blocks, it will
+    /// use snap sync. Otherwise, it will sync block-by-block.
+    #[serde(default = "default_snap_sync_threshold")]
+    pub snap_sync_threshold: u64,
+}
+
+impl Default for SyncConfig {
+    fn default() -> Self {
+        Self {
+            snap_sync_enabled: DEFAULT_SNAP_SYNC_ENABLED,
+            min_sync_peers: DEFAULT_MIN_SYNC_PEERS,
+            sync_timeout_secs: DEFAULT_SYNC_TIMEOUT_SECS,
+            snap_sync_threshold: DEFAULT_SNAP_SYNC_THRESHOLD,
+        }
+    }
+}
+
 /// Node configuration
 ///
 /// Keys are loaded from the keyring backend specified by `keyring_backend`.
@@ -199,6 +277,13 @@ pub struct NodeConfig {
     /// Port for Prometheus metrics endpoint (default: 9100)
     #[serde(default = "default_metrics_port")]
     pub metrics_port: u16,
+
+    /// Sync configuration for state synchronization.
+    ///
+    /// Controls snap sync behavior for fast bootstrap when joining
+    /// the network or recovering from being behind.
+    #[serde(default)]
+    pub sync: SyncConfig,
 }
 
 /// Test configuration with keypairs for local testing
@@ -253,6 +338,7 @@ impl NodeConfig {
             rpc_http_port: DEFAULT_RPC_HTTP_PORT + (index as u16),
             rpc_ws_port: DEFAULT_RPC_WS_PORT + (index as u16),
             metrics_port: DEFAULT_METRICS_PORT + (index as u16),
+            sync: SyncConfig::default(),
         };
 
         LocalTestConfig {

crates/node/src/execution_bridge.rs

@@ -631,6 +631,86 @@ impl ExecutionBridge {
         let execution = self.execution.read().await;
         Arc::clone(execution.staking_precompile())
     }
+
+    /// Get the last executed block hash.
+    ///
+    /// Returns the hash of the most recently executed block, or B256::ZERO
+    /// if no blocks have been executed yet.
+    pub fn last_block_hash(&self) -> B256 {
+        self.last_block_hash
+            .read()
+            .map(|guard| *guard)
+            .unwrap_or(B256::ZERO)
+    }
+
+    /// Get the current block number.
+    ///
+    /// Returns the block number of the most recently executed block.
+    /// This requires acquiring the execution lock.
+    pub async fn current_block_number(&self) -> u64 {
+        let execution = self.execution.read().await;
+        execution.current_block_number()
+    }
+
+    /// Execute a block directly from BlockInput (for sync replay).
+    ///
+    /// This method is used during sync to replay blocks that have been
+    /// downloaded from peers. It bypasses the Cut conversion since the
+    /// transactions are already flattened.
+    ///
+    /// # Arguments
+    ///
+    /// * `input` - Block input containing ordered transactions
+    ///
+    /// # Returns
+    ///
+    /// Returns `BlockExecutionResult` containing execution result with properly
+    /// computed block hash and parent hash.
+    pub async fn execute_block_input(
+        &self,
+        input: BlockInput,
+    ) -> anyhow::Result<BlockExecutionResult> {
+        let block_number = input.block_number;
+        let timestamp = input.timestamp;
+        let parent_hash = input.parent_hash;
+        let transactions = input.transactions.clone();
+
+        let mut execution = self.execution.write().await;
+
+        let result = execution
+            .execute_block(input)
+            .map_err(|e| anyhow::anyhow!("Block execution failed: {}", e))?;
+
+        // Compute and store the new block hash for the next block's parent_hash
+        let new_block_hash = compute_block_hash(
+            block_number,
+            timestamp,
+            parent_hash,
+            result.state_root,
+            result.transactions_root,
+            result.receipts_root,
+        );
+
+        // Update the last block hash for the next execution
+        if let Ok(mut guard) = self.last_block_hash.write() {
+            *guard = new_block_hash;
+        }
+
+        debug!(
+            height = block_number,
+            block_hash = %new_block_hash,
+            parent_hash = %parent_hash,
+            "Sync block hash updated"
+        );
+
+        Ok(BlockExecutionResult {
+            execution_result: result,
+            block_hash: new_block_hash,
+            parent_hash,
+            timestamp,
+            executed_transactions: transactions,
+        })
+    }
 }
 
 /// Compute a deterministic block hash from block components.

crates/node/src/lib.rs

@@ -15,15 +15,20 @@ pub mod network;
 pub mod network_api;
 pub mod node;
 pub mod supervisor;
+pub mod sync_executor;
+pub mod sync_network;
+pub mod sync_runner;
+pub mod sync_server;
 pub mod util;
 pub mod worker_pool_adapter;
 
 pub use client_config::ClientConfig;
 pub use config::{
-    generate_keypair, generate_local_configs, LocalTestConfig, NodeConfig, PeerConfig,
+    generate_keypair, generate_local_configs, LocalTestConfig, NodeConfig, PeerConfig, SyncConfig,
     CIPHERD_GENESIS_PATH_ENV, CIPHERD_HOME_ENV, DEFAULT_GENESIS_FILENAME, DEFAULT_HOME_DIR,
     DEFAULT_KEYRING_BACKEND, DEFAULT_KEYS_DIR, DEFAULT_KEY_NAME, DEFAULT_METRICS_PORT,
-    DEFAULT_RPC_HTTP_PORT, DEFAULT_RPC_WS_PORT,
+    DEFAULT_MIN_SYNC_PEERS, DEFAULT_RPC_HTTP_PORT, DEFAULT_RPC_WS_PORT, DEFAULT_SNAP_SYNC_ENABLED,
+    DEFAULT_SNAP_SYNC_THRESHOLD, DEFAULT_SYNC_TIMEOUT_SECS,
 };
 pub use execution_bridge::ExecutionBridge;
 pub use execution_sync::{ExecutionSyncConfig, ExecutionSyncTracker, SyncAction};
@@ -36,4 +41,8 @@ pub use mempool_state::ExecutionStateValidator;
 pub use network_api::{NodeNetworkApi, TcpNetworkApi};
 pub use node::Node;
 pub use supervisor::{NodeSupervisor, ShutdownError};
+pub use sync_executor::ExecutionBridgeSyncExecutor;
+pub use sync_network::{create_sync_adapter, wire_sync_to_network, SyncNetworkAdapter};
+pub use sync_runner::{create_sync_manager, run_snap_sync, should_snap_sync, SyncResult};
+pub use sync_server::SnapSyncServer;
 pub use worker_pool_adapter::WorkerPoolAdapter;

crates/node/src/main.rs

@@ -560,6 +560,7 @@ fn cmd_init(
         rpc_http_port: cipherd::DEFAULT_RPC_HTTP_PORT,
         rpc_ws_port: cipherd::DEFAULT_RPC_WS_PORT,
         metrics_port: cipherd::DEFAULT_METRICS_PORT,
+        sync: cipherd::SyncConfig::default(),
     };
 
     let config_path = config_dir.join("node.json");
@@ -975,6 +976,7 @@ fn cmd_testnet_init_files(
             rpc_http_port: cipherd::DEFAULT_RPC_HTTP_PORT + (i as u16 * 10),
             rpc_ws_port: cipherd::DEFAULT_RPC_WS_PORT + (i as u16 * 10),
             metrics_port: cipherd::DEFAULT_METRICS_PORT + (i as u16 * 10),
+            sync: cipherd::SyncConfig::default(),
         };
 
         let config_path = config_dir.join("node.json");