fix(gc): backfill new db records since snapshot epoch

f3-sidecar/ffi_impl.go

@@ -13,12 +13,14 @@ func init() {
 	logging.SetAllLoggers(logging.LevelInfo)
 	err := logging.SetLogLevel("dht", "error")
 	checkError(err)
-	err = logging.SetLogLevel("dht/RtRefreshManager", "warn")
+	err = logging.SetLogLevel("dht/RtRefreshManager", "error")
 	checkError(err)
 	err = logging.SetLogLevel("net/identify", "error")
 	checkError(err)
 	err = logging.SetLogLevel("pubsub", "warn")
 	checkError(err)
+	err = logging.SetLogLevel("swarm2", "error")
+	checkError(err)
 	err = logging.SetLogLevel("f3/sidecar", "debug")
 	checkError(err)
 	GoF3NodeImpl = &f3Impl{ctx: context.Background()}

src/chain/mod.rs

@@ -18,21 +18,21 @@ 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>,
+    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)?
+    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_required(&db, &head_key)?;
     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>,
+    db: &Arc<impl Blockstore + Send + Sync + 'static>,
     tipset: &Tipset,
     lookup_depth: ChainEpochDelta,
     writer: impl AsyncWrite + Unpin,
@@ -53,8 +53,8 @@ pub async fn export<D: Digest>(
         // block size is between 1kb and 2kb.
         1024,
         stream_chain(
-            Arc::clone(&db),
-            tipset.clone().chain_owned(Arc::clone(&db)),
+            Arc::clone(db),
+            tipset.clone().chain_owned(Arc::clone(db)),
             stateroot_lookup_limit,
         )
         .with_seen(seen),

src/daemon/context.rs

@@ -198,7 +198,7 @@ async fn setup_db(opts: &CliOpts, config: &Config) -> anyhow::Result<(Arc<DbType
     maybe_migrate_db(config);
     let chain_data_path = chain_path(config);
     let db_root_dir = db_root(&chain_data_path)?;
-    let db_writer = Arc::new(open_db(db_root_dir.clone(), config.db_config().clone())?);
+    let db_writer = Arc::new(open_db(db_root_dir.clone(), config.db_config())?);
     let db = Arc::new(ManyCar::new(db_writer.clone()));
     let forest_car_db_dir = db_root_dir.join(CAR_DB_DIR_NAME);
     load_all_forest_cars(&db, &forest_car_db_dir)?;

src/db/car/many.rs

@@ -11,7 +11,8 @@
 use super::{AnyCar, ZstdFrameCache};
 use crate::blocks::TipsetKey;
 use crate::db::{
-    EthMappingsStore, IndicesStore, MemoryDB, PersistentStore, SettingsStore, SettingsStoreExt,
+    BlockstoreWriteOpsSubscribable, EthMappingsStore, IndicesStore, MemoryDB, PersistentStore,
+    SettingsStore, SettingsStoreExt,
 };
 use crate::libp2p_bitswap::BitswapStoreReadWrite;
 use crate::rpc::eth::types::EthHash;
@@ -272,6 +273,16 @@ impl<T: Blockstore + SettingsStore> super::super::HeaviestTipsetKeyProvider for
     }
 }
 
+impl<WriterT: BlockstoreWriteOpsSubscribable> BlockstoreWriteOpsSubscribable for ManyCar<WriterT> {
+    fn subscribe_write_ops(&self) -> tokio::sync::broadcast::Receiver<(Cid, Vec<u8>)> {
+        self.writer().subscribe_write_ops()
+    }
+
+    fn unsubscribe_write_ops(&self) {
+        self.writer().unsubscribe_write_ops()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::super::AnyCar;