Merge branch 'main' into elmattic/fix-chain-get-events

Author: elmattic

docs/docs/users/reference/env_variables.md

@@ -47,6 +47,7 @@ process.
 | `FOREST_TRACE_FILTER_MAX_RESULT`                          | positive integer                 | 500                                            | 1000                                                          | Sets the maximum results returned per request by `trace_filter`                                                       |
 | `FOREST_CHAIN_INDEXER_ENABLED`                            | 1 or true                        | false                                          | 1                                                             | Whether or not to index the chain to support the Ethereum RPC API                                                     |
 | `FOREST_MESSAGES_IN_TIPSET_CACHE_SIZE`                    | positive integer                 | 100                                            | 42                                                            | The size of an internal cache of tipsets to messages                                                                  |
+| `FOREST_STATE_MIGRATION_DB_WRITE_BUFFER`                  | non-negative integer             | 10000                                          | 100000                                                        | The size of db write buffer for state migration (`~10MB` RAM per `10k` buffer)                                        |
 
 ### `FOREST_F3_SIDECAR_FFI_BUILD_OPT_OUT`
 

docs/yarn.lock

@@ -10271,8 +10271,8 @@ __metadata:
   linkType: hard
 
 "http-proxy-middleware@npm:^2.0.3":
-  version: 2.0.7
-  resolution: "http-proxy-middleware@npm:2.0.7"
+  version: 2.0.9
+  resolution: "http-proxy-middleware@npm:2.0.9"
   dependencies:
     "@types/http-proxy": "npm:^1.17.8"
     http-proxy: "npm:^1.18.1"
@@ -10284,7 +10284,7 @@ __metadata:
   peerDependenciesMeta:
     "@types/express":
       optional: true
-  checksum: 10c0/8d00a61eb215b83826460b07489d8bb095368ec16e02a9d63e228dcf7524e7c20d61561e5476de1391aecd4ec32ea093279cdc972115b311f8e0a95a24c9e47e
+  checksum: 10c0/8e9032af625f7c9f2f0d318f6cdb14eb725cc16ffe7b4ccccea25cf591fa819bb7c3bb579e0b543e0ae9c73059b505a6d728290c757bff27bae526a6ed11c05e
   languageName: node
   linkType: hard
 

src/db/blockstore_with_write_buffer.rs

@@ -0,0 +1,115 @@
+// Copyright 2019-2025 ChainSafe Systems
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+use ahash::{HashMap, HashMapExt};
+use cid::Cid;
+use fvm_ipld_blockstore::Blockstore;
+use itertools::Itertools;
+use parking_lot::RwLock;
+
+pub struct BlockstoreWithWriteBuffer<DB: Blockstore> {
+    inner: DB,
+    buffer: RwLock<HashMap<Cid, Vec<u8>>>,
+    buffer_capacity: usize,
+}
+
+impl<DB: Blockstore> Blockstore for BlockstoreWithWriteBuffer<DB> {
+    fn get(&self, k: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
+        if let Some(v) = self.buffer.read().get(k) {
+            return Ok(Some(v.clone()));
+        }
+        self.inner.get(k)
+    }
+
+    fn has(&self, k: &Cid) -> anyhow::Result<bool> {
+        Ok(self.buffer.read().contains_key(k) || self.inner.has(k)?)
+    }
+
+    fn put_keyed(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
+        {
+            let mut buffer = self.buffer.write();
+            buffer.insert(*k, block.to_vec());
+        }
+        self.flush_buffer_if_needed()
+    }
+}
+
+impl<DB: Blockstore> BlockstoreWithWriteBuffer<DB> {
+    pub fn new_with_capacity(inner: DB, buffer_capacity: usize) -> Self {
+        Self {
+            inner,
+            buffer_capacity,
+            buffer: RwLock::new(HashMap::with_capacity(buffer_capacity)),
+        }
+    }
+
+    fn flush_buffer(&self) -> anyhow::Result<()> {
+        let records = {
+            let mut buffer = self.buffer.write();
+            buffer.drain().collect_vec()
+        };
+        self.inner.put_many_keyed(records)
+    }
+
+    fn flush_buffer_if_needed(&self) -> anyhow::Result<()> {
+        if self.buffer.read().len() >= self.buffer_capacity {
+            self.flush_buffer()
+        } else {
+            Ok(())
+        }
+    }
+}
+
+impl<DB: Blockstore> Drop for BlockstoreWithWriteBuffer<DB> {
+    fn drop(&mut self) {
+        if let Err(e) = self.flush_buffer() {
+            tracing::warn!("{e}");
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{db::MemoryDB, utils::rand::forest_rng};
+    use fvm_ipld_encoding::DAG_CBOR;
+    use multihash_codetable::Code::Blake2b256;
+    use multihash_codetable::MultihashDigest as _;
+    use rand::Rng as _;
+    use std::sync::Arc;
+
+    #[test]
+    fn test_buffer_flush() {
+        const BUFFER_SIZE: usize = 10;
+        const N_RECORDS: usize = 15;
+        let mem_db = Arc::new(MemoryDB::default());
+        let buf_db = BlockstoreWithWriteBuffer::new_with_capacity(mem_db.clone(), BUFFER_SIZE);
+        let mut records = Vec::with_capacity(N_RECORDS);
+        for _ in 0..N_RECORDS {
+            let mut record = [0; 1024];
+            forest_rng().fill(&mut record);
+            let key = Cid::new_v1(DAG_CBOR, Blake2b256.digest(record.as_slice()));
+            records.push((key, record));
+        }
+
+        buf_db.put_many_keyed(records.clone()).unwrap();
+
+        for (i, (k, v)) in records.iter().enumerate() {
+            assert!(buf_db.has(k).unwrap());
+            assert_eq!(buf_db.get(k).unwrap().unwrap().as_slice(), v);
+            if i < BUFFER_SIZE {
+                assert!(mem_db.has(k).unwrap());
+                assert_eq!(mem_db.get(k).unwrap().unwrap().as_slice(), v);
+            } else {
+                assert!(!mem_db.has(k).unwrap());
+            }
+        }
+
+        drop(buf_db);
+
+        for (k, v) in records.iter() {
+            assert!(mem_db.has(k).unwrap());
+            assert_eq!(mem_db.get(k).unwrap().unwrap().as_slice(), v);
+        }
+    }
+}

src/db/mod.rs

@@ -1,13 +1,15 @@
 // Copyright 2019-2025 ChainSafe Systems
 // SPDX-License-Identifier: Apache-2.0, MIT
 
+mod blockstore_with_write_buffer;
 pub mod car;
 mod memory;
 pub mod parity_db;
 pub mod parity_db_config;
 
 mod gc;
 pub mod ttl;
+pub use blockstore_with_write_buffer::BlockstoreWithWriteBuffer;
 pub use gc::MarkAndSweep;
 pub use memory::MemoryDB;
 use setting_keys::ETH_MAPPING_UP_TO_DATE_KEY;

src/state_migration/mod.rs

@@ -6,6 +6,7 @@ use std::sync::{
     atomic::{self, AtomicBool},
 };
 
+use crate::db::BlockstoreWithWriteBuffer;
 use crate::networks::{ChainConfig, Height, NetworkChain};
 use crate::shim::clock::ChainEpoch;
 use crate::shim::state_tree::StateRoot;
@@ -92,6 +93,12 @@ pub fn run_state_migrations<DB>(
 where
     DB: Blockstore + Send + Sync,
 {
+    // ~10MB RAM per 10k buffer
+    let db_write_buffer = match std::env::var("FOREST_STATE_MIGRATION_DB_WRITE_BUFFER") {
+        Ok(v) => v.parse().ok(),
+        _ => None,
+    }
+    .unwrap_or(10000);
     let mappings = get_migrations(&chain_config.network);
 
     // Make sure bundle is defined.
@@ -115,7 +122,11 @@ where
         if epoch == chain_config.epoch(height) {
             tracing::info!("Running {height} migration at epoch {epoch}");
             let start_time = std::time::Instant::now();
-            let new_state = migrate(chain_config, db, parent_state, epoch)?;
+            let db = Arc::new(BlockstoreWithWriteBuffer::new_with_capacity(
+                db.clone(),
+                db_write_buffer,
+            ));
+            let new_state = migrate(chain_config, &db, parent_state, epoch)?;
             let elapsed = start_time.elapsed();
             // `new_state_actors` is the Go state migration output, log for comparision
             let new_state_actors = db

src/tool/subcommands/shed_cmd/migration.rs

@@ -7,8 +7,10 @@ use std::time::Instant;
 
 use cid::Cid;
 use clap::Args;
+use fvm_ipld_blockstore::Blockstore;
 use itertools::Itertools;
 
+use crate::db::BlockstoreWithWriteBuffer;
 use crate::utils::db::CborStoreExt;
 use crate::{
     blocks::CachingBlockHeader,
@@ -36,6 +38,9 @@ pub struct MigrateStateCommand {
     /// Filecoin network chain
     #[arg(long, required = true)]
     chain: NetworkChain,
+    /// Size of database write buffer, use 0 to disable write buffer
+    #[arg(long, default_value_t = 10000)]
+    db_write_buffer: usize,
 }
 
 impl MigrateStateCommand {
@@ -45,6 +50,7 @@ impl MigrateStateCommand {
             block_to_look_back,
             db,
             chain,
+            db_write_buffer,
         } = self;
         let db = {
             let db = if let Some(db) = db {
@@ -53,7 +59,15 @@ impl MigrateStateCommand {
                 let (_, config) = read_config(None, Some(chain.clone()))?;
                 db_root(&chain_path(&config))?
             };
-            load_db(&db)?
+            let db = load_db(&db)?;
+            Arc::new(if db_write_buffer > 0 {
+                Either::Left(BlockstoreWithWriteBuffer::new_with_capacity(
+                    db,
+                    db_write_buffer,
+                ))
+            } else {
+                Either::Right(db)
+            })
         };
         let block: CachingBlockHeader = db.get_cbor_required(&block_to_look_back)?;
         let chain_config = Arc::new(ChainConfig::from_chain(&chain));
@@ -91,3 +105,72 @@ pub(super) fn load_db(db_root: &Path) -> anyhow::Result<Arc<ManyCar<ParityDb>>>
     load_all_forest_cars(&db, &forest_car_db_dir)?;
     Ok(Arc::new(db))
 }
+
+enum Either<A: Blockstore, B: Blockstore> {
+    Left(A),
+    Right(B),
+}
+
+impl<A: Blockstore, B: Blockstore> Blockstore for Either<A, B> {
+    fn has(&self, k: &Cid) -> anyhow::Result<bool> {
+        match self {
+            Self::Left(v) => v.has(k),
+            Self::Right(v) => v.has(k),
+        }
+    }
+
+    #[allow(clippy::disallowed_types)]
+    fn put<D>(
+        &self,
+        mh_code: multihash_codetable::Code,
+        block: &fvm_ipld_blockstore::Block<D>,
+    ) -> anyhow::Result<Cid>
+    where
+        Self: Sized,
+        D: AsRef<[u8]>,
+    {
+        match self {
+            Self::Left(v) => v.put(mh_code, block),
+            Self::Right(v) => v.put(mh_code, block),
+        }
+    }
+
+    #[allow(clippy::disallowed_types)]
+    fn put_many<D, I>(&self, blocks: I) -> anyhow::Result<()>
+    where
+        Self: Sized,
+        D: AsRef<[u8]>,
+        I: IntoIterator<Item = (multihash_codetable::Code, fvm_ipld_blockstore::Block<D>)>,
+    {
+        match self {
+            Self::Left(v) => v.put_many(blocks),
+            Self::Right(v) => v.put_many(blocks),
+        }
+    }
+
+    fn put_many_keyed<D, I>(&self, blocks: I) -> anyhow::Result<()>
+    where
+        Self: Sized,
+        D: AsRef<[u8]>,
+        I: IntoIterator<Item = (Cid, D)>,
+    {
+        match self {
+            Self::Left(v) => v.put_many_keyed(blocks),
+            Self::Right(v) => v.put_many_keyed(blocks),
+        }
+    }
+
+    fn get(&self, k: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
+        match self {
+            Self::Left(v) => v.get(k),
+            Self::Right(v) => v.get(k),
+        }
+    }
+
+    fn put_keyed(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
+        match self {
+            Self::Left(v) => v.put_keyed(k, block),
+            Self::Right(v) => v.put_keyed(k, block),
+        }
+    }
+}