forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-27 07:06:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

core/filtermaps: ensure proper shutdown in case of unexpected indexer error

Browse source
This commit is contained in:
Zsolt Felfoldi 2025-03-21 07:12:34 +01:00
parent b919bf009b
commit 8f8c8e2ff5
1 changed files with 16 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
core/filtermaps/indexer.go
View file

@ -44,13 +44,18 @@ func (f *FilterMaps) indexerLoop() {
if !f.indexedRange.initialized { if !f.indexedRange.initialized {
if err := f.init(); err != nil { if err := f.init(); err != nil {
log.Error("Error initializing log index", "error", err) log.Error("Error initializing log index", "error", err)
f.waitForEvent() // 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 continue
} }
} }
if !f.targetHeadIndexed() { if !f.targetHeadIndexed() {
if !f.tryIndexHead() { if !f.tryIndexHead() {
f.waitForEvent() // either shutdown or unexpected error; in the latter case ensure
// that proper shutdown is still possible.
f.processSingleEvent(true)
} }
} else { } else {
if f.finalBlock != f.lastFinal { if f.finalBlock != f.lastFinal {
@ -60,7 +65,7 @@ func (f *FilterMaps) indexerLoop() {
f.lastFinal = f.finalBlock f.lastFinal = f.finalBlock
} }
if f.tryIndexTail() && f.tryUnindexTail() { if f.tryIndexTail() && f.tryUnindexTail() {
f.waitForEvent() f.waitForNewHead()
} }
} }
} }
@ -119,8 +124,9 @@ func (f *FilterMaps) WaitIdle() {
} }
} }
// waitForEvent blocks until an event happens that the indexer might react to. // waitForNewHead blocks until there is a new target head to index and block
func (f *FilterMaps) waitForEvent() { // processing has been finished.
func (f *FilterMaps) waitForNewHead() {
for !f.stop && (f.blockProcessing || f.targetHeadIndexed()) { for !f.stop && (f.blockProcessing || f.targetHeadIndexed()) {
f.processSingleEvent(true) f.processSingleEvent(true)
} }
@ -129,13 +135,16 @@ func (f *FilterMaps) waitForEvent() {
// processEvents processes all events, blocking only if a block processing is // processEvents processes all events, blocking only if a block processing is
// happening and indexing should be suspended. // happening and indexing should be suspended.
func (f *FilterMaps) processEvents() { func (f *FilterMaps) processEvents() {
for !f.stop && f.processSingleEvent(f.blockProcessing) { for f.processSingleEvent(f.blockProcessing) {
} }
} }
// processSingleEvent processes a single event either in a blocking or // processSingleEvent processes a single event either in a blocking or
// non-blocking manner. // non-blocking manner. It returns true if it did process an event.
func (f *FilterMaps) processSingleEvent(blocking bool) bool { func (f *FilterMaps) processSingleEvent(blocking bool) bool {
if f.stop {
return false
}
if !f.hasTempRange { if !f.hasTempRange {
for _, mb := range f.matcherSyncRequests { for _, mb := range f.matcherSyncRequests {
mb.synced() mb.synced()