core/filtermaps: check cutoff during checkpoint sync; revert to unindexed mode if init fails

core/filtermaps/filtermaps.go

@@ -53,7 +53,8 @@ type FilterMaps struct {
 	// This is configured by the --history.logs.disable Geth flag.
 	// We chose to implement disabling this way because it requires less special
 	// case logic in eth/filters.
-	disabled bool
+	disabled   bool
+	disabledCh chan struct{} // closed by indexer if disabled
 
 	closeCh        chan struct{}
 	closeWg        sync.WaitGroup
@@ -196,6 +197,7 @@ func NewFilterMaps(db ethdb.KeyValueStore, initView *ChainView, historyCutoff, f
 		blockProcessingCh: make(chan bool, 1),
 		history:           config.History,
 		disabled:          config.Disabled,
+		disabledCh:        make(chan struct{}),
 		exportFileName:    config.ExportFileName,
 		Params:            params,
 		indexedRange: filterMapsRange{
@@ -278,8 +280,13 @@ func (f *FilterMaps) initChainView(chainView *ChainView) *ChainView {
 }
 
 // reset un-initializes the FilterMaps structure and removes all related data from
-// the database. The function returns true if everything was successfully removed.
-func (f *FilterMaps) reset() bool {
+// the database.
+// Note that in case of leveldb database the fallback implementation of DeleteRange
+// might take a long time to finish and deleting the entire database may be
+// interrupted by a shutdown. Deleting the filterMapsRange entry first does
+// guarantee though that the next init() will not return successfully until the
+// entire database has been cleaned.
+func (f *FilterMaps) reset() {
 	f.indexLock.Lock()
 	f.indexedRange = filterMapsRange{}
 	f.indexedView = nil
@@ -292,11 +299,16 @@ func (f *FilterMaps) reset() bool {
 	// 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)
-	return f.safeDeleteRange(rawdb.DeleteFilterMapsDb, "Resetting log index database")
+	f.safeDeleteRange(rawdb.DeleteFilterMapsDb, "Resetting log index database")
 }
 
 // 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") {
+		return errors.New("could not reset log index database")
+	}
+
 	f.indexLock.Lock()
 	defer f.indexLock.Unlock()
 
@@ -317,6 +329,13 @@ func (f *FilterMaps) init() error {
 			bestIdx, bestLen = idx, max
 		}
 	}
+	var initBlockNumber uint64
+	if bestLen > 0 {
+		initBlockNumber = checkpoints[bestIdx][bestLen-1].BlockNumber
+	}
+	if initBlockNumber < f.historyCutoff {
+		return errors.New("cannot start indexing before history cutoff point")
+	}
 	batch := f.db.NewBatch()
 	for epoch := range bestLen {
 		cp := checkpoints[bestIdx][epoch]

core/filtermaps/indexer.go

@@ -36,6 +36,7 @@ func (f *FilterMaps) indexerLoop() {
 
 	if f.disabled {
 		f.reset()
+		close(f.disabledCh)
 		return
 	}
 	log.Info("Started log indexer")
@@ -48,12 +49,11 @@ func (f *FilterMaps) indexerLoop() {
 				continue
 			}
 			if err := f.init(); err != nil {
-				log.Error("Error initializing log index", "error", err)
-				// unexpected error; there is not a lot we can do here, maybe it
-				// recovers, maybe not. Calling event processing here ensures
-				// that we can still properly shutdown in case of an infinite loop.
-				f.processSingleEvent(true)
-				continue
+				log.Error("Error initializing log index; reverting to unindexed mode", "error", err)
+				f.reset()
+				f.disabled = true
+				close(f.disabledCh)
+				return
 			}
 		}
 		if !f.targetHeadIndexed() {

core/filtermaps/matcher_backend.go

@@ -141,10 +141,6 @@ func (fm *FilterMapsMatcherBackend) synced() {
 // range that has not been changed and has been consistent with all states of the
 // chain since the previous SyncLogIndex or the creation of the matcher backend.
 func (fm *FilterMapsMatcherBackend) SyncLogIndex(ctx context.Context) (SyncRange, error) {
-	if fm.f.disabled {
-		return SyncRange{HeadNumber: fm.f.targetView.headNumber}, nil
-	}
-
 	syncCh := make(chan SyncRange, 1)
 	fm.f.matchersLock.Lock()
 	fm.syncCh = syncCh
@@ -154,12 +150,16 @@ func (fm *FilterMapsMatcherBackend) SyncLogIndex(ctx context.Context) (SyncRange
 	case fm.f.matcherSyncCh <- fm:
 	case <-ctx.Done():
 		return SyncRange{}, ctx.Err()
+	case <-fm.f.disabledCh:
+		return SyncRange{HeadNumber: fm.f.targetView.headNumber}, nil
 	}
 	select {
 	case vr := <-syncCh:
 		return vr, nil
 	case <-ctx.Done():
 		return SyncRange{}, ctx.Err()
+	case <-fm.f.disabledCh:
+		return SyncRange{HeadNumber: fm.f.targetView.headNumber}, nil
 	}
 }