fix: don't GC manifests

CHANGELOG.md

@@ -38,6 +38,8 @@
 
 ### Fixed
 
+- [#4959](https://github.com/ChainSafe/forest/pull/4959) Re-enable garbage
+  collection after implementing a "persistent" storage for manifests.
 - [#4988](https://github.com/ChainSafe/forest/pull/4988) Fix the `logs` member
   in `EthTxReceipt` that was initialized with a default value.
 

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "forest-filecoin"
-version = "0.22.0"
+version = "0.22.1"
 authors = ["ChainSafe Systems <[email protected]>"]
 repository = "https://github.com/ChainSafe/forest"
 edition = "2021"

src/daemon/bundle.rs

@@ -1,29 +1,27 @@
 // Copyright 2019-2024 ChainSafe Systems
 // SPDX-License-Identifier: Apache-2.0, MIT
 
+use crate::db::PersistentStore;
 use crate::{
     networks::{ActorBundleInfo, NetworkChain, ACTOR_BUNDLES},
     utils::{
-        db::{
-            car_stream::{CarBlock, CarStream},
-            car_util::load_car,
-        },
+        db::car_stream::{CarBlock, CarStream},
         net::http_get,
     },
 };
 use ahash::HashSet;
 use anyhow::ensure;
 use cid::Cid;
 use futures::{stream::FuturesUnordered, TryStreamExt};
-use fvm_ipld_blockstore::Blockstore;
 use std::mem::discriminant;
 use std::{io::Cursor, path::Path};
+use tokio::io::BufReader;
 use tracing::{info, warn};
 
 /// Tries to load the missing actor bundles to the blockstore. If the bundle is
 /// not present, it will be downloaded.
 pub async fn load_actor_bundles(
-    db: &impl Blockstore,
+    db: &impl PersistentStore,
     network: &NetworkChain,
 ) -> anyhow::Result<()> {
     if let Some(bundle_path) = match std::env::var("FOREST_ACTOR_BUNDLE_PATH") {
@@ -40,7 +38,7 @@ pub async fn load_actor_bundles(
 }
 
 pub async fn load_actor_bundles_from_path(
-    db: &impl Blockstore,
+    db: &impl PersistentStore,
     network: &NetworkChain,
     bundle_path: impl AsRef<Path>,
 ) -> anyhow::Result<()> {
@@ -70,15 +68,15 @@ pub async fn load_actor_bundles_from_path(
 
     // Load into DB
     while let Some(CarBlock { cid, data }) = car_stream.try_next().await? {
-        db.put_keyed(&cid, &data)?;
+        db.put_keyed_persistent(&cid, &data)?;
     }
 
     Ok(())
 }
 
 /// Loads the missing actor bundle, returns the CIDs of the loaded bundles.
 pub async fn load_actor_bundles_from_server(
-    db: &impl Blockstore,
+    db: &impl PersistentStore,
     network: &NetworkChain,
     bundles: &[ActorBundleInfo],
 ) -> anyhow::Result<Vec<Cid>> {
@@ -106,7 +104,12 @@ pub async fn load_actor_bundles_from_server(
                         http_get(alt_url).await?
                     };
                     let bytes = response.bytes().await?;
-                    let header = load_car(db, Cursor::new(bytes)).await?;
+
+                    let mut stream = CarStream::new(BufReader::new(Cursor::new(bytes))).await?;
+                    while let Some(block) = stream.try_next().await? {
+                        db.put_keyed_persistent(&block.cid, &block.data)?;
+                    }
+                    let header = stream.header;
                     ensure!(header.roots.len() == 1);
                     ensure!(header.roots.first() == root);
                     Ok(*header.roots.first())

src/daemon/mod.rs

@@ -246,11 +246,7 @@ pub(super) async fn start(
         genesis_header.clone(),
     )?);
 
-    // Network Upgrade manifests are stored in the blockstore but may not be
-    // garbage collected. Until this is fixed, the GC has to be disabled.
-    // Tracking issue: https://github.com/ChainSafe/forest/issues/4926
-    // if !opts.no_gc {
-    if false {
+    if !opts.no_gc {
         let mut db_garbage_collector = {
             let chain_store = chain_store.clone();
             let depth = cmp::max(

src/db/car/any.rs

@@ -10,6 +10,7 @@
 
 use super::{CacheKey, RandomAccessFileReader, ZstdFrameCache};
 use crate::blocks::Tipset;
+use crate::db::PersistentStore;
 use crate::utils::io::EitherMmapOrRandomAccessFile;
 use cid::Cid;
 use fvm_ipld_blockstore::Blockstore;
@@ -124,6 +125,19 @@ where
     }
 }
 
+impl<ReaderT> PersistentStore for AnyCar<ReaderT>
+where
+    ReaderT: ReadAt,
+{
+    fn put_keyed_persistent(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
+        match self {
+            AnyCar::Forest(forest) => forest.put_keyed_persistent(k, block),
+            AnyCar::Plain(plain) => plain.put_keyed_persistent(k, block),
+            AnyCar::Memory(mem) => mem.put_keyed_persistent(k, block),
+        }
+    }
+}
+
 impl<ReaderT> From<super::ForestCar<ReaderT>> for AnyCar<ReaderT> {
     fn from(car: super::ForestCar<ReaderT>) -> Self {
         Self::Forest(car)

src/db/car/forest.rs

@@ -50,6 +50,7 @@ use super::{CacheKey, ZstdFrameCache};
 use crate::blocks::{Tipset, TipsetKey};
 use crate::db::car::plain::write_skip_frame_header_async;
 use crate::db::car::RandomAccessFileReader;
+use crate::db::PersistentStore;
 use crate::utils::db::car_stream::{CarBlock, CarHeader};
 use crate::utils::encoding::from_slice_with_fallback;
 use crate::utils::io::EitherMmapOrRandomAccessFile;
@@ -73,6 +74,7 @@ use std::{
 use tokio::io::{AsyncWrite, AsyncWriteExt};
 use tokio_util::codec::{Decoder, Encoder as _};
 use unsigned_varint::codec::UviBytes;
+
 #[cfg(feature = "benchmark-private")]
 pub mod index;
 #[cfg(not(feature = "benchmark-private"))]
@@ -237,6 +239,15 @@ where
     }
 }
 
+impl<ReaderT> PersistentStore for ForestCar<ReaderT>
+where
+    ReaderT: ReadAt,
+{
+    fn put_keyed_persistent(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
+        self.put_keyed(k, block)
+    }
+}
+
 fn decode_zstd_single_frame<ReaderT: Read>(reader: ReaderT) -> io::Result<BytesMut> {
     let mut zstd_frame = vec![];
     zstd::Decoder::new(reader)?

src/db/car/many.rs

@@ -9,7 +9,7 @@
 //! A single z-frame cache is shared between all read-only stores.
 
 use super::{AnyCar, ZstdFrameCache};
-use crate::db::{EthMappingsStore, MemoryDB, SettingsStore};
+use crate::db::{EthMappingsStore, MemoryDB, PersistentStore, SettingsStore};
 use crate::libp2p_bitswap::BitswapStoreReadWrite;
 use crate::rpc::eth::types::EthHash;
 use crate::shim::clock::ChainEpoch;
@@ -169,6 +169,12 @@ impl<WriterT: Blockstore> Blockstore for ManyCar<WriterT> {
     }
 }
 
+impl<WriterT: PersistentStore> PersistentStore for ManyCar<WriterT> {
+    fn put_keyed_persistent(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
+        self.writer.put_keyed_persistent(k, block)
+    }
+}
+
 impl<WriterT: BitswapStoreRead + Blockstore> BitswapStoreRead for ManyCar<WriterT> {
     fn contains(&self, cid: &Cid) -> anyhow::Result<bool> {
         Blockstore::has(self, cid)

src/db/car/plain.rs

@@ -71,6 +71,7 @@ use cid::Cid;
 use fvm_ipld_blockstore::Blockstore;
 use integer_encoding::VarIntReader;
 
+use crate::db::PersistentStore;
 use nunny::Vec as NonEmpty;
 use parking_lot::RwLock;
 use positioned_io::ReadAt;
@@ -233,6 +234,15 @@ where
     }
 }
 
+impl<ReaderT> PersistentStore for PlainCar<ReaderT>
+where
+    ReaderT: ReadAt,
+{
+    fn put_keyed_persistent(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
+        self.put_keyed(k, block)
+    }
+}
+
 pub async fn write_skip_frame_header_async(
     mut writer: impl AsyncWrite + Unpin,
     data_len: u32,

src/db/gc/mod.rs

@@ -238,7 +238,7 @@ mod test {
     use crate::blocks::{CachingBlockHeader, Tipset};
     use crate::chain::{ChainEpochDelta, ChainStore};
 
-    use crate::db::{GarbageCollectable, MarkAndSweep, MemoryDB};
+    use crate::db::{GarbageCollectable, MarkAndSweep, MemoryDB, PersistentStore};
     use crate::message_pool::test_provider::{mock_block, mock_block_with_parents};
     use crate::networks::ChainConfig;
 
@@ -247,7 +247,11 @@ mod test {
     use core::time::Duration;
 
     use crate::shim::clock::ChainEpoch;
+    use cid::multihash::Code::Identity;
+    use cid::multihash::MultihashDigest;
+    use cid::Cid;
     use fvm_ipld_blockstore::Blockstore;
+    use fvm_ipld_encoding::DAG_CBOR;
     use std::sync::Arc;
 
     const ZERO_DURATION: Duration = Duration::from_secs(0);
@@ -473,4 +477,43 @@ mod test {
             current_epoch + 1 + depth * 2
         );
     }
+
+    #[tokio::test]
+    async fn persistent_data_resilient_to_gc() {
+        let depth = 5 as ChainEpochDelta;
+        let current_epoch = 0 as ChainEpochDelta;
+
+        let tester = GCTester::new();
+        let mut gc = MarkAndSweep::new(
+            tester.db.clone(),
+            tester.get_heaviest_tipset_fn(),
+            depth,
+            ZERO_DURATION,
+        );
+
+        let depth = depth as ChainEpochDelta;
+        let current_epoch = current_epoch as ChainEpochDelta;
+
+        let persistent_data = [1, 55];
+        let persistent_cid = Cid::new_v1(DAG_CBOR, Identity.digest(&persistent_data));
+
+        // Make sure we run enough epochs to initiate GC.
+        tester.run_epochs(current_epoch);
+        tester.run_epochs(depth);
+        tester
+            .db
+            .put_keyed_persistent(&persistent_cid, &persistent_data)
+            .unwrap();
+        // Mark.
+        gc.gc_workflow(ZERO_DURATION).await.unwrap();
+        tester.run_epochs(depth);
+        // Sweep.
+        gc.gc_workflow(ZERO_DURATION).await.unwrap();
+
+        // Make sure persistent data stays.
+        assert_eq!(
+            tester.db.get(&persistent_cid).unwrap(),
+            Some(persistent_data.to_vec())
+        );
+    }
 }

src/db/memory.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0, MIT
 
 use crate::cid_collections::CidHashSet;
-use crate::db::GarbageCollectable;
+use crate::db::{GarbageCollectable, PersistentStore};
 use crate::libp2p_bitswap::{BitswapStoreRead, BitswapStoreReadWrite};
 use crate::rpc::eth::types::EthHash;
 use ahash::HashMap;
@@ -16,6 +16,7 @@ use super::{EthMappingsStore, SettingsStore};
 #[derive(Debug, Default)]
 pub struct MemoryDB {
     blockchain_db: RwLock<HashMap<Vec<u8>, Vec<u8>>>,
+    blockchain_persistent_db: RwLock<HashMap<Vec<u8>, Vec<u8>>>,
     settings_db: RwLock<HashMap<String, Vec<u8>>>,
     eth_mappings_db: RwLock<HashMap<EthHash, Vec<u8>>>,
 }
@@ -109,7 +110,16 @@ impl EthMappingsStore for MemoryDB {
 
 impl Blockstore for MemoryDB {
     fn get(&self, k: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
-        Ok(self.blockchain_db.read().get(&k.to_bytes()).cloned())
+        Ok(self
+            .blockchain_db
+            .read()
+            .get(&k.to_bytes())
+            .cloned()
+            .or(self
+                .blockchain_persistent_db
+                .read()
+                .get(&k.to_bytes())
+                .cloned()))
     }
 
     fn put_keyed(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
@@ -120,6 +130,15 @@ impl Blockstore for MemoryDB {
     }
 }
 
+impl PersistentStore for MemoryDB {
+    fn put_keyed_persistent(&self, k: &Cid, block: &[u8]) -> anyhow::Result<()> {
+        self.blockchain_persistent_db
+            .write()
+            .insert(k.to_bytes(), block.to_vec());
+        Ok(())
+    }
+}
+
 impl BitswapStoreRead for MemoryDB {
     fn contains(&self, cid: &Cid) -> anyhow::Result<bool> {
         Ok(self.blockchain_db.read().contains_key(&cid.to_bytes()))

src/db/migration/migration_map.rs

@@ -9,6 +9,7 @@ use std::{
 
 use crate::db::migration::v0_16_0::Migration0_15_2_0_16_0;
 use crate::db::migration::v0_19_0::Migration0_18_0_0_19_0;
+use crate::db::migration::v0_22_1::Migration0_22_0_0_22_1;
 use crate::Config;
 use anyhow::bail;
 use anyhow::Context as _;
@@ -80,6 +81,7 @@ create_migrations!(
     "0.12.1" -> "0.13.0" @ Migration0_12_1_0_13_0,
     "0.15.2" -> "0.16.0" @ Migration0_15_2_0_16_0,
     "0.18.0" -> "0.19.0" @ Migration0_18_0_0_19_0,
+    "0.22.0" -> "0.22.1" @ Migration0_22_0_0_22_1,
 );
 
 pub struct Migration {

src/db/migration/mod.rs

@@ -6,6 +6,7 @@ mod migration_map;
 mod v0_12_1;
 mod v0_16_0;
 mod v0_19_0;
+mod v0_22_1;
 mod void_migration;
 
 pub use db_migration::DbMigration;

src/db/migration/v0_16_0.rs

@@ -210,7 +210,7 @@ mod paritydb_0_15_1 {
         // if it changes and then this migration should either be maintained or removed.
         pub(super) fn open(path: impl Into<PathBuf>) -> anyhow::Result<db::parity_db::ParityDb> {
             let opts = Self::to_options(path.into());
-            let db = db::parity_db::ParityDb::wrap(Db::open_or_create(&opts)?, false);
+            let db = db::parity_db::ParityDb::wrap(Db::open_or_create(&opts)?, false, false);
             Ok(db)
         }
     }