feat: snapshot GC

Author: hanabi1224

f3-sidecar/ec.go

@@ -17,14 +17,14 @@ type ForestEC struct {
 	closer        jsonrpc.ClientCloser
 }
 
-func NewForestEC(rpcEndpoint, jwt string) (ForestEC, error) {
+func NewForestEC(ctx context.Context, rpcEndpoint, jwt string) (ForestEC, error) {
 	f3api := F3Api{}
 	headers := make(http.Header)
 	isJwtProvided := len(jwt) > 0
 	if isJwtProvided {
 		headers.Add("Authorization", fmt.Sprintf("Bearer %s", jwt))
 	}
-	closer, err := jsonrpc.NewClient(context.Background(), rpcEndpoint, "F3", &f3api, headers)
+	closer, err := jsonrpc.NewClient(ctx, rpcEndpoint, "F3", &f3api, headers)
 	if err != nil {
 		return ForestEC{}, err
 	}

f3-sidecar/ec_test.go

@@ -16,7 +16,7 @@ var (
 )
 
 func init() {
-	ec, err := NewForestEC("http://127.0.0.1:2345/rpc/v1", "")
+	ec, err := NewForestEC(ctx, "http://127.0.0.1:2345/rpc/v1", "")
 	if err != nil {
 		panic(err)
 	}

f3-sidecar/run.go

@@ -21,7 +21,7 @@ import (
 func run(ctx context.Context, rpcEndpoint string, jwt string, f3RpcEndpoint string, initialPowerTable string, bootstrapEpoch int64, finality int64, f3Root string, contract_manifest_poll_interval_seconds uint64) error {
 	api := FilecoinApi{}
 	isJwtProvided := len(jwt) > 0
-	closer, err := jsonrpc.NewClient(context.Background(), rpcEndpoint, "Filecoin", &api, nil)
+	closer, err := jsonrpc.NewClient(ctx, rpcEndpoint, "Filecoin", &api, nil)
 	if err != nil {
 		return err
 	}
@@ -46,7 +46,7 @@ func run(ctx context.Context, rpcEndpoint string, jwt string, f3RpcEndpoint stri
 	if err != nil {
 		return err
 	}
-	ec, err := NewForestEC(rpcEndpoint, jwt)
+	ec, err := NewForestEC(ctx, rpcEndpoint, jwt)
 	if err != nil {
 		return err
 	}

src/chain/mod.rs

@@ -2,9 +2,10 @@
 // SPDX-License-Identifier: Apache-2.0, MIT
 pub mod store;
 mod weight;
-use crate::blocks::Tipset;
+use crate::blocks::{Tipset, TipsetKey};
 use crate::cid_collections::CidHashSet;
 use crate::db::car::forest;
+use crate::db::{SettingsStore, SettingsStoreExt};
 use crate::ipld::stream_chain;
 use crate::utils::io::{AsyncWriterWithChecksum, Checksum};
 use crate::utils::stream::par_buffer;
@@ -16,6 +17,20 @@ use tokio::io::{AsyncWrite, AsyncWriteExt, BufWriter};
 
 pub use self::{store::*, weight::*};
 
+pub async fn export_from_head<D: Digest>(
+    db: Arc<impl Blockstore + SettingsStore + Send + Sync + 'static>,
+    lookup_depth: ChainEpochDelta,
+    writer: impl AsyncWrite + Unpin,
+    seen: CidHashSet,
+    skip_checksum: bool,
+) -> anyhow::Result<(Tipset, Option<digest::Output<D>>), Error> {
+    let head_key = SettingsStoreExt::read_obj::<TipsetKey>(&db, crate::db::setting_keys::HEAD_KEY)?
+        .context("chain head key not found")?;
+    let head_ts = Tipset::load(&db, &head_key)?.context("chain head not found")?;
+    let digest = export::<D>(db, &head_ts, lookup_depth, writer, seen, skip_checksum).await?;
+    Ok((head_ts, digest))
+}
+
 pub async fn export<D: Digest>(
     db: Arc<impl Blockstore + Send + Sync + 'static>,
     tipset: &Tipset,

src/cli/subcommands/chain_cmd.rs

@@ -1,6 +1,9 @@
 // Copyright 2019-2025 ChainSafe Systems
 // SPDX-License-Identifier: Apache-2.0, MIT
 
+mod prune;
+use prune::ChainPruneCommands;
+
 use crate::blocks::{Tipset, TipsetKey};
 use crate::lotus_json::HasLotusJson;
 use crate::message::ChainMessage;
@@ -68,6 +71,8 @@ pub enum ChainCommands {
         #[arg(short, long, aliases = ["yes", "no-confirm"], short_alias = 'y')]
         force: bool,
     },
+    #[command(subcommand)]
+    Prune(ChainPruneCommands),
 }
 
 impl ChainCommands {
@@ -123,6 +128,7 @@ impl ChainCommands {
                 .await?;
                 Ok(())
             }
+            Self::Prune(cmd) => cmd.run(client).await,
         }
     }
 }

src/cli/subcommands/chain_cmd/prune.rs

@@ -0,0 +1,26 @@
+// Copyright 2019-2025 ChainSafe Systems
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+use crate::rpc::{self, RpcMethodExt, chain::ChainPruneSnapshot};
+use clap::Subcommand;
+use std::time::Duration;
+
+#[derive(Debug, Subcommand)]
+pub enum ChainPruneCommands {
+    /// Run snapshot GC
+    Snap,
+}
+
+impl ChainPruneCommands {
+    pub async fn run(self, client: rpc::Client) -> anyhow::Result<()> {
+        match self {
+            Self::Snap => {
+                client
+                    .call(ChainPruneSnapshot::request(())?.with_timeout(Duration::MAX))
+                    .await?;
+            }
+        }
+
+        Ok(())
+    }
+}

src/daemon/mod.rs

@@ -21,7 +21,8 @@ use crate::daemon::db_util::{
 };
 use crate::db::SettingsStore;
 use crate::db::car::ManyCar;
-use crate::db::{MarkAndSweep, MemoryDB, SettingsExt, ttl::EthMappingCollector};
+use crate::db::gc::{MarkAndSweep, SnapshotGarbageCollector};
+use crate::db::{MemoryDB, SettingsExt, ttl::EthMappingCollector};
 use crate::libp2p::{Libp2pService, PeerManager};
 use crate::message_pool::{MessagePool, MpoolConfig, MpoolRpcProvider};
 use crate::networks::{self, ChainConfig};
@@ -44,6 +45,7 @@ use once_cell::sync::Lazy;
 use raw_sync_2::events::{Event, EventInit as _, EventState};
 use shared_memory::ShmemConf;
 use std::path::Path;
+use std::sync::OnceLock;
 use std::time::Duration;
 use std::{cmp, sync::Arc};
 use tempfile::{Builder, TempPath};
@@ -66,6 +68,8 @@ static IPC_PATH: Lazy<TempPath> = Lazy::new(|| {
         .into_temp_path()
 });
 
+pub static GLOBAL_SNAPSHOT_GC: OnceLock<Arc<SnapshotGarbageCollector<DbType>>> = OnceLock::new();
+
 // The parent process and the daemonized child communicate through an Event in
 // shared memory. The identity of the shared memory object is written to a
 // temporary file. The parent process is responsible for cleaning up the file
@@ -467,12 +471,12 @@ fn maybe_start_rpc_service(
     Ok(())
 }
 
-fn maybe_start_f3_service(
-    services: &mut JoinSet<anyhow::Result<()>>,
-    opts: &CliOpts,
-    config: &Config,
-    ctx: &AppContext,
-) {
+fn maybe_start_f3_service(opts: &CliOpts, config: &Config, ctx: &AppContext) {
+    // already running
+    if crate::rpc::f3::F3_LEASE_MANAGER.get().is_some() {
+        return;
+    }
+
     if !config.client.enable_rpc {
         if crate::f3::is_sidecar_ffi_enabled(ctx.state_manager.chain_config()) {
             tracing::warn!("F3 sidecar is enabled but not run because RPC is disabled. ")
@@ -485,7 +489,7 @@ fn maybe_start_f3_service(
         let state_manager = &ctx.state_manager;
         let p2p_peer_id = ctx.p2p_peer_id;
         let admin_jwt = ctx.admin_jwt.clone();
-        services.spawn_blocking({
+        tokio::task::spawn_blocking({
             crate::rpc::f3::F3_LEASE_MANAGER
                 .set(crate::rpc::f3::F3LeaseManager::new(
                     state_manager.chain_config().network.clone(),
@@ -516,7 +520,6 @@ fn maybe_start_f3_service(
                     std::env::var("FOREST_F3_ROOT")
                         .unwrap_or(default_f3_root.display().to_string()),
                 );
-                Ok(())
             }
         });
     }
@@ -595,13 +598,41 @@ fn maybe_start_indexer_service(
 pub(super) async fn start(
     start_time: chrono::DateTime<chrono::Utc>,
     opts: CliOpts,
-    mut config: Config,
+    config: Config,
     shutdown_send: mpsc::Sender<()>,
 ) -> anyhow::Result<()> {
     startup_init(&opts, &config)?;
+    let (snap_gc, snap_gc_reboot_rx) = SnapshotGarbageCollector::new(&config)?;
+    let snap_gc = Arc::new(snap_gc);
+    GLOBAL_SNAPSHOT_GC
+        .set(snap_gc.clone())
+        .ok()
+        .context("failed to set GLOBAL_SNAPSHOT_GC")?;
+    loop {
+        tokio::select! {
+            _ = snap_gc_reboot_rx.recv_async() => {
+                snap_gc.cleanup_before_reboot().await;
+            }
+            result = start_services(start_time, &opts, config.clone(), shutdown_send.clone(), |ctx| {
+                snap_gc.set_db(ctx.db.clone());
+            }) => {
+                break result
+            }
+        }
+    }
+}
+
+pub(super) async fn start_services(
+    start_time: chrono::DateTime<chrono::Utc>,
+    opts: &CliOpts,
+    mut config: Config,
+    shutdown_send: mpsc::Sender<()>,
+    on_app_context_initialized: impl Fn(&AppContext),
+) -> anyhow::Result<()> {
     let mut services = JoinSet::new();
-    maybe_start_track_peak_rss_service(&mut services, &opts);
-    let ctx = AppContext::init(&opts, &config).await?;
+    maybe_start_track_peak_rss_service(&mut services, opts);
+    let ctx = AppContext::init(opts, &config).await?;
+    on_app_context_initialized(&ctx);
     info!(
         "Using network :: {}",
         get_actual_chain_name(&ctx.network_name)
@@ -611,10 +642,8 @@ pub(super) async fn start(
         return Ok(());
     }
     let p2p_service = create_p2p_service(&mut services, &mut config, &ctx).await?;
-
     let mpool = create_mpool(&mut services, &p2p_service, &ctx)?;
-
-    let chain_follower = create_chain_follower(&opts, &p2p_service, mpool.clone(), &ctx)?;
+    let chain_follower = create_chain_follower(opts, &p2p_service, mpool.clone(), &ctx)?;
 
     info!(
         "Starting network:: {}",
@@ -631,20 +660,20 @@ pub(super) async fn start(
         &ctx,
     )?;
 
-    maybe_import_snapshot(&opts, &mut config, &ctx).await?;
+    maybe_import_snapshot(opts, &mut config, &ctx).await?;
     if opts.halt_after_import {
         // Cancel all async services
         services.shutdown().await;
         return Ok(());
     }
     ctx.state_manager.populate_cache();
     maybe_start_metrics_service(&mut services, &config, &ctx).await?;
-    maybe_start_gc_service(&mut services, &opts, &config, &ctx);
-    maybe_start_f3_service(&mut services, &opts, &config, &ctx);
+    maybe_start_gc_service(&mut services, opts, &config, &ctx);
+    maybe_start_f3_service(opts, &config, &ctx);
     maybe_start_health_check_service(&mut services, &config, &p2p_service, &chain_follower, &ctx)
         .await?;
-    maybe_populate_eth_mappings_in_background(&mut services, &opts, config.clone(), &ctx);
-    maybe_start_indexer_service(&mut services, &opts, &config, &ctx);
+    maybe_populate_eth_mappings_in_background(&mut services, opts, config.clone(), &ctx);
+    maybe_start_indexer_service(&mut services, opts, &config, &ctx);
     if !opts.stateless {
         ensure_proof_params_downloaded().await?;
     }

src/db/gc/mod.rs

@@ -69,6 +69,9 @@
 //! depth-first search algorithm, with `O(V+E)` complexity, where V is the number of vertices and E
 //! is the number of edges.
 
+mod snapshot;
+pub use snapshot::SnapshotGarbageCollector;
+
 use crate::blocks::Tipset;
 use crate::chain::ChainEpochDelta;
 
@@ -236,7 +239,7 @@ impl<DB: Blockstore + SettingsStore + GarbageCollectable<CidHashSet> + Sync + Se
 mod test {
     use crate::blocks::{CachingBlockHeader, Tipset};
     use crate::chain::{ChainEpochDelta, ChainStore};
-    use crate::db::{GarbageCollectable, MarkAndSweep, MemoryDB, PersistentStore};
+    use crate::db::{GarbageCollectable, MemoryDB, PersistentStore, gc::MarkAndSweep};
     use crate::message_pool::test_provider::{mock_block, mock_block_with_parents};
     use crate::networks::ChainConfig;
     use crate::shim::clock::ChainEpoch;