feat: database garbage collection

CHANGELOG.md

@@ -21,6 +21,8 @@ Notable updates:
 - [forest daemon] Support for NV18.
   [#2558](https://github.com/ChainSafe/forest/pull/2558)
   [#2579](https://github.com/ChainSafe/forest/pull/2579)
+- [forest daemon] Automatic database garbage collection.
+  [#2638](https://github.com/ChainSafe/forest/pull/2638)
 
 ### Changed
 

Cargo.toml

@@ -61,7 +61,7 @@ bls-signatures = { version = "0.12", default-features = false, features = ["blst
 byteorder = "1.4.3"
 bytes = "1.2"
 cfg-if = "1"
-chrono = { version = "0.4", default-features = false, features = [] }
+chrono = { version = "0.4", default-features = false, features = ["clock"] }
 cid = { version = "0.8", default-features = false, features = ["std"] }
 clap = { version = "4.0", features = ["derive"] }
 console-subscriber = { version = "0.1", features = ["parking_lot"] }
@@ -125,6 +125,7 @@ serde_ipld_dagcbor = "0.2"
 serde_json = "1.0"
 serde_repr = "0.1.8"
 serde_with = { version = "2.0.1", features = ["chrono_0_4"] }
+serde_yaml = "0.9"
 sha2 = { version = "0.10.5", default-features = false }
 tempfile = "3.4"
 thiserror = "1.0"

blockchain/chain/Cargo.toml

@@ -27,7 +27,6 @@ forest_metrics.workspace = true
 forest_networks.workspace = true
 forest_shim.workspace = true
 forest_utils.workspace = true
-futures.workspace = true
 fvm_ipld_amt.workspace = true
 fvm_ipld_blockstore.workspace = true
 fvm_ipld_car.workspace = true

blockchain/chain/src/store/chain_store.rs

@@ -1,29 +1,25 @@
 // Copyright 2019-2023 ChainSafe Systems
 // SPDX-License-Identifier: Apache-2.0, MIT
 
-use std::{collections::VecDeque, num::NonZeroUsize, path::Path, sync::Arc, time::SystemTime};
+use std::{num::NonZeroUsize, path::Path, sync::Arc, time::SystemTime};
 
 use ahash::{HashMap, HashMapExt, HashSet};
 use anyhow::Result;
 use async_stream::stream;
 use bls_signatures::Serialize as SerializeBls;
-use cid::{
-    multihash::{Code, Code::Blake2b256},
-    Cid,
-};
+use cid::{multihash::Code::Blake2b256, Cid};
 use digest::Digest;
 use forest_beacon::{BeaconEntry, IGNORE_DRAND_VAR};
 use forest_blocks::{Block, BlockHeader, FullTipset, Tipset, TipsetKeys, TxMeta};
 use forest_encoding::de::DeserializeOwned;
 use forest_interpreter::BlockMessages;
-use forest_ipld::{recurse_links_hash, CidHashSet};
+use forest_ipld::{should_save_block_to_snapshot, walk_snapshot};
 use forest_libp2p_bitswap::{BitswapStoreRead, BitswapStoreReadWrite};
 use forest_message::{ChainMessage, Message as MessageTrait, SignedMessage};
 use forest_metrics::metrics;
 use forest_networks::ChainConfig;
 use forest_shim::{
     address::Address,
-    clock::EPOCHS_IN_DAY,
     crypto::{Signature, SignatureType},
     econ::TokenAmount,
     executor::Receipt,
@@ -37,7 +33,6 @@ use forest_utils::{
     },
     io::Checksum,
 };
-use futures::Future;
 use fvm_ipld_amt::Amtv0 as Amt;
 use fvm_ipld_blockstore::Blockstore;
 use fvm_ipld_car::CarHeader;
@@ -563,23 +558,18 @@ where
 
         // Walks over tipset and historical data, sending all blocks visited into the
         // car writer.
-        Self::walk_snapshot(tipset, recent_roots, |cid| {
+        walk_snapshot(tipset, recent_roots, |cid| {
             let tx_clone = tx.clone();
             async move {
                 let block = self
                     .blockstore()
                     .get(&cid)?
                     .ok_or_else(|| Error::Other(format!("Cid {cid} not found in blockstore")))?;
 
-                // Don't include identity CIDs.
-                // We only include raw and dagcbor, for now.
-                // Raw for "code" CIDs.
-                if u64::from(Code::Identity) != cid.hash().code()
-                    && (cid.codec() == fvm_shared::IPLD_RAW
-                        || cid.codec() == fvm_ipld_encoding::DAG_CBOR)
-                {
+                if should_save_block_to_snapshot(&cid) {
                     tx_clone.send_async((cid, block.clone())).await?;
                 }
+
                 Ok(block)
             }
         })
@@ -605,60 +595,6 @@ where
         let digest = writer.lock().await.get_mut().finalize();
         Ok(digest)
     }
-
-    /// Walks over tipset and state data and loads all blocks not yet seen.
-    /// This is tracked based on the callback function loading blocks.
-    pub async fn walk_snapshot<F, T>(
-        tipset: &Tipset,
-        recent_roots: ChainEpoch,
-        mut load_block: F,
-    ) -> Result<(), Error>
-    where
-        F: FnMut(Cid) -> T + Send,
-        T: Future<Output = Result<Vec<u8>, anyhow::Error>> + Send,
-    {
-        let mut seen = CidHashSet::default();
-        let mut blocks_to_walk: VecDeque<Cid> = tipset.cids().to_vec().into();
-        let mut current_min_height = tipset.epoch();
-        let incl_roots_epoch = tipset.epoch() - recent_roots;
-
-        while let Some(next) = blocks_to_walk.pop_front() {
-            if !seen.insert(&next) {
-                continue;
-            }
-
-            let data = load_block(next).await?;
-
-            let h = BlockHeader::unmarshal_cbor(&data)?;
-
-            if current_min_height > h.epoch() {
-                current_min_height = h.epoch();
-                if current_min_height % EPOCHS_IN_DAY == 0 {
-                    info!(target: "chain_api", "export at: {}", current_min_height);
-                }
-            }
-
-            if h.epoch() > incl_roots_epoch {
-                recurse_links_hash(&mut seen, *h.messages(), &mut load_block).await?;
-            }
-
-            if h.epoch() > 0 {
-                for p in h.parents().cids() {
-                    blocks_to_walk.push_back(*p);
-                }
-            } else {
-                for p in h.parents().cids() {
-                    load_block(*p).await?;
-                }
-            }
-
-            if h.epoch() == 0 || h.epoch() > incl_roots_epoch {
-                recurse_links_hash(&mut seen, *h.state_root(), &mut load_block).await?;
-            }
-        }
-
-        Ok(())
-    }
 }
 
 pub(crate) type TipsetCache = Mutex<LruCache<TipsetKeys, Arc<Tipset>>>;

forest/cli/Cargo.toml

@@ -13,6 +13,7 @@ anyhow.workspace = true
 atty = "0.2"
 base64.workspace = true
 boa_engine = { version = "0.16.0", features = ["console"] }
+chrono.workspace = true
 cid.workspace = true
 clap.workspace = true
 convert_case = "0.6.0"

forest/cli/src/cli/db_cmd.rs

@@ -1,17 +1,21 @@
 // Copyright 2019-2023 ChainSafe Systems
 // SPDX-License-Identifier: Apache-2.0, MIT
 
+use chrono::Utc;
 use clap::Subcommand;
 use forest_cli_shared::{chain_path, cli::Config};
-use forest_db::db_engine::db_path;
+use forest_db::db_engine::db_root;
+use forest_rpc_client::db_ops::db_gc;
 use log::error;
 
-use crate::cli::prompt_confirm;
+use crate::cli::{handle_rpc_err, prompt_confirm};
 
 #[derive(Debug, Subcommand)]
 pub enum DBCommands {
     /// Show DB stats
     Stats,
+    /// Run DB garbage collection
+    GC,
     /// DB Clean up
     Clean {
         /// Answer yes to all forest-cli yes/no questions without prompting
@@ -21,19 +25,33 @@ pub enum DBCommands {
 }
 
 impl DBCommands {
-    pub fn run(&self, config: &Config) -> anyhow::Result<()> {
+    pub async fn run(&self, config: &Config) -> anyhow::Result<()> {
         match self {
             Self::Stats => {
                 use human_repr::HumanCount;
 
-                let dir = db_path(&chain_path(config));
+                let dir = db_root(&chain_path(config));
                 println!("Database path: {}", dir.display());
                 let size = fs_extra::dir::get_size(dir).unwrap_or_default();
                 println!("Database size: {}", size.human_count_bytes());
                 Ok(())
             }
+            Self::GC => {
+                let start = Utc::now();
+
+                db_gc((), &config.client.rpc_token)
+                    .await
+                    .map_err(handle_rpc_err)?;
+
+                println!(
+                    "DB GC completed. took {}s",
+                    (Utc::now() - start).num_seconds()
+                );
+
+                Ok(())
+            }
             Self::Clean { force } => {
-                let dir = db_path(&chain_path(config));
+                let dir = db_root(&chain_path(config));
                 if !dir.is_dir() {
                     println!(
                         "Aborted. Database path {} is not a valid directory",