core, eth, beacon: implement override builder flag

beacon/engine/types.go

@@ -86,6 +86,7 @@ type ExecutionPayloadEnvelope struct {
 	ExecutionPayload *ExecutableData `json:"executionPayload"  gencodec:"required"`
 	BlockValue       *big.Int        `json:"blockValue"  gencodec:"required"`
 	BlobsBundle      *BlobsBundleV1  `json:"blobsBundle"`
+	OverrideBuilder  bool            `json:"shouldOverrideBuilder"  gencodec:"required"`
 }
 
 type BlobsBundleV1 struct {

core/blockchain.go

@@ -215,7 +215,8 @@ type BlockChain struct {
 	futureBlocks *lru.Cache[common.Hash, *types.Block]
 
 	// Reorged transactions
-	droppedTxCache *lru.Cache[common.Hash, struct{}]
+	droppedTxCache     *lru.Cache[common.Hash, int]
+	forceLocalBuilding bool
 
 	wg            sync.WaitGroup //
 	quit          chan struct{}  // shutdown signal, closed in Stop.
@@ -271,7 +272,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis
 		blockCache:     lru.NewCache[common.Hash, *types.Block](blockCacheLimit),
 		txLookupCache:  lru.NewCache[common.Hash, *rawdb.LegacyTxLookupEntry](txLookupCacheLimit),
 		futureBlocks:   lru.NewCache[common.Hash, *types.Block](maxFutureBlocks),
-		droppedTxCache: lru.NewCache[common.Hash, struct{}](droppedTxCacheLimit),
+		droppedTxCache: lru.NewCache[common.Hash, int](droppedTxCacheLimit),
 		engine:         engine,
 		vmConfig:       vmConfig,
 	}
@@ -2155,10 +2156,15 @@ func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Block) error {
 	// transaction indexes, canonical chain indexes which above the head.
 	indexesBatch := bc.db.NewBatch()
 	for _, tx := range types.HashDifference(deletedTxs, addedTxs) {
-		if bc.droppedTxCache.Contains(tx) {
-			log.Warn("Transaction reorged twice", "hash", tx)
+		if key, ok := bc.droppedTxCache.Get(tx); ok {
+			if key > 2 {
+				// Transaction dropped at least three times
+				// indicates a censoring event
+				bc.forceLocalBuilding = true
+			}
+			bc.droppedTxCache.Add(tx, key+1)
 		} else {
-			bc.droppedTxCache.Add(tx, struct{}{})
+			bc.droppedTxCache.Add(tx, 1)
 		}
 		rawdb.DeleteTxLookupEntry(indexesBatch, tx)
 	}

core/blockchain_reader.go

@@ -411,3 +411,8 @@ func (bc *BlockChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscript
 func (bc *BlockChain) SubscribeBlockProcessingEvent(ch chan<- bool) event.Subscription {
 	return bc.scope.Track(bc.blockProcFeed.Subscribe(ch))
 }
+
+// ShouldForceLocalBuilding returns true if the node suspects transactions being censored.
+func (bc *BlockChain) ShouldForceLocalBuilding() bool {
+	return bc.forceLocalBuilding
+}

eth/catalyst/api.go

@@ -418,6 +418,7 @@ func (api *ConsensusAPI) getPayload(payloadID engine.PayloadID) (*engine.Executi
 	if data == nil {
 		return nil, engine.UnknownPayload
 	}
+	data.OverrideBuilder = api.eth.BlockChain().ShouldForceLocalBuilding()
 	return data, nil
 }