fix

Author: hanabi1224

scripts/tests/calibnet_other_check.sh

@@ -24,10 +24,11 @@ if [ "$cid" != "bafy2bzacecgqgzh3gxpariy3mzqb37y2vvxoaw5nwbrlzkhso6owus3zqckwe"
   exit 1
 fi
 
-echo "Test snapshot GC"
 forest_check_db_stats
+echo "Run snapshot GC"
 forest_run_snap_gc
 sleep 5
+echo "Wait the node to sync"
 forest_wait_for_sync
 forest_check_db_stats
 

src/daemon/mod.rs

@@ -609,6 +609,10 @@ pub(super) async fn start(
         .set(snap_gc.clone())
         .ok()
         .context("failed to set GLOBAL_SNAPSHOT_GC")?;
+    tokio::task::spawn({
+        let snap_gc = snap_gc.clone();
+        async move { snap_gc.event_toop().await }
+    });
     loop {
         tokio::select! {
             _ = snap_gc_reboot_rx.recv_async() => {

src/db/gc/snapshot.rs

@@ -31,10 +31,13 @@ pub struct SnapshotGarbageCollector<DB> {
     recent_state_roots: i64,
     db_config: DbConfig,
     running: AtomicBool,
-    reboot_tx: flume::Sender<()>,
     exported_chain_head: RwLock<Option<Tipset>>,
     blessed_lite_snapshot: RwLock<Option<PathBuf>>,
     db: RwLock<Option<Arc<DB>>>,
+    reboot_tx: flume::Sender<()>,
+    trigger_tx: flume::Sender<()>,
+    trigger_rx: flume::Receiver<()>,
+    progress_tx: RwLock<Option<flume::Sender<()>>>,
 }
 
 impl<DB> SnapshotGarbageCollector<DB>
@@ -47,46 +50,58 @@ where
         let car_db_dir = db_root_dir.join(CAR_DB_DIR_NAME);
         let recent_state_roots = config.sync.recent_state_roots;
         let (reboot_tx, reboot_rx) = flume::bounded(1);
+        let (trigger_tx, trigger_rx) = flume::bounded(1);
         Ok((
             Self {
                 db_root_dir,
                 car_db_dir,
                 recent_state_roots,
                 db_config: config.db_config().clone(),
                 running: AtomicBool::new(false),
-                reboot_tx,
                 exported_chain_head: RwLock::new(None),
                 blessed_lite_snapshot: RwLock::new(None),
                 db: RwLock::new(None),
+                reboot_tx,
+                trigger_tx,
+                trigger_rx,
+                progress_tx: RwLock::new(None),
             },
             reboot_rx,
         ))
     }
 
-    pub fn running(&self) -> bool {
-        self.running.load(Ordering::Relaxed)
-    }
-
     pub fn set_db(&self, db: Arc<DB>) {
         *self.db.write() = Some(db);
     }
 
-    pub async fn run(&self) -> anyhow::Result<()> {
-        if self.running() {
-            anyhow::bail!("snap gc has already been running");
+    pub async fn event_toop(&self) {
+        while self.trigger_rx.recv_async().await.is_ok() {
+            if self.running.load(Ordering::Relaxed) {
+                tracing::warn!("snap gc has already been running");
+            } else {
+                self.running.store(true, Ordering::Relaxed);
+                if let Err(e) = self.export_snapshot().await {
+                    tracing::warn!("{e}");
+                }
+            }
         }
-        if let Err(e) = self.export_snapshot().await {
-            self.running.store(false, Ordering::Relaxed);
-            Err(e)
-        } else {
-            Ok(())
+    }
+
+    pub fn trigger(&self) -> flume::Receiver<()> {
+        let (progress_tx, progress_rx) = flume::unbounded();
+        *self.progress_tx.write() = Some(progress_tx);
+        if self.trigger_tx.try_send(()).is_err() {
+            tracing::warn!("snap gc has already been triggered");
         }
+        progress_rx
     }
 
     async fn export_snapshot(&self) -> anyhow::Result<()> {
         let db = self.db.read().clone().context("db not yet initialzied")?;
-        self.running.store(true, Ordering::Relaxed);
-        tracing::info!("exporting lite snapshot");
+        tracing::info!(
+            "exporting lite snapshot with {} recent state roots",
+            self.recent_state_roots
+        );
         let temp_path = tempfile::NamedTempFile::new_in(&self.car_db_dir)?.into_temp_path();
         let file = tokio::fs::File::create(&temp_path).await?;
         let (head_ts, _) = crate::chain::export_from_head::<Sha256>(
@@ -97,9 +112,11 @@ where
             true,
         )
         .await?;
-        let target_path = self
-            .car_db_dir
-            .join(format!("lite_{}.forest.car.zst", head_ts.epoch()));
+        let target_path = self.car_db_dir.join(format!(
+            "lite_{}_{}.forest.car.zst",
+            self.recent_state_roots,
+            head_ts.epoch()
+        ));
         temp_path.persist(&target_path)?;
         tracing::info!("exported lite snapshot at {}", target_path.display());
         *self.exported_chain_head.write() = Some(head_ts);
@@ -112,6 +129,7 @@ where
     }
 
     pub async fn cleanup_before_reboot(&self) {
+        drop(self.progress_tx.write().take());
         if let Err(e) = self.cleanup_before_reboot_inner().await {
             tracing::warn!("{e}");
         }

src/rpc/methods/chain.rs

@@ -200,7 +200,7 @@ impl RpcMethod<0> for ChainPruneSnapshot {
     const NAME: &'static str = "Forest.SnapshotGC";
     const PARAM_NAMES: [&'static str; 0] = [];
     const API_PATHS: BitFlags<ApiPaths> = ApiPaths::all();
-    const PERMISSION: Permission = Permission::Read;
+    const PERMISSION: Permission = Permission::Admin;
 
     type Params = ();
     type Ok = ();
@@ -210,7 +210,9 @@ impl RpcMethod<0> for ChainPruneSnapshot {
         (): Self::Params,
     ) -> Result<Self::Ok, ServerError> {
         if let Some(gc) = crate::daemon::GLOBAL_SNAPSHOT_GC.get() {
-            Ok(gc.run().await?)
+            let progress_rx = gc.trigger();
+            while progress_rx.recv_async().await.is_ok() {}
+            Ok(())
         } else {
             Err(anyhow::anyhow!("snapshot gc is not enabled").into())
         }