feat: enhance transaction execution with isolated runtime snapshots and improve state management

Author: jamesatomc

crates/kanari-core/src/engine.rs

@@ -326,9 +326,20 @@ impl BlockchainEngine {
             let mut executed_count = 0;
             let failed_count = 0;
             for signed_tx in transactions {
+                let snapshot_runtime = if persist_objects {
+                    None
+                } else {
+                    let snapshot_store = {
+                        let state = state_arc.read().unwrap_or_else(|e| e.into_inner());
+                        state.snapshot_store()?
+                    };
+                    Some(self.runtime_pool[0].spawn_isolated_worker_from_store(snapshot_store)?)
+                };
+                let runtime = snapshot_runtime.as_ref().unwrap_or(&self.runtime_pool[0]);
+
                 let changeset = self.execute_transaction_with_runtime_internal(
                     &signed_tx.transaction,
-                    &self.runtime_pool[0],
+                    runtime,
                     state_arc,
                     false,
                     timestamp,
@@ -369,12 +380,27 @@ impl BlockchainEngine {
         }
 
         for wave in waves {
+            let wave_snapshot_store = if persist_objects {
+                None
+            } else {
+                let state = state_arc.read().unwrap_or_else(|e| e.into_inner());
+                Some(state.snapshot_store()?)
+            };
+
             let results: Vec<Result<ChangeSet>> = if has_module_publish {
                 wave.iter()
                     .map(|signed_tx| {
+                        let snapshot_runtime = match &wave_snapshot_store {
+                            Some(store) => Some(
+                                self.runtime_pool[0]
+                                    .spawn_isolated_worker_from_store(store.clone())?,
+                            ),
+                            None => None,
+                        };
+                        let runtime = snapshot_runtime.as_ref().unwrap_or(&self.runtime_pool[0]);
                         self.execute_transaction_with_runtime_internal(
                             &signed_tx.transaction,
-                            &self.runtime_pool[0],
+                            runtime,
                             state_arc,
                             false,
                             timestamp,
@@ -386,7 +412,17 @@ impl BlockchainEngine {
                 wave.par_iter()
                     .enumerate()
                     .map(|(i, signed_tx)| {
-                        let runtime = &self.runtime_pool[i % self.runtime_pool.len()];
+                        let runtime_idx = i % self.runtime_pool.len();
+                        let snapshot_runtime = match &wave_snapshot_store {
+                            Some(store) => Some(
+                                self.runtime_pool[runtime_idx]
+                                    .spawn_isolated_worker_from_store(store.clone())?,
+                            ),
+                            None => None,
+                        };
+                        let runtime = snapshot_runtime
+                            .as_ref()
+                            .unwrap_or(&self.runtime_pool[runtime_idx]);
                         self.execute_transaction_with_runtime_internal(
                             &signed_tx.transaction,
                             runtime,

crates/kanari-core/src/engine/apply_checkpoint.rs

@@ -30,7 +30,16 @@ impl BlockchainEngine {
         timestamp_ms: u64,
         persist_objects: bool,
     ) -> Result<()> {
-        let runtime = &self.runtime_pool[0];
+        let snapshot_runtime = if persist_objects {
+            None
+        } else {
+            let snapshot_store = {
+                let state = state_arc.read().unwrap_or_else(|e| e.into_inner());
+                state.snapshot_store()?
+            };
+            Some(self.runtime_pool[0].spawn_isolated_worker_from_store(snapshot_store)?)
+        };
+        let runtime = snapshot_runtime.as_ref().unwrap_or(&self.runtime_pool[0]);
         let mut state_write = state_arc.write().unwrap_or_else(|e| e.into_inner());
         let clock_id = runtime.ensure_system_clock(&mut state_write)?;
         let changeset = runtime.execute_clock_consensus_commit_prologue(clock_id, timestamp_ms)?;
@@ -184,7 +193,32 @@ impl BlockchainEngine {
             )?;
         }
 
-        self.finalize_checkpoint(checkpoint, verified_state)
+        let checkpoint_transactions = checkpoint.transactions.clone();
+        self.finalize_checkpoint(checkpoint, verified_state)?;
+        self.persist_checkpoint_publish_modules(&checkpoint_transactions)
+    }
+
+    fn persist_checkpoint_publish_modules(&self, transactions: &[SignedTransaction]) -> Result<()> {
+        let mut persisted_any = false;
+
+        for signed_tx in transactions {
+            let kanari_types::transaction::Transaction::PublishModule { module_bytes, .. } =
+                &signed_tx.transaction
+            else {
+                continue;
+            };
+
+            self.runtime_pool[0].persist_published_module_bytes(module_bytes)?;
+            persisted_any = true;
+        }
+
+        if persisted_any {
+            for runtime in &self.runtime_pool {
+                runtime.reload_vm_cache()?;
+            }
+        }
+
+        Ok(())
     }
 
     pub fn apply_checkpoint(&self, checkpoint: Checkpoint) -> Result<()> {

crates/kanari-core/src/engine/dag_integration.rs

@@ -410,13 +410,19 @@ impl DagConsensusIntegration {
             return Ok(checkpoint);
         }
 
-        if !allow_root_override {
+        if !allow_root_override
+            || BlockchainEngine::strict_checkpoint_roots_required()
+            || checkpoint.state_root != previous_checkpoint_root
+        {
             anyhow::bail!(
-                "{} Checkpoint {} state root mismatch: expected={}, computed={}",
+                "{} Checkpoint {} state root mismatch: expected={}, computed={}, previous={}, allow_override={}, strict_roots={}",
                 log_prefix,
                 checkpoint.sequence,
                 hex::encode(&checkpoint.state_root),
-                hex::encode(&computed_root)
+                hex::encode(&computed_root),
+                hex::encode(&previous_checkpoint_root),
+                allow_root_override,
+                BlockchainEngine::strict_checkpoint_roots_required()
             );
         }
 

crates/kanari-core/src/engine/mempool.rs

@@ -219,8 +219,12 @@ impl BlockchainEngine {
                 }
             }
             let state_arc = Arc::new(RwLock::new(state_snapshot));
+            let snapshot_store = {
+                let state = state_arc.read().unwrap_or_else(|e| e.into_inner());
+                state.snapshot_store()?
+            };
             let runtime = self.runtime_pool[0]
-                .spawn_isolated_worker()
+                .spawn_isolated_worker_from_store(snapshot_store)
                 .context("Failed to create isolated runtime for immediate execution")?;
             let changeset =
                 self.execute_transaction_with_runtime(&tx, &runtime, &state_arc, None)?;
@@ -245,16 +249,24 @@ impl BlockchainEngine {
         txs.into_par_iter()
             .map(|tx| {
                 let thread_idx = rayon::current_thread_index().unwrap_or(0);
-                let runtime = &self.runtime_pool[thread_idx % self.runtime_pool.len()];
-
-                let result = self.execute_transaction_with_runtime_internal(
-                    &tx.transaction,
-                    runtime,
-                    state_arc,
-                    true,
-                    None,
-                    false,
-                );
+                let runtime_idx = thread_idx % self.runtime_pool.len();
+
+                let result = (|| {
+                    let snapshot_store = {
+                        let state = state_arc.read().unwrap_or_else(|e| e.into_inner());
+                        state.snapshot_store()?
+                    };
+                    let runtime = self.runtime_pool[runtime_idx]
+                        .spawn_isolated_worker_from_store(snapshot_store)?;
+                    self.execute_transaction_with_runtime_internal(
+                        &tx.transaction,
+                        &runtime,
+                        state_arc,
+                        true,
+                        None,
+                        false,
+                    )
+                })();
 
                 let final_result = match result {
                     Ok(cs) => Ok((tx.transaction_hash().to_vec(), cs)),

crates/kanari-core/src/engine/queries.rs

@@ -104,12 +104,21 @@ impl BlockchainEngine {
         };
 
         let module_id = ModuleId::new(addr, ident);
-        let runtime = &self.runtime_pool[0];
+        let snapshot_store = self.state_read().snapshot_store().ok()?;
+        let runtime = self.runtime_pool[0]
+            .spawn_isolated_worker_from_store(snapshot_store)
+            .ok()?;
         runtime.get_module_bytes(&module_id)
     }
 
     pub fn list_all_modules(&self) -> Vec<(String, String)> {
-        let runtime = &self.runtime_pool[0];
+        let Ok(snapshot_store) = self.state_read().snapshot_store() else {
+            return Vec::new();
+        };
+        let Ok(runtime) = self.runtime_pool[0].spawn_isolated_worker_from_store(snapshot_store)
+        else {
+            return Vec::new();
+        };
         runtime
             .list_modules()
             .into_iter()
@@ -359,12 +368,12 @@ impl BlockchainEngine {
         let (computed_root, verified_state, to_execute) =
             self.prepare_checkpoint_state(&checkpoint_to_apply)?;
 
-        if checkpoint_to_apply.transactions.is_empty() {
-            let previous_checkpoint_root = {
-                let chain = self.blockchain.read().unwrap_or_else(|e| e.into_inner());
-                chain.latest_checkpoint().state_root.clone()
-            };
+        let previous_checkpoint_root = {
+            let chain = self.blockchain.read().unwrap_or_else(|e| e.into_inner());
+            chain.latest_checkpoint().state_root.clone()
+        };
 
+        if checkpoint_to_apply.transactions.is_empty() {
             if checkpoint_to_apply.state_root != previous_checkpoint_root {
                 info!(
                     "[SYNC] Canonicalizing empty checkpoint #{} root: advertised={}, canonical={}",
@@ -380,8 +389,18 @@ impl BlockchainEngine {
             &computed_root,
             &checkpoint_to_apply.state_root,
         )? {
+            if checkpoint_to_apply.state_root != previous_checkpoint_root {
+                anyhow::bail!(
+                    "[SYNC] Rejecting checkpoint #{} with non-canonical state root: advertised={}, computed={}, previous={}",
+                    checkpoint_to_apply.sequence,
+                    hex::encode(&checkpoint_to_apply.state_root),
+                    hex::encode(&computed_root),
+                    hex::encode(&previous_checkpoint_root)
+                );
+            }
+
             info!(
-                "[SYNC] Canonicalizing checkpoint #{} root after replay: advertised={}, computed={}",
+                "[SYNC] Canonicalizing provisional checkpoint #{} root after replay: advertised_previous_root={}, computed={}",
                 checkpoint_to_apply.sequence,
                 hex::encode(&checkpoint_to_apply.state_root),
                 hex::encode(&computed_root)
@@ -408,7 +427,8 @@ impl BlockchainEngine {
         type_args: &[String],
         args: &[Vec<u8>],
     ) -> Result<serde_json::Value> {
-        let runtime = &self.runtime_pool[0];
+        let snapshot_store = self.state_read().snapshot_store()?;
+        let runtime = self.runtime_pool[0].spawn_isolated_worker_from_store(snapshot_store)?;
         runtime.execute_view_function(package_addr, module_name, function_name, type_args, args)
     }
 }
@@ -417,6 +437,19 @@ impl BlockchainEngine {
 mod tests {
     use super::*;
     use centauri::consensus::Checkpoint;
+    use kanari_crypto::keys::{CurveType, generate_keypair};
+    use kanari_move_runtime_v1::state::Account;
+    use kanari_types::balance::BalanceRecord;
+    use kanari_types::kanari::KANARI_TOKEN_TYPE;
+    use kanari_types::transaction::{SignedTransaction, Transaction};
+    use move_core_types::account_address::AccountAddress;
+
+    fn fund_account(engine: &BlockchainEngine, address: &str, balance: u64) {
+        let addr = AccountAddress::from_hex_literal(address).unwrap();
+        let mut account = Account::with_native_balance(addr, balance);
+        account.set_token_balance(KANARI_TOKEN_TYPE.to_string(), BalanceRecord::new(balance));
+        engine.state_write().save_account(&account).unwrap();
+    }
 
     #[test]
     fn sync_checkpoint_from_data_applies_next_checkpoint() {
@@ -456,4 +489,47 @@ mod tests {
         assert_eq!(chain.latest_checkpoint().sequence, 1);
         assert_eq!(chain.latest_checkpoint().state_root, previous_root);
     }
+
+    #[test]
+    fn sync_checkpoint_from_data_rejects_non_provisional_root_mismatch() {
+        let engine = BlockchainEngine::new_in_memory().unwrap();
+        let sender = generate_keypair(CurveType::Ed25519).unwrap();
+        let recipient = generate_keypair(CurveType::Ed25519).unwrap();
+        fund_account(&engine, &sender.address, 2_000_000);
+
+        let transaction = Transaction::new_transfer_with_gas(
+            sender.tagged_address(),
+            recipient.address,
+            1,
+            0,
+            1_000_000,
+            1,
+        );
+        let mut signed_tx = SignedTransaction::new(transaction);
+        signed_tx
+            .sign(&sender.private_key, sender.curve_type)
+            .unwrap();
+
+        let (prev_hash, previous_root) = {
+            let chain = engine.blockchain.read().unwrap_or_else(|e| e.into_inner());
+            (
+                chain.latest_checkpoint().hash().unwrap(),
+                chain.latest_checkpoint().state_root.clone(),
+            )
+        };
+        let mut bad_root = vec![9u8; 32];
+        if bad_root == previous_root {
+            bad_root = vec![8u8; 32];
+        }
+        let checkpoint = Checkpoint::new(1, vec![], vec![signed_tx], bad_root, 42, prev_hash);
+        let sync_data = CheckpointSyncData { checkpoint };
+
+        let err = engine.sync_checkpoint_from_data(&sync_data).unwrap_err();
+
+        assert!(
+            err.to_string().contains("non-canonical state root"),
+            "unexpected error: {err:#}"
+        );
+        assert_eq!(engine.get_stats().height, 0);
+    }
 }

crates/kanari-node/src/app.rs

@@ -5,7 +5,7 @@ use anyhow::Result;
 use kanari_core::BlockchainEngine;
 use kanari_core::engine::AccountInfo;
 use kanari_crypto::wallet::list_wallet_files;
-use kanari_rpc_server::start_server;
+use kanari_rpc_server::start_server_with_transaction_broadcaster;
 use kanari_types::address::Address as KanariAddress;
 use kanari_types::kanari::{KANARI_TOKEN_TYPE, KanariModule};
 use libp2p::identity::Keypair;
@@ -421,8 +421,27 @@ pub async fn run_node(
 
     let engine_for_rpc = engine.clone();
     let bind_addr_clone = bind_addr.clone();
+    let rpc_network_tx = network_tx.clone();
     tokio::spawn(async move {
-        if let Err(e) = start_server(engine_for_rpc, &bind_addr_clone).await {
+        let transaction_broadcaster = Arc::new(
+            move |signed_tx: &kanari_types::transaction::SignedTransaction| {
+                serialize_and_queue_message(
+                    &rpc_network_tx,
+                    signed_tx,
+                    P2PMessage::NewTransaction,
+                    "Failed to serialize transaction for broadcast",
+                    "Failed to queue transaction broadcast",
+                );
+            },
+        );
+
+        if let Err(e) = start_server_with_transaction_broadcaster(
+            engine_for_rpc,
+            &bind_addr_clone,
+            transaction_broadcaster,
+        )
+        .await
+        {
             tracing::error!("RPC server error: {}", e);
         }
     });