Implement network version 12 state migration / actors v4 migration (#…

…1101)

* initial code for nv12 state migration

* simple root migration impl

* fix borrow issues with blockstore

* /s/v2/v3 and /s/v3/v4

* migration works, but incorrect

* few nits

* try for_each_concurrent and a few refactors

* revert to next.await and refactors

* migration log

* address code review comments

* Box<dyn Error> -> String

* make blockstore Send + Sync

* concurrent state migration

* chore: cargo fmt

* replace Mutex<Map<_>> with DashMap

* move common migration code out of miner migrator

* clippy fixes

* cleanup unused

* cargo fmt

* more clippy fixes

* relocate a few comments

* Address code review changes

* code review: rename arc_store to store

* more compact in_state initialization in miner.rs

* migrate_state should return error when prev state not equal new state post migration

* make clippy happy

* cargo fmt

Cargo.toml

@@ -16,6 +16,7 @@ members = [
     "vm/runtime",
     "vm/state_tree",
     "vm/interpreter",
+    "vm/state_migration",
     "node/clock",
     "node/db",
     "node/rpc",

blockchain/state_manager/src/lib.rs

@@ -229,17 +229,19 @@ where
         V: ProofVerifier,
         CB: FnMut(&Cid, &ChainMessage, &ApplyRet) -> Result<(), String>,
     {
-        let mut buf_store = BufferedBlockStore::new(self.blockstore());
+        let db = self.blockstore_cloned();
+        let mut buf_store = Arc::new(BufferedBlockStore::new(db.as_ref()));
+        let store = buf_store.as_ref();
         let lb_wrapper = SMLookbackWrapper {
             sm: self,
-            store: &buf_store,
+            store,
             tipset,
             verifier: PhantomData::<V>::default(),
         };
 
         let mut vm = VM::<_, _, _, _, _, V>::new(
             p_state,
-            &buf_store,
+            store,
             epoch,
             rand,
             base_fee,
@@ -249,14 +251,18 @@ where
         )?;
 
         // Apply tipset messages
-        let receipts = vm.apply_block_messages(messages, parent_epoch, epoch, callback)?;
+        let receipts =
+            vm.apply_block_messages(messages, parent_epoch, epoch, buf_store.clone(), callback)?;
 
         // Construct receipt root from receipts
         let rect_root = Amt::new_from_iter(self.blockstore(), receipts)?;
         // Flush changes to blockstore
         let state_root = vm.flush()?;
         // Persist changes connected to root
-        buf_store.flush(&state_root)?;
+        Arc::get_mut(&mut buf_store)
+            .expect("failed getting store reference")
+            .flush(&state_root)
+            .expect("buffered blockstore flush failed");
 
         Ok((state_root, rect_root))
     }
@@ -1283,7 +1289,7 @@ impl<'sm, 'ts, DB, BS, V> LookbackStateGetter<'sm, BS> for SMLookbackWrapper<'sm
 where
     // Yes, both are needed, because the VM should only use the buffered store
     DB: BlockStore + Send + Sync + 'static,
-    BS: BlockStore,
+    BS: BlockStore + Send + Sync,
     V: ProofVerifier,
 {
     fn state_lookback(&self, round: ChainEpoch) -> Result<StateTree<'sm, BS>, Box<dyn StdError>> {

ipld/blockstore/Cargo.toml

@@ -13,6 +13,8 @@ db = { package = "forest_db", version = "0.1" }
 encoding = { package = "forest_encoding", version = "0.2.1" }
 forest_ipld = { optional = true, version = "0.1" }
 byteorder = "1.3.2"
+dashmap = "4.0.2"
+
 [dev-dependencies]
 commcid = { path = "../../utils/commcid" }