Merge pull request #220 from ethpandaops/pk910/fix-nil-pointer-in-con…

…tract-indexer fix nil pointer panic & make system routines more resilient against panics
ethpandaops · Jan 21, 2025 · dbc8579 · dbc8579
2 parents d523bf7 + d7600b8
commit dbc8579
Show file tree

Hide file tree

Showing 8 changed files with 22 additions and 10 deletions.
diff --git a/indexer/beacon/synchronizer.go b/indexer/beacon/synchronizer.go
@@ -114,7 +114,7 @@ func (s *synchronizer) stopSync() {
 }
 
 func (sync *synchronizer) runSync() {
-	defer utils.HandleSubroutinePanic("runSync")
+	defer utils.HandleSubroutinePanic("runSync", nil)
 
 	sync.runMutex.Lock()
 	defer sync.runMutex.Unlock()

diff --git a/indexer/execution/consolidation_indexer.go b/indexer/execution/consolidation_indexer.go
@@ -90,7 +90,7 @@ func (ci *ConsolidationIndexer) GetMatcherHeight() uint64 {
 
 // runConsolidationIndexerLoop is the main loop for the consolidation indexer
 func (ci *ConsolidationIndexer) runConsolidationIndexerLoop() {
-	defer utils.HandleSubroutinePanic("ConsolidationIndexer.runConsolidationIndexerLoop")
+	defer utils.HandleSubroutinePanic("ConsolidationIndexer.runConsolidationIndexerLoop", ci.runConsolidationIndexerLoop)
 
 	for {
 		time.Sleep(30 * time.Second)
@@ -170,6 +170,9 @@ func (ci *ConsolidationIndexer) parseRequestLog(log *types.Log, forkId *beacon.F
 	// get the validator indices for the source and target pubkeys
 	var sourceIndex, targetIndex *uint64
 	for index, validator := range ci.indexerCtx.beaconIndexer.GetValidatorSet(forkId) {
+		if validator == nil {
+			continue
+		}
 		if sourceIndex == nil && bytes.Equal(validator.PublicKey[:], sourcePubkey) {
 			index := uint64(index)
 			sourceIndex = &index

diff --git a/indexer/execution/deposit_indexer.go b/indexer/execution/deposit_indexer.go
@@ -87,7 +87,7 @@ func NewDepositIndexer(indexer *IndexerCtx) *DepositIndexer {
 
 // runDepositIndexerLoop is the main loop for the deposit indexer
 func (ds *DepositIndexer) runDepositIndexerLoop() {
-	defer utils.HandleSubroutinePanic("DepositIndexer.runDepositIndexerLoop")
+	defer utils.HandleSubroutinePanic("DepositIndexer.runDepositIndexerLoop", ds.runDepositIndexerLoop)
 
 	for {
 		time.Sleep(60 * time.Second)

diff --git a/indexer/execution/withdrawal_indexer.go b/indexer/execution/withdrawal_indexer.go
@@ -91,7 +91,7 @@ func (wi *WithdrawalIndexer) GetMatcherHeight() uint64 {
 
 // runWithdrawalIndexerLoop is the main loop for the withdrawal indexer
 func (wi *WithdrawalIndexer) runWithdrawalIndexerLoop() {
-	defer utils.HandleSubroutinePanic("WithdrawalIndexer.runWithdrawalIndexerLoop")
+	defer utils.HandleSubroutinePanic("WithdrawalIndexer.runWithdrawalIndexerLoop", wi.runWithdrawalIndexerLoop)
 
 	for {
 		time.Sleep(30 * time.Second)
@@ -172,7 +172,7 @@ func (wi *WithdrawalIndexer) parseRequestLog(log *types.Log, forkId *beacon.Fork
 
 	var validatorIndex *uint64
 	for index, validator := range validatorSet {
-		if bytes.Equal(validator.PublicKey[:], validatorPubkey) {
+		if validator != nil && bytes.Equal(validator.PublicKey[:], validatorPubkey) {
 			index := uint64(index)
 			validatorIndex = &index
 			break

diff --git a/indexer/mevrelay/mevindexer.go b/indexer/mevrelay/mevindexer.go
@@ -66,7 +66,7 @@ func (mev *MevIndexer) StartUpdater() {
 }
 
 func (mev *MevIndexer) runUpdaterLoop() {
-	defer utils.HandleSubroutinePanic("MevIndexer.runUpdaterLoop")
+	defer utils.HandleSubroutinePanic("MevIndexer.runUpdaterLoop", mev.runUpdaterLoop)
 
 	for {
 		time.Sleep(15 * time.Second)

diff --git a/services/fnsignatures.go b/services/fnsignatures.go
@@ -168,7 +168,7 @@ func (tss *TxSignaturesService) LookupSignatures(sigBytes []types.TxSignatureByt
 }
 
 func (tss *TxSignaturesService) runLookupLoop() {
-	defer utils.HandleSubroutinePanic("txsig.loop")
+	defer utils.HandleSubroutinePanic("TxSignaturesService.runLookupLoop", tss.runLookupLoop)
 
 	loopInterval := utils.Config.TxSignature.LookupInterval
 	if loopInterval == 0 {
@@ -204,7 +204,7 @@ func (tss *TxSignaturesService) processPendingSignatures() {
 
 		wg.Add(1)
 		go func(lookup *TxSignaturesLookup) {
-			defer utils.HandleSubroutinePanic("txsig.lookup")
+			defer utils.HandleSubroutinePanic("TxSignaturesService.processPendingSignatures.func1", nil)
 			err := tss.lookupSignature(lookup)
 			if err != nil {
 				logger_tss.Warnf("tx signatures lookup failed: %v", err)

diff --git a/services/validatornames.go b/services/validatornames.go
@@ -72,7 +72,7 @@ func (vn *ValidatorNames) StartUpdater() {
 }
 
 func (vn *ValidatorNames) runUpdaterLoop() {
-	defer utils.HandleSubroutinePanic("ValidatorNames.runUpdaterLoop")
+	defer utils.HandleSubroutinePanic("ValidatorNames.runUpdaterLoop", vn.runUpdaterLoop)
 
 	for {
 		time.Sleep(30 * time.Second)

diff --git a/utils/process.go b/utils/process.go
@@ -4,6 +4,7 @@ import (
 	"os"
 	"os/signal"
 	"runtime/debug"
+	"time"
 
 	"github.com/sirupsen/logrus"
 )
@@ -15,8 +16,16 @@ func WaitForCtrlC() {
 	<-c
 }
 
-func HandleSubroutinePanic(identifier string) {
+func HandleSubroutinePanic(identifier string, restartFn func()) {
 	if err := recover(); err != nil {
 		logrus.WithError(err.(error)).Errorf("uncaught panic in %v subroutine: %v, stack: %v", identifier, err, string(debug.Stack()))
+
+		if restartFn != nil {
+			go func() {
+				// Wait for 10 seconds before restarting the subroutine
+				time.Sleep(10 * time.Second)
+				restartFn()
+			}()
+		}
 	}
 }