core/filtermaps: hashdb safe delete range

core/filtermaps/filtermaps.go

@@ -29,7 +29,6 @@ import (
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethdb"
-	"github.com/ethereum/go-ethereum/ethdb/leveldb"
 	"github.com/ethereum/go-ethereum/log"
 )
 
@@ -59,6 +58,7 @@ type FilterMaps struct {
 	closeCh        chan struct{}
 	closeWg        sync.WaitGroup
 	history        uint64
+	hashDbSafe     bool // use hashdb-safe delete range method
 	exportFileName string
 	Params
 
@@ -69,6 +69,7 @@ type FilterMaps struct {
 	// Matcher backend can read them under indexLock read lock.
 	indexLock           sync.RWMutex
 	indexedRange        filterMapsRange
+	cleanedEpochsBefore uint32     // all unindexed data cleaned before this point
 	indexedView         *ChainView // always consistent with the log index
 	hasTempRange        bool
 
@@ -180,6 +181,7 @@ type Config struct {
 	// This option enables the checkpoint JSON file generator.
 	// If set, the given file will be updated with checkpoint information.
 	ExportFileName string
+	HashDbSafe     bool
 }
 
 // NewFilterMaps creates a new FilterMaps and starts the indexer.
@@ -197,6 +199,7 @@ func NewFilterMaps(db ethdb.KeyValueStore, initView *ChainView, historyCutoff, f
 		blockProcessingCh: make(chan bool, 1),
 		history:           config.History,
 		disabled:          config.Disabled,
+		hashDbSafe:        config.HashDbSafe,
 		disabledCh:        make(chan struct{}),
 		exportFileName:    config.ExportFileName,
 		Params:            params,
@@ -208,6 +211,8 @@ func NewFilterMaps(db ethdb.KeyValueStore, initView *ChainView, historyCutoff, f
 			maps:             common.NewRange(rs.MapsFirst, rs.MapsAfterLast-rs.MapsFirst),
 			tailPartialEpoch: rs.TailPartialEpoch,
 		},
+		// deleting last unindexed epoch might have been interrupted by shutdown
+		cleanedEpochsBefore: max(rs.MapsFirst>>params.logMapsPerEpoch, 1) - 1,
 		historyCutoff:       historyCutoff,
 		finalBlock:          finalBlock,
 		matcherSyncCh:       make(chan *FilterMapsMatcherBackend),
@@ -301,14 +306,24 @@ func (f *FilterMaps) reset() {
 	// deleting the range first ensures that resetDb will be called again at next
 	// startup and any leftover data will be removed even if it cannot finish now.
 	rawdb.DeleteFilterMapsRange(f.db)
-	f.safeDeleteRange(rawdb.DeleteFilterMapsDb, "Resetting log index database")
+	f.safeDeleteWithLogs(rawdb.DeleteFilterMapsDb, "Resetting log index database", f.isShuttingDown)
+}
+
+// isShuttingDown returns true if FilterMaps is shutting down.
+func (f *FilterMaps) isShuttingDown() bool {
+	select {
+	case <-f.closeCh:
+		return true
+	default:
+		return false
+	}
 }
 
 // init initializes an empty log index according to the current targetView.
 func (f *FilterMaps) init() error {
 	// ensure that there is no remaining data in the filter maps key range
-	if !f.safeDeleteRange(rawdb.DeleteFilterMapsDb, "Resetting log index database") {
+	if err := f.safeDeleteWithLogs(rawdb.DeleteFilterMapsDb, "Resetting log index database", f.isShuttingDown); err != nil {
-		return errors.New("could not reset log index database")
+		return err
 	}
 
 	f.indexLock.Lock()
@@ -358,38 +373,37 @@ func (f *FilterMaps) init() error {
 
 // removeBloomBits removes old bloom bits data from the database.
 func (f *FilterMaps) removeBloomBits() {
-	f.safeDeleteRange(rawdb.DeleteBloomBitsDb, "Removing old bloom bits database")
+	f.safeDeleteWithLogs(rawdb.DeleteBloomBitsDb, "Removing old bloom bits database", f.isShuttingDown)
 	f.closeWg.Done()
 }
 
-// safeDeleteRange calls the specified database range deleter function
+// safeDeleteWithLogs is a wrapper for a function that performs a safe range
-// repeatedly as long as it returns leveldb.ErrTooManyKeys.
+// delete operation using rawdb.SafeDeleteRange. It emits log messages if the
-// This wrapper is necessary because of the leveldb fallback implementation
+// process takes long enough to call the stop callback.
-// of DeleteRange.
+func (f *FilterMaps) safeDeleteWithLogs(deleteFn func(db ethdb.KeyValueStore, hashDbSafe bool, stopCb func() bool) error, action string, stopCb func() bool) error {
-func (f *FilterMaps) safeDeleteRange(removeFn func(ethdb.KeyValueRangeDeleter) error, action string) bool {
+	var (
-	start := time.Now()
+		start          = time.Now()
-	var retry bool
+		logPrinted     bool
-	for {
+		lastLogPrinted = start
-		err := removeFn(f.db)
+	)
-		if err == nil {
+	switch err := deleteFn(f.db, f.hashDbSafe, func() bool {
-			if retry {
+		if !logPrinted || time.Since(lastLogPrinted) > time.Second*10 {
+			log.Info(action+" in progress...", "elapsed", time.Since(start))
+			logPrinted, lastLogPrinted = true, time.Now()
+		}
+		return stopCb()
+	}); err {
+	case nil:
+		if logPrinted {
 			log.Info(action+" finished", "elapsed", time.Since(start))
 		}
-			return true
+		return nil
-		}
+	case rawdb.ErrDeleteRangeInterrupted:
-		if err != leveldb.ErrTooManyKeys {
+		log.Warn(action+" interrupted", "elapsed", time.Since(start))
-			log.Error(action+" failed", "error", err)
+		return err
-			return false
-		}
-		select {
-		case <-f.closeCh:
-			return false
 	default:
-		}
+		log.Error(action+" failed", "error", err)
-		if !retry {
+		return err
-			log.Info(action+" in progress...", "elapsed", time.Since(start))
-			retry = true
-		}
 	}
 }
 
@@ -658,54 +672,95 @@ func (f *FilterMaps) deleteLastBlockOfMap(batch ethdb.Batch, mapIndex uint32) {
 	rawdb.DeleteFilterMapLastBlock(batch, mapIndex)
 }
 
-// deleteTailEpoch deletes index data from the earliest, either fully or partially
+// deleteTailEpoch deletes index data from the specified epoch. The last block
-// indexed epoch. The last block pointer for the last map of the epoch and the
+// pointer for the last map of the epoch and the corresponding block log value
-// corresponding block log value pointer are retained as these are always assumed
+// pointer are retained as these are always assumed to be available for each
-// to be available for each epoch.
+// epoch as boundary markers.
-func (f *FilterMaps) deleteTailEpoch(epoch uint32) error {
+// The function returns true if all index data related to the epoch (except for
+// the boundary markers) has been fully removed.
+func (f *FilterMaps) deleteTailEpoch(epoch uint32) (bool, error) {
 	f.indexLock.Lock()
 	defer f.indexLock.Unlock()
 
+	// determine epoch boundaries
 	firstMap := epoch << f.logMapsPerEpoch
 	lastBlock, _, err := f.getLastBlockOfMap(firstMap + f.mapsPerEpoch - 1)
 	if err != nil {
-		return fmt.Errorf("failed to retrieve last block of deleted epoch %d: %v", epoch, err)
+		return false, fmt.Errorf("failed to retrieve last block of deleted epoch %d: %v", epoch, err)
 	}
 	var firstBlock uint64
 	if epoch > 0 {
 		firstBlock, _, err = f.getLastBlockOfMap(firstMap - 1)
 		if err != nil {
-			return fmt.Errorf("failed to retrieve last block before deleted epoch %d: %v", epoch, err)
+			return false, fmt.Errorf("failed to retrieve last block before deleted epoch %d: %v", epoch, err)
 		}
 		firstBlock++
 	}
-	fmr := f.indexedRange
+	// update rendered range if necessary
-	if f.indexedRange.maps.First() == firstMap &&
+	var (
-		f.indexedRange.maps.AfterLast() > firstMap+f.mapsPerEpoch &&
+		fmr            = f.indexedRange
-		f.indexedRange.tailPartialEpoch == 0 {
+		firstEpoch     = f.indexedRange.maps.First() >> f.logMapsPerEpoch
-		fmr.maps.SetFirst(firstMap + f.mapsPerEpoch)
+		afterLastEpoch = (f.indexedRange.maps.AfterLast() + f.mapsPerEpoch - 1) >> f.logMapsPerEpoch
-		fmr.blocks.SetFirst(lastBlock + 1)
+	)
-	} else if f.indexedRange.maps.First() == firstMap+f.mapsPerEpoch {
+	if f.indexedRange.tailPartialEpoch != 0 && firstEpoch > 0 {
-		fmr.tailPartialEpoch = 0
+		firstEpoch--
-	} else {
-		return errors.New("invalid tail epoch number")
 	}
+	switch {
+	case epoch < firstEpoch:
+	// cleanup of already unindexed epoch; range not affected
+	case epoch == firstEpoch && epoch+1 < afterLastEpoch:
+		// first fully or partially rendered epoch and there is at least one
+		// rendered map in the next epoch; remove from indexed range
+		fmr.tailPartialEpoch = 0
+		fmr.maps.SetFirst((epoch + 1) << f.logMapsPerEpoch)
+		fmr.blocks.SetFirst(lastBlock + 1)
 		f.setRange(f.db, f.indexedView, fmr, false)
+	default:
+		// cannot be cleaned or unindexed; return with error
+		return false, errors.New("invalid tail epoch number")
+	}
+	// remove index data
+	if err := f.safeDeleteWithLogs(func(db ethdb.KeyValueStore, hashDbSafe bool, stopCb func() bool) error {
 		first := f.mapRowIndex(firstMap, 0)
 		count := f.mapRowIndex(firstMap+f.mapsPerEpoch, 0) - first
-	rawdb.DeleteFilterMapRows(f.db, common.NewRange(first, count))
+		if err := rawdb.DeleteFilterMapRows(f.db, common.NewRange(first, count), hashDbSafe, stopCb); err != nil {
+			return err
+		}
 		for mapIndex := firstMap; mapIndex < firstMap+f.mapsPerEpoch; mapIndex++ {
 			f.filterMapCache.Remove(mapIndex)
 		}
-	rawdb.DeleteFilterMapLastBlocks(f.db, common.NewRange(firstMap, f.mapsPerEpoch-1)) // keep last enrty
+		if err := rawdb.DeleteFilterMapLastBlocks(f.db, common.NewRange(firstMap, f.mapsPerEpoch-1), hashDbSafe, stopCb); err != nil { // keep last enrty
+			return err
+		}
 		for mapIndex := firstMap; mapIndex < firstMap+f.mapsPerEpoch-1; mapIndex++ {
 			f.lastBlockCache.Remove(mapIndex)
 		}
-	rawdb.DeleteBlockLvPointers(f.db, common.NewRange(firstBlock, lastBlock-firstBlock)) // keep last enrty
+		if err := rawdb.DeleteBlockLvPointers(f.db, common.NewRange(firstBlock, lastBlock-firstBlock), hashDbSafe, stopCb); err != nil { // keep last enrty
+			return err
+		}
 		for blockNumber := firstBlock; blockNumber < lastBlock; blockNumber++ {
 			f.lvPointerCache.Remove(blockNumber)
 		}
 		return nil
+	}, fmt.Sprintf("Deleting tail epoch #%d", epoch), func() bool {
+		f.processEvents()
+		return f.stop || !f.targetHeadIndexed()
+	}); err == nil {
+		// everything removed; mark as cleaned and report success
+		if f.cleanedEpochsBefore == epoch {
+			f.cleanedEpochsBefore = epoch + 1
+		}
+		return true, nil
+	} else {
+		// more data left in epoch range; mark as dirty and report unfinished
+		if f.cleanedEpochsBefore > epoch {
+			f.cleanedEpochsBefore = epoch
+		}
+		if err == rawdb.ErrDeleteRangeInterrupted {
+			return false, nil
+		}
+		return false, err
+	}
 }
 
 // exportCheckpoints exports epoch checkpoints in the format used by checkpoints.go.

core/filtermaps/indexer.go

@@ -67,14 +67,17 @@ func (f *FilterMaps) indexerLoop() {
 				}
 				f.lastFinal = f.finalBlock
 			}
-			if done, err := f.tryIndexTail(); err != nil {
+			// always attempt unindexing before indexing the tail in order to
-				f.disableForError("tail rendering", err)
+			// ensure that a potentially dirty previously unindexed epoch is
+			// always cleaned up before any new maps are rendered.
+			if done, err := f.tryUnindexTail(); err != nil {
+				f.disableForError("tail unindexing", err)
 				return
 			} else if !done {
 				continue
 			}
-			if done, err := f.tryUnindexTail(); err != nil {
+			if done, err := f.tryIndexTail(); err != nil {
-				f.disableForError("tail unindexing", err)
+				f.disableForError("tail rendering", err)
 				return
 			} else if !done {
 				continue
@@ -349,25 +352,24 @@ func (f *FilterMaps) tryIndexTail() (bool, error) {
 // Note that unindexing is very quick as it only removes continuous ranges of
 // data from the database and is also called while running head indexing.
 func (f *FilterMaps) tryUnindexTail() (bool, error) {
-	for {
+	firstEpoch := f.indexedRange.maps.First() >> f.logMapsPerEpoch
-		firstEpoch := (f.indexedRange.maps.First() - f.indexedRange.tailPartialEpoch) >> f.logMapsPerEpoch
+	if f.indexedRange.tailPartialEpoch > 0 && firstEpoch > 0 {
-		if f.needTailEpoch(firstEpoch) {
+		firstEpoch--
-			break
-		}
-		f.processEvents()
-		if f.stop {
-			return false, nil
 	}
+	for epoch := min(firstEpoch, f.cleanedEpochsBefore); !f.needTailEpoch(epoch); epoch++ {
 		if !f.startedTailUnindex {
 			f.startedTailUnindexAt = time.Now()
 			f.startedTailUnindex = true
 			f.ptrTailUnindexMap = f.indexedRange.maps.First() - f.indexedRange.tailPartialEpoch
 			f.ptrTailUnindexBlock = f.indexedRange.blocks.First() - f.tailPartialBlocks()
 		}
-		if err := f.deleteTailEpoch(firstEpoch); err != nil {
+		if done, err := f.deleteTailEpoch(epoch); !done {
-			log.Error("Log index tail epoch unindexing failed", "error", err)
 			return false, err
 		}
+		f.processEvents()
+		if f.stop || !f.targetHeadIndexed() {
+			return false, nil
+		}
 	}
 	if f.startedTailUnindex && f.indexedRange.hasIndexedBlocks() {
 		log.Info("Log index tail unindexing finished",

core/rawdb/accessors_indexes.go

@@ -354,10 +354,8 @@ func WriteFilterMapBaseRows(db ethdb.KeyValueWriter, mapRowIndex uint64, rows []
 	}
 }
 
-func DeleteFilterMapRows(db ethdb.KeyValueRangeDeleter, mapRows common.Range[uint64]) {
+func DeleteFilterMapRows(db ethdb.KeyValueStore, mapRows common.Range[uint64], hashDbSafe bool, stopCallback func() bool) error {
-	if err := db.DeleteRange(filterMapRowKey(mapRows.First(), false), filterMapRowKey(mapRows.AfterLast(), false)); err != nil {
+	return SafeDeleteRange(db, filterMapRowKey(mapRows.First(), false), filterMapRowKey(mapRows.AfterLast(), false), hashDbSafe, stopCallback)
-		log.Crit("Failed to delete range of filter map rows", "err", err)
-	}
 }
 
 // ReadFilterMapLastBlock retrieves the number of the block that generated the
@@ -394,10 +392,8 @@ func DeleteFilterMapLastBlock(db ethdb.KeyValueWriter, mapIndex uint32) {
 	}
 }
 
-func DeleteFilterMapLastBlocks(db ethdb.KeyValueRangeDeleter, maps common.Range[uint32]) {
+func DeleteFilterMapLastBlocks(db ethdb.KeyValueStore, maps common.Range[uint32], hashDbSafe bool, stopCallback func() bool) error {
-	if err := db.DeleteRange(filterMapLastBlockKey(maps.First()), filterMapLastBlockKey(maps.AfterLast())); err != nil {
+	return SafeDeleteRange(db, filterMapLastBlockKey(maps.First()), filterMapLastBlockKey(maps.AfterLast()), hashDbSafe, stopCallback)
-		log.Crit("Failed to delete range of filter map last block pointers", "err", err)
-	}
 }
 
 // ReadBlockLvPointer retrieves the starting log value index where the log values
@@ -431,10 +427,8 @@ func DeleteBlockLvPointer(db ethdb.KeyValueWriter, blockNumber uint64) {
 	}
 }
 
-func DeleteBlockLvPointers(db ethdb.KeyValueRangeDeleter, blocks common.Range[uint64]) {
+func DeleteBlockLvPointers(db ethdb.KeyValueStore, blocks common.Range[uint64], hashDbSafe bool, stopCallback func() bool) error {
-	if err := db.DeleteRange(filterMapBlockLVKey(blocks.First()), filterMapBlockLVKey(blocks.AfterLast())); err != nil {
+	return SafeDeleteRange(db, filterMapBlockLVKey(blocks.First()), filterMapBlockLVKey(blocks.AfterLast()), hashDbSafe, stopCallback)
-		log.Crit("Failed to delete range of block log value pointers", "err", err)
-	}
 }
 
 // FilterMapsRange is a storage representation of the block range covered by the
@@ -485,22 +479,22 @@ func DeleteFilterMapsRange(db ethdb.KeyValueWriter) {
 }
 
 // deletePrefixRange deletes everything with the given prefix from the database.
-func deletePrefixRange(db ethdb.KeyValueRangeDeleter, prefix []byte) error {
+func deletePrefixRange(db ethdb.KeyValueStore, prefix []byte, hashDbSafe bool, stopCallback func() bool) error {
 	end := bytes.Clone(prefix)
 	end[len(end)-1]++
-	return db.DeleteRange(prefix, end)
+	return SafeDeleteRange(db, prefix, end, hashDbSafe, stopCallback)
 }
 
 // DeleteFilterMapsDb removes the entire filter maps database
-func DeleteFilterMapsDb(db ethdb.KeyValueRangeDeleter) error {
+func DeleteFilterMapsDb(db ethdb.KeyValueStore, hashDbSafe bool, stopCallback func() bool) error {
-	return deletePrefixRange(db, []byte(filterMapsPrefix))
+	return deletePrefixRange(db, []byte(filterMapsPrefix), hashDbSafe, stopCallback)
 }
 
 // DeleteFilterMapsDb removes the old bloombits database and the associated
 // chain indexer database.
-func DeleteBloomBitsDb(db ethdb.KeyValueRangeDeleter) error {
+func DeleteBloomBitsDb(db ethdb.KeyValueStore, hashDbSafe bool, stopCallback func() bool) error {
-	if err := deletePrefixRange(db, bloomBitsPrefix); err != nil {
+	if err := deletePrefixRange(db, bloomBitsPrefix, hashDbSafe, stopCallback); err != nil {
 		return err
 	}
-	return deletePrefixRange(db, bloomBitsMetaPrefix)
+	return deletePrefixRange(db, bloomBitsMetaPrefix, hashDbSafe, stopCallback)
 }

core/rawdb/database.go

@@ -28,12 +28,16 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/ethdb/leveldb"
 	"github.com/ethereum/go-ethereum/ethdb/memorydb"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/olekukonko/tablewriter"
 )
 
+var ErrDeleteRangeInterrupted = errors.New("safe delete range operation interrupted")
+
 // freezerdb is a database wrapper that enables ancient chain segment freezing.
 type freezerdb struct {
 	ethdb.KeyValueStore
@@ -607,3 +611,71 @@ func ReadChainMetadata(db ethdb.KeyValueStore) [][]string {
 	}
 	return data
 }
+
+// SafeDeleteRange deletes all of the keys (and values) in the range
+// [start,end) (inclusive on start, exclusive on end).
+// If hashDbSafe is true then it always uses an iterator and skips hashdb trie
+// node entries. If it is false and the backing db is pebble db then it uses
+// the fast native range delete.
+// In case of fallback mode (hashdb or leveldb) the range deletion might be
+// very slow depending on the number of entries. In this case stopCallback
+// is periodically called and if it returns an error then SafeDeleteRange
+// stops and also returns that error. The callback is not called if native
+// range delete is used or there are a small number of keys only.
+func SafeDeleteRange(db ethdb.KeyValueStore, start, end []byte, hashDbSafe bool, stopCallback func() bool) error {
+	if !hashDbSafe {
+		// delete entire range; use fast native range delete on pebble db
+		for {
+			switch err := db.DeleteRange(start, end); err {
+			case nil:
+				return nil
+			case leveldb.ErrTooManyKeys:
+				if stopCallback() {
+					return ErrDeleteRangeInterrupted
+				}
+			default:
+				return err
+			}
+		}
+	}
+
+	var (
+		count, deleted, skipped int
+		buff                    = crypto.NewKeccakState()
+		startTime               = time.Now()
+	)
+
+	batch := db.NewBatch()
+	it := db.NewIterator(nil, start)
+	defer func() {
+		it.Release() // it might be replaced during the process
+		log.Debug("SafeDeleteRange finished", "deleted", deleted, "skipped", skipped, "elapsed", time.Since(startTime))
+	}()
+
+	for it.Next() && bytes.Compare(end, it.Key()) > 0 {
+		// Prevent deletion for trie nodes in hash mode
+		if len(it.Key()) != 32 || crypto.HashData(buff, it.Value()) != common.BytesToHash(it.Key()) {
+			if err := batch.Delete(it.Key()); err != nil {
+				return err
+			}
+			deleted++
+		} else {
+			skipped++
+		}
+		count++
+		if count > 10000 { // should not block for more than a second
+			if err := batch.Write(); err != nil {
+				return err
+			}
+			if stopCallback() {
+				return ErrDeleteRangeInterrupted
+			}
+			start = append(bytes.Clone(it.Key()), 0)
+			it.Release()
+			batch = db.NewBatch()
+			it = db.NewIterator(nil, start)
+			count = 0
+		}
+	}
+	return batch.Write()
+}

eth/backend.go

@@ -239,7 +239,12 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 	if err != nil {
 		return nil, err
 	}
-	fmConfig := filtermaps.Config{History: config.LogHistory, Disabled: config.LogNoHistory, ExportFileName: config.LogExportCheckpoints}
+	fmConfig := filtermaps.Config{
+		History:        config.LogHistory,
+		Disabled:       config.LogNoHistory,
+		ExportFileName: config.LogExportCheckpoints,
+		HashDbSafe:     scheme == rawdb.HashScheme,
+	}
 	chainView := eth.newChainView(eth.blockchain.CurrentBlock())
 	historyCutoff := eth.blockchain.HistoryPruningCutoff()
 	var finalBlock uint64