Merge pull request #2 from streamingfast/stepd/prevent-finalized-stal…

…l-on-syncing

do not advertise firehose finalized block when syncing behind beacon finalized block (release geth-v1.10.23-fh2.2)

core/state_processor.go

@@ -143,7 +143,15 @@ func (p *StateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg
 			td = new(big.Int).Add(difficulty, ptd)
 		}
 
-		firehoseContext.EndBlock(block, p.bc.CurrentFinalizedBlock(), td)
+		finalizedBlock := p.bc.CurrentFinalizedBlock()
+
+		if finalizedBlock != nil && firehose.SyncingBehindFinalized() {
+			// if beaconFinalizedBlockNum is in the future, the 'finalizedBlock' will not progress until we reach it.
+			// we don't want to advertise a super old finalizedBlock when reprocessing.
+			finalizedBlock = nil
+		}
+
+		firehoseContext.EndBlock(block, finalizedBlock, td)
 	}
 
 	return receipts, allLogs, *usedGas, nil

eth/catalyst/api.go

@@ -33,6 +33,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth"
 	"github.com/ethereum/go-ethereum/eth/downloader"
+	"github.com/ethereum/go-ethereum/firehose"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/rpc"
@@ -190,12 +191,26 @@ func (api *ConsensusAPI) ForkchoiceUpdatedV1(update beacon.ForkchoiceStateV1, pa
 			merger.ReachTTD()
 			api.eth.Downloader().Cancel()
 		}
+
+		if firehose.Enabled {
+			if update.FinalizedBlockHash != (common.Hash{}) {
+				if finalBlock := api.eth.BlockChain().GetBlockByHash(update.FinalizedBlockHash); finalBlock == nil {
+					// advertised finalized block is in the future, we are syncing
+					firehose.SetSyncingBehindFinalized(true)
+				}
+			}
+		}
+
 		log.Info("Forkchoice requested sync to new head", "number", header.Number, "hash", header.Hash())
 		if err := api.eth.Downloader().BeaconSync(api.eth.SyncMode(), header); err != nil {
 			return beacon.STATUS_SYNCING, err
 		}
 		return beacon.STATUS_SYNCING, nil
 	}
+
+	if firehose.Enabled {
+		firehose.SetSyncingBehindFinalized(false)
+	}
 	// Block is known locally, just sanity check that the beacon client does not
 	// attempt to push us back to before the merge.
 	if block.Difficulty().BitLen() > 0 || block.NumberU64() == 0 {

firehose/types.go

@@ -2,12 +2,27 @@ package firehose
 
 import (
 	"math/big"
+	"sync/atomic"
 
 	"github.com/golang-collections/collections/stack"
 )
 
 var EmptyValue = new(big.Int)
 
+var behindFinalized int32
+
+func SyncingBehindFinalized() bool {
+	return atomic.LoadInt32(&behindFinalized) != 0
+}
+
+func SetSyncingBehindFinalized(behind bool) {
+	if behind {
+		atomic.StoreInt32(&behindFinalized, 1)
+	} else {
+		atomic.StoreInt32(&behindFinalized, 0)
+	}
+}
+
 type logItem = map[string]interface{}
 
 type ExtendedStack struct {