Merge pull request #5744 from filecoin-project/feat/splitstore-gc

garbage collect hotstore after compaction

blockstore/badger/blockstore.go

@@ -131,6 +131,21 @@ func (b *Blockstore) Close() error {
 	return b.DB.Close()
 }
 
+// CollectGarbage runs garbage collection on the value log
+func (b *Blockstore) CollectGarbage() error {
+	if atomic.LoadInt64(&b.state) != stateOpen {
+		return ErrBlockstoreClosed
+	}
+
+	err := b.DB.RunValueLogGC(0.125)
+	if err == badger.ErrNoRewrite {
+		// not really an error in this case
+		return nil
+	}
+
+	return err
+}
+
 // View implements blockstore.Viewer, which leverages zero-copy read-only
 // access to values.
 func (b *Blockstore) View(cid cid.Cid, fn func([]byte) error) error {

blockstore/splitstore/splitstore.go

@@ -710,6 +710,8 @@ func (s *SplitStore) compactSimple(curTs *types.TipSet) error {
 		return xerrors.Errorf("error syncing tracker: %w", err)
 	}
 
+	s.gcHotstore()
+
 	err = s.setBaseEpoch(coldEpoch)
 	if err != nil {
 		return xerrors.Errorf("error saving base epoch: %w", err)
@@ -795,6 +797,19 @@ func (s *SplitStore) purgeTracking(cids []cid.Cid) error {
 	return s.purgeBatch(cids, s.tracker.DeleteBatch)
 }
 
+func (s *SplitStore) gcHotstore() {
+	if gc, ok := s.hot.(interface{ CollectGarbage() error }); ok {
+		log.Infof("garbage collecting hotstore")
+		startGC := time.Now()
+		err := gc.CollectGarbage()
+		if err != nil {
+			log.Warnf("error garbage collecting hotstore: %s", err)
+		} else {
+			log.Infow("garbage collection done", "took", time.Since(startGC))
+		}
+	}
+}
+
 func (s *SplitStore) compactFull(curTs *types.TipSet) error {
 	currentEpoch := curTs.Height()
 	coldEpoch := s.baseEpoch + CompactionCold
@@ -1005,6 +1020,8 @@ func (s *SplitStore) compactFull(curTs *types.TipSet) error {
 		return xerrors.Errorf("error syncing tracker: %w", err)
 	}
 
+	s.gcHotstore()
+
 	err = s.setBaseEpoch(coldEpoch)
 	if err != nil {
 		return xerrors.Errorf("error saving base epoch: %w", err)