feat: remove legacy orphans background

batch.go

@@ -1,6 +1,8 @@
 package iavl
 
 import (
+	"sync"
+
 	dbm "github.com/cosmos/cosmos-db"
 )
 
@@ -11,6 +13,7 @@ type BatchWithFlusher struct {
 	db    dbm.DB    // This is only used to create new batch
 	batch dbm.Batch // Batched writing buffer.
 
+	mtx            sync.Mutex
 	flushThreshold int // The threshold to flush the batch to disk.
 }
 
@@ -46,6 +49,9 @@ func (b *BatchWithFlusher) estimateSizeAfterSetting(key []byte, value []byte) (i
 // the batch is flushed to disk, cleared, and a new one is created with buffer pre-allocated to threshold.
 // The addition entry is then added to the batch.
 func (b *BatchWithFlusher) Set(key, value []byte) error {
+	b.mtx.Lock()
+	defer b.mtx.Unlock()
+
 	batchSizeAfter, err := b.estimateSizeAfterSetting(key, value)
 	if err != nil {
 		return err
@@ -67,6 +73,9 @@ func (b *BatchWithFlusher) Set(key, value []byte) error {
 // the batch is flushed to disk, cleared, and a new one is created with buffer pre-allocated to threshold.
 // The deletion entry is then added to the batch.
 func (b *BatchWithFlusher) Delete(key []byte) error {
+	b.mtx.Lock()
+	defer b.mtx.Unlock()
+
 	batchSizeAfter, err := b.estimateSizeAfterSetting(key, []byte{})
 	if err != nil {
 		return err

nodedb.go

@@ -10,6 +10,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"cosmossdk.io/log"
 	dbm "github.com/cosmos/cosmos-db"
@@ -470,9 +471,38 @@ func (ndb *nodeDB) deleteLegacyVersions() error {
 	ndb.legacyLatestVersion = -1
 
 	// Delete all orphan nodes of the legacy versions
-	return ndb.traversePrefix(legacyOrphanKeyFormat.Key(), func(key, value []byte) error {
-		return ndb.batch.Delete(key)
-	})
+	go func() {
+		if err := ndb.deleteOrphans(); err != nil {
+			ndb.logger.Error("failed to clean legacy orphans", "err", err)
+		}
+	}()
+
+	return nil
+}
+
+// deleteOrphans cleans all legacy orphans from the nodeDB.
+func (ndb *nodeDB) deleteOrphans() error {
+	itr, err := dbm.IteratePrefix(ndb.db, legacyOrphanKeyFormat.Key())
+	if err != nil {
+		return err
+	}
+	defer itr.Close()
+
+	count := 0
+	for ; itr.Valid(); itr.Next() {
+		if err := ndb.batch.Delete(itr.Key()); err != nil {
+			return err
+		}
+
+		// Sleep for a while to avoid blocking the main thread i/o.
+		count++
+		if count > 1000 {
+			count = 0
+			time.Sleep(100 * time.Millisecond)
+		}
+	}
+
+	return nil
 }
 
 // DeleteVersionsFrom permanently deletes all tree versions from the given version upwards.