add FORCE_FINALIZED_BLOCK_ABOVE_THRESHOLD to prevent infinite lagging…

… when beacon misbehaves, ex. on testnet

core/blockchain.go

@@ -1792,13 +1792,8 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 				ptd := bc.GetTd(block.ParentHash(), block.NumberU64()-1)
 				td := new(big.Int).Add(block.Difficulty(), ptd)
 
-				var curFinalBlock *types.Header
-				if firehose.ReprocessingWithSyncTarget {
-					curFinalBlock = block.Header()
-				} else {
-					curFinalBlock = bc.CurrentFinalBlock()
-				}
-				firehoseContext.EndBlock(block, curFinalBlock, td)
+				finalBlockHeader := firehose.LastFinalBlock(block, bc.CurrentFinalBlock(), bc.GetBlockByNumber)
+				firehoseContext.EndBlock(block, finalBlockHeader, td)
 
 				firehoseContext.FlushBlock()
 			}
@@ -1879,17 +1874,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 				td = new(big.Int).Add(difficulty, ptd)
 			}
 
-			var finalBlockHeader *types.Header
-			if firehose.ReprocessingWithSyncTarget {
-				finalBlockHeader = block.Header()
-			} else {
-				if cur := bc.CurrentFinalBlock(); cur != 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.
-					finalBlockHeader = cur
-				}
-			}
-
+			finalBlockHeader := firehose.LastFinalBlock(block, bc.CurrentFinalBlock(), bc.GetBlockByNumber)
 			firehoseContext.EndBlock(block, finalBlockHeader, td)
 		}
 

firehose/context.go

@@ -171,6 +171,35 @@ func (ctx *Context) FinalizeBlock(block *types.Block) {
 	ctx.printer.Print("FINALIZE_BLOCK", Uint64(block.NumberU64()))
 }
 
+type blockGetter func(uint64) *types.Block
+
+func LastFinalBlock(currentBlock *types.Block, candidateLIB *types.Header, getBlockByNum blockGetter) *types.Header {
+	if ReprocessingWithSyncTarget {
+		return currentBlock.Header()
+	}
+
+	currentNum := currentBlock.NumberU64()
+	if candidateLIB != nil &&
+		candidateLIB.Number.Uint64() <= currentNum &&
+		(ForceFinalizedBlockAboveThreshold == 0 || candidateLIB.Number.Uint64()+ForceFinalizedBlockAboveThreshold > currentNum) {
+		return candidateLIB
+	}
+
+	if ForceFinalizedBlockAboveThreshold == 0 {
+		return nil
+	}
+
+	if currentNum < ForceFinalizedBlockAboveThreshold {
+		return nil
+	}
+
+	blk := getBlockByNum(currentNum - 200)
+	if blk != nil {
+		return blk.Header()
+	}
+	return nil
+}
+
 func (ctx *Context) EndBlock(block *types.Block, finalBlockHeader *types.Header, totalDifficulty *big.Int) {
 	endData := map[string]interface{}{
 		"header":          block.Header(),

firehose/context_test.go

@@ -13,6 +13,92 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestLastFinalBlock(t *testing.T) {
+
+	getBlockGetter := func(expectedBlockNum uint64) func(uint64) *types.Block {
+		return func(in uint64) *types.Block {
+			if in != expectedBlockNum {
+				panic("unexpected block number")
+			}
+			header := &types.Header{Number: big.NewInt(int64(expectedBlockNum))}
+			return types.NewBlockWithHeader(header)
+		}
+	}
+
+	tests := []struct {
+		name         string
+		current      *types.Block
+		candidate    *types.Header
+		threshold    uint64
+		getBlockNum  func(uint64) *types.Block
+		reprocessing bool
+		expect       uint64
+		expectNil    bool
+	}{
+		{
+			name:      "candidate is good",
+			current:   types.NewBlockWithHeader(&types.Header{Number: big.NewInt(10)}),
+			candidate: &types.Header{Number: big.NewInt(5)},
+			expect:    5,
+		},
+		{
+			name:      "candidate is good, with forceThreshold",
+			current:   types.NewBlockWithHeader(&types.Header{Number: big.NewInt(10)}),
+			candidate: &types.Header{Number: big.NewInt(5)},
+			threshold: 200,
+			expect:    5,
+		},
+		{
+			name:         "is reprocessing",
+			current:      types.NewBlockWithHeader(&types.Header{Number: big.NewInt(10)}),
+			candidate:    &types.Header{Number: big.NewInt(5)},
+			reprocessing: true,
+			expect:       10,
+		},
+		{
+			name:      "no candidate, no threshold",
+			current:   types.NewBlockWithHeader(&types.Header{Number: big.NewInt(10)}),
+			expectNil: true,
+		},
+		{
+			name:      "no candidate, below threshold",
+			current:   types.NewBlockWithHeader(&types.Header{Number: big.NewInt(10)}),
+			threshold: 200,
+			expectNil: true,
+		},
+		{
+			name:        "no candidate, above threshold",
+			current:     types.NewBlockWithHeader(&types.Header{Number: big.NewInt(1000)}),
+			threshold:   200,
+			getBlockNum: getBlockGetter(800),
+			expect:      800,
+		},
+		{
+			name:        "candidate too old",
+			current:     types.NewBlockWithHeader(&types.Header{Number: big.NewInt(1000)}),
+			candidate:   &types.Header{Number: big.NewInt(5)},
+			threshold:   200,
+			getBlockNum: getBlockGetter(800),
+			expect:      800,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ReprocessingWithSyncTarget = tt.reprocessing
+			ForceFinalizedBlockAboveThreshold = tt.threshold
+
+			if tt.expectNil {
+				assert.Nil(t, LastFinalBlock(tt.current, tt.candidate, tt.getBlockNum))
+			} else {
+				lib := LastFinalBlock(tt.current, tt.candidate, tt.getBlockNum)
+				require.NotNil(t, lib)
+				assert.Equal(t, tt.expect, lib.Number.Uint64())
+			}
+		})
+	}
+}
+
 func TestAccessList_marshal(t *testing.T) {
 	tests := []struct {
 		name    string

firehose/globals.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"os"
 	"reflect"
+	"strconv"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/params"
@@ -56,6 +57,9 @@ var GenesisConfig interface{}
 // This setting will cause the "Last Finalized Block" (LIB) set to follow each block that we process.
 var ReprocessingWithSyncTarget = false
 
+// ForceFinalizedBlockAboveThreshold can be used to assume that a part of the chain is finalized when a certain number of confirmations have passed, even when the beacon chain is drifting.
+var ForceFinalizedBlockAboveThreshold = uint64(0)
+
 var MissingGenesisPanicMessage = "Firehose requires to have the genesis config to properly emit genesis block for this chain " +
 	"but it appears it was not set properly. Ensure you are using either chain's specific flag like " +
 	"'--mainnet' or if using a custom network, you can use '--firehose-genesis' flag to provide. Firehose " +
@@ -106,6 +110,13 @@ func Init(
 		}
 	}
 
+	if forceFinalized := os.Getenv("FORCE_FINALIZED_BLOCK_ABOVE_THRESHOLD"); forceFinalized != "" {
+		thresh, err := strconv.ParseUint(forceFinalized, 10, 64)
+		if err == nil && thresh > 0 {
+			ForceFinalizedBlockAboveThreshold = thresh
+		}
+	}
+
 	if Enabled {
 		AllocateBuffers()
 	}