test: unit tests for state migrations

.github/workflows/coverage.yml

@@ -39,6 +39,7 @@ env:
   FOREST_F3_SIDECAR_FFI_BUILD_OPT_OUT: 1
   FIL_PROOFS_PARAMETER_CACHE: /var/tmp/filecoin-proof-parameters
   RUST_LOG: error
+  FOREST_ACTOR_BUNDLE_PATH: /var/tmp/forest_actor_bundle.car.zst
 
 jobs:
   codecov:
@@ -53,7 +54,7 @@ jobs:
       - uses: taiki-e/install-action@nextest
       - name: Fetch proof params and RPC test snapshots
         run: |
-          cargo run --bin forest-dev --no-default-features --profile quick -- fetch-rpc-tests
+          cargo run --bin forest-dev --no-default-features --profile quick -- fetch-test-snapshots --actor-bundle $FOREST_ACTOR_BUNDLE_PATH
           ls -ahl $FIL_PROOFS_PARAMETER_CACHE
       - name: Generate code coverage
         run: mise codecov

.github/workflows/unit-tests.yml

@@ -77,7 +77,7 @@ jobs:
         uses: taiki-e/install-action@nextest
       - name: Fetch proof params and RPC test snapshots
         run: |
-          cargo run --bin forest-dev --no-default-features --profile quick -- fetch-rpc-tests
+          cargo run --bin forest-dev --no-default-features --profile quick -- fetch-test-snapshots
           ls -ahl $FIL_PROOFS_PARAMETER_CACHE
       - uses: jdx/mise-action@v3
       - run: |

docs/docs/users/reference/cli.sh

@@ -157,7 +157,7 @@ generate_markdown_section "forest-tool" "index backfill"
 
 generate_markdown_section "forest-dev" ""
 
-generate_markdown_section "forest-dev" "fetch-rpc-tests"
+generate_markdown_section "forest-dev" "fetch-test-snapshots"
 
 generate_markdown_section "forest-dev" "state"
 generate_markdown_section "forest-dev" "state compute"

src/benchmark_private/tipset_validation.rs

@@ -19,29 +19,39 @@ pub fn bench_tipset_validation(c: &mut Criterion) {
     let mut group = c.benchmark_group("tipset_validation");
 
     group
-        .bench_function("calibnet@3111900", |b| {
+        .bench_function("calibnet@3408952", |b| {
             let chain = NetworkChain::Calibnet;
-            let epoch = 3111900;
-            let (state_manager, ts) = rt
+            let epoch = 3408952;
+            let (state_manager, ts, ts_next) = rt
                 .block_on(async {
                     let snapshot = get_state_compute_snapshot(&chain, epoch).await?;
-                    prepare_state_compute(&chain, &snapshot, true).await
+                    prepare_state_compute(&chain, &snapshot).await
                 })
                 .unwrap();
-            b.to_async(&rt)
-                .iter(|| state_compute(black_box(state_manager.clone()), black_box(ts.clone())))
+            b.to_async(&rt).iter(|| {
+                state_compute(
+                    black_box(&state_manager),
+                    black_box(ts.clone()),
+                    black_box(&ts_next),
+                )
+            })
         })
-        .bench_function("mainnet@5427431", |b| {
+        .bench_function("mainnet@5709604", |b| {
             let chain = NetworkChain::Mainnet;
-            let epoch = 5427431;
-            let (state_manager, ts) = rt
+            let epoch = 5709604;
+            let (state_manager, ts, ts_next) = rt
                 .block_on(async {
                     let snapshot = get_state_compute_snapshot(&chain, epoch).await?;
-                    prepare_state_compute(&chain, &snapshot, true).await
+                    prepare_state_compute(&chain, &snapshot).await
                 })
                 .unwrap();
-            b.to_async(&rt)
-                .iter(|| state_compute(black_box(state_manager.clone()), black_box(ts.clone())))
+            b.to_async(&rt).iter(|| {
+                state_compute(
+                    black_box(&state_manager),
+                    black_box(ts.clone()),
+                    black_box(&ts_next),
+                )
+            })
         });
 
     group.finish();

src/blocks/tipset.rs

@@ -317,6 +317,10 @@ impl Tipset {
     pub fn parent_state(&self) -> &Cid {
         &self.min_ticket_block().state_root
     }
+    /// Returns the message receipt root for the tipset parent.
+    pub fn parent_message_receipts(&self) -> &Cid {
+        &self.min_ticket_block().message_receipts
+    }
     /// Returns the tipset's calculated weight
     pub fn weight(&self) -> &BigInt {
         &self.min_ticket_block().weight

src/chain_sync/tipset_syncer.rs

@@ -11,6 +11,7 @@ use crate::shim::{
     address::Address, crypto::verify_bls_aggregate, econ::BLOCK_GAS_LIMIT,
     gas::price_list_by_network_version, message::Message, state_tree::StateTree,
 };
+use crate::state_manager::StateLookupPolicy;
 use crate::state_manager::{Error as StateManagerError, StateManager, utils::is_valid_for_sending};
 use crate::{
     blocks::{Block, CachingBlockHeader, Error as ForestBlockError, FullTipset, Tipset},
@@ -278,7 +279,7 @@ async fn validate_block<DB: Blockstore + Sync + Send + 'static>(
         async move {
             let header = block.header();
             let (state_root, receipt_root) = state_manager
-                .tipset_state(&base_tipset)
+                .tipset_state(&base_tipset, StateLookupPolicy::Disabled)
                 .await
                 .map_err(|e| {
                     TipsetSyncerError::Calculation(format!("Failed to calculate state: {e}"))
@@ -439,7 +440,7 @@ async fn check_block_messages<DB: Blockstore + Send + Sync + 'static>(
 
     let mut account_sequences: HashMap<Address, u64> = HashMap::default();
     let (state_root, _) = state_manager
-        .tipset_state(&base_tipset)
+        .tipset_state(&base_tipset, StateLookupPolicy::Disabled)
         .await
         .map_err(|e| TipsetSyncerError::Calculation(format!("Could not update state: {e}")))?;
     let tree =

src/dev/subcommands/mod.rs

@@ -4,6 +4,7 @@
 mod state_cmd;
 
 use crate::cli_shared::cli::HELP_MESSAGE;
+use crate::networks::generate_actor_bundle;
 use crate::rpc::Client;
 use crate::utils::net::{DownloadFileOption, download_file_with_cache};
 use crate::utils::proofs_api::ensure_proof_params_downloaded;
@@ -30,26 +31,43 @@ pub struct Cli {
 /// forest-dev sub-commands
 #[derive(clap::Subcommand)]
 pub enum Subcommand {
-    /// Fetch RPC test snapshots to the local cache
-    FetchRpcTests,
+    /// Fetch test snapshots to the local cache
+    FetchTestSnapshots {
+        // Save actor bundle to
+        #[arg(long)]
+        actor_bundle: Option<PathBuf>,
+    },
     #[command(subcommand)]
     State(state_cmd::StateCommand),
 }
 
 impl Subcommand {
     pub async fn run(self, _client: Client) -> anyhow::Result<()> {
         match self {
-            Self::FetchRpcTests => fetch_rpc_tests().await,
+            Self::FetchTestSnapshots { actor_bundle } => fetch_test_snapshots(actor_bundle).await,
             Self::State(cmd) => cmd.run().await,
         }
     }
 }
 
-async fn fetch_rpc_tests() -> anyhow::Result<()> {
+async fn fetch_test_snapshots(actor_bundle: Option<PathBuf>) -> anyhow::Result<()> {
+    // Prepare proof parameter files
     crate::utils::proofs_api::maybe_set_proofs_parameter_cache_dir_env(
         &crate::Config::default().client.data_dir,
     );
     ensure_proof_params_downloaded().await?;
+
+    // Prepare actor bundles
+    if let Some(actor_bundle) = actor_bundle {
+        generate_actor_bundle(&actor_bundle).await?;
+        println!("Wrote the actors bundle to {}", actor_bundle.display());
+    }
+
+    // Prepare RPC test snapshots
+    fetch_rpc_tests().await
+}
+
+async fn fetch_rpc_tests() -> anyhow::Result<()> {
     let tests = include_str!("../../tool/subcommands/api_cmd/test_snapshots.txt")
         .lines()
         .map(|i| {

src/dev/subcommands/state_cmd.rs

@@ -6,11 +6,7 @@ use crate::{
     chain::{ChainStore, index::ResolveNullTipset},
     chain_sync::{load_full_tipset, tipset_syncer::validate_tipset},
     cli_shared::{chain_path, read_config},
-    db::{
-        MemoryDB, SettingsStoreExt,
-        car::{AnyCar, ManyCar},
-        db_engine::db_root,
-    },
+    db::{SettingsStoreExt, db_engine::db_root},
     genesis::read_genesis_header,
     interpreter::VMTrace,
     networks::{ChainConfig, NetworkChain},
@@ -73,7 +69,7 @@ impl ComputeCommand {
             let (_, config) = read_config(None, Some(chain.clone()))?;
             db_root(&chain_path(&config))?
         };
-        let db = generate_test_snapshot::load_db(&db_root_path)?;
+        let db = generate_test_snapshot::load_db(&db_root_path, Some(&chain)).await?;
         let chain_config = Arc::new(ChainConfig::from_chain(&chain));
         let genesis_header =
             read_genesis_header(None, chain_config.genesis_bytes(&db).await?.as_deref(), &db)
@@ -85,18 +81,30 @@ impl ComputeCommand {
             chain_config,
             genesis_header,
         )?);
-        let ts = {
+        let (ts, ts_next) = {
             // We don't want to track all entries that are visited by `tipset_by_height`
             db.pause_tracking();
             let ts = chain_store.chain_index().tipset_by_height(
                 epoch,
                 chain_store.heaviest_tipset(),
                 ResolveNullTipset::TakeOlder,
             )?;
+            let ts_next = chain_store.chain_index().tipset_by_height(
+                epoch + 1,
+                chain_store.heaviest_tipset(),
+                ResolveNullTipset::TakeNewer,
+            )?;
             db.resume_tracking();
-            SettingsStoreExt::write_obj(&db.tracker, crate::db::setting_keys::HEAD_KEY, ts.key())?;
-            // Only track the desired tipset
-            Tipset::load_required(&db, ts.key())?
+            SettingsStoreExt::write_obj(
+                &db.tracker,
+                crate::db::setting_keys::HEAD_KEY,
+                ts_next.key(),
+            )?;
+            // Only track the desired tipsets
+            (
+                Tipset::load_required(&db, ts.key())?,
+                Tipset::load_required(&db, ts_next.key())?,
+            )
         };
         let epoch = ts.epoch();
         let state_manager = Arc::new(StateManager::new(chain_store)?);
@@ -114,6 +122,16 @@ impl ComputeCommand {
             "epoch: {epoch}, state_root: {state_root}, receipt_root: {receipt_root}, db_snapshot_size: {}",
             human_bytes::human_bytes(db_snapshot.len() as f64)
         );
+        let expected_state_root = *ts_next.parent_state();
+        let expected_receipt_root = *ts_next.parent_message_receipts();
+        anyhow::ensure!(
+            state_root == expected_state_root,
+            "state root mismatch, state_root: {state_root}, expected_state_root: {expected_state_root}"
+        );
+        anyhow::ensure!(
+            receipt_root == expected_receipt_root,
+            "receipt root mismatch, receipt_root: {receipt_root}, expected_receipt_root: {expected_receipt_root}"
+        );
         if let Some(export_db_to) = export_db_to {
             std::fs::write(export_db_to, db_snapshot)?;
         }
@@ -138,39 +156,12 @@ pub struct ReplayComputeCommand {
 impl ReplayComputeCommand {
     pub async fn run(self) -> anyhow::Result<()> {
         let Self { snapshot, chain, n } = self;
-        let snap_car = AnyCar::try_from(&snapshot)?;
-        let ts = snap_car.heaviest_tipset()?;
-        let epoch = ts.epoch();
-        let db = Arc::new(ManyCar::new(MemoryDB::default()).with_read_only(snap_car)?);
-        let chain_config = Arc::new(ChainConfig::from_chain(&chain));
-        let genesis_header =
-            read_genesis_header(None, chain_config.genesis_bytes(&db).await?.as_deref(), &db)
+        let (sm, ts, ts_next) =
+            crate::state_manager::utils::state_compute::prepare_state_compute(&chain, &snapshot)
                 .await?;
-        let chain_store = Arc::new(ChainStore::new(
-            db.clone(),
-            db.clone(),
-            db.clone(),
-            chain_config,
-            genesis_header,
-        )?);
-        let state_manager = Arc::new(StateManager::new(chain_store)?);
         for _ in 0..n.get() {
-            let start = Instant::now();
-            let StateOutput {
-                state_root,
-                receipt_root,
-                ..
-            } = state_manager
-                .compute_tipset_state(
-                    ts.clone(),
-                    crate::state_manager::NO_CALLBACK,
-                    VMTrace::NotTraced,
-                )
+            crate::state_manager::utils::state_compute::state_compute(&sm, ts.clone(), &ts_next)
                 .await?;
-            println!(
-                "epoch: {epoch}, state_root: {state_root}, receipt_root: {receipt_root}, took {}.",
-                humantime::format_duration(start.elapsed())
-            );
         }
         Ok(())
     }
@@ -208,7 +199,7 @@ impl ValidateCommand {
             let (_, config) = read_config(None, Some(chain.clone()))?;
             db_root(&chain_path(&config))?
         };
-        let db = generate_test_snapshot::load_db(&db_root_path)?;
+        let db = generate_test_snapshot::load_db(&db_root_path, Some(&chain)).await?;
         let chain_config = Arc::new(ChainConfig::from_chain(&chain));
         let genesis_header =
             read_genesis_header(None, chain_config.genesis_bytes(&db).await?.as_deref(), &db)
@@ -267,27 +258,14 @@ pub struct ReplayValidateCommand {
 impl ReplayValidateCommand {
     pub async fn run(self) -> anyhow::Result<()> {
         let Self { snapshot, chain, n } = self;
-        let snap_car = AnyCar::try_from(&snapshot)?;
-        let ts = snap_car.heaviest_tipset()?;
-        let epoch = ts.epoch();
-        let db = Arc::new(ManyCar::new(MemoryDB::default()).with_read_only(snap_car)?);
-        let chain_config = Arc::new(ChainConfig::from_chain(&chain));
-        let genesis_header =
-            read_genesis_header(None, chain_config.genesis_bytes(&db).await?.as_deref(), &db)
+        let (sm, fts) =
+            crate::state_manager::utils::state_compute::prepare_state_validate(&chain, &snapshot)
                 .await?;
-        let chain_store = Arc::new(ChainStore::new(
-            db.clone(),
-            db.clone(),
-            db.clone(),
-            chain_config,
-            genesis_header,
-        )?);
-        let state_manager = Arc::new(StateManager::new(chain_store)?);
-        let fts = load_full_tipset(state_manager.chain_store(), ts.key())?;
+        let epoch = fts.epoch();
         for _ in 0..n.get() {
             let fts = fts.clone();
             let start = Instant::now();
-            validate_tipset(&state_manager, fts, None).await?;
+            validate_tipset(&sm, fts, None).await?;
             println!(
                 "epoch: {epoch}, took {}.",
                 humantime::format_duration(start.elapsed())
@@ -298,5 +276,7 @@ impl ReplayValidateCommand {
 }
 
 fn disable_tipset_cache() {
-    unsafe { std::env::set_var("FOREST_TIPSET_CACHE_DISABLED", "1") };
+    unsafe {
+        std::env::set_var("FOREST_TIPSET_CACHE_DISABLED", "1");
+    }
 }