txpool: Respect updates to block gas limit

There is a bug in the txpool which can manifest either as a race
condition, or in more novel network configurations.  When a transaction
is classified as either having `NotTooMuchGas` or its complement, this
classification will never change unless a change to the sender account
triggers reevaluation of that transaction.

If the block gas limit changes upwards, then any transactions previously
not marked as `NotTooMuchGas` will remain ineligible for promotion.
Similarly, if the block gas limit changes downwards, then any
transactions previously marked as `NotTooMuchGas` may inappropriately
remain as pending.

Although block gas limits may be relatively static in practice, there is
additionally a race condition which can cause this bug to manifest.
When the tx pool is first constructed, the gas limit is implicitly set
to 0.  Until the first update arrives with the configured gas limit, all
transactions received will be categorized as using too much gas, and
will never recover.

This change adds a check in the `OnNewBlock` path which checks if the
gas limit has changed, and if so triggers the re-evaluation of the
`NotTooMuchGas` status for all transactions.  In the normal case of
consistent gas limits, this code path is not executed.

erigon-lib/txpool/pool.go

@@ -400,7 +400,37 @@ func (p *TxPool) OnNewBlock(ctx context.Context, stateChanges *remote.StateChang
 	pendingBlobFee := stateChanges.PendingBlobFeePerGas
 	p.setBlobFee(pendingBlobFee)
 
-	p.blockGasLimit.Store(stateChanges.BlockGasLimit)
+	oldGasLimit := p.blockGasLimit.Swap(stateChanges.BlockGasLimit)
+	if oldGasLimit != stateChanges.BlockGasLimit {
+		p.all.ascendAll(func(mt *metaTx) bool {
+			var updated bool
+			if mt.Tx.Gas < stateChanges.BlockGasLimit {
+				updated = (mt.subPool & NotTooMuchGas) > 0
+				mt.subPool |= NotTooMuchGas
+			} else {
+				updated = (mt.subPool & NotTooMuchGas) == 0
+				mt.subPool &^= NotTooMuchGas
+			}
+
+			if mt.Tx.Traced {
+				p.logger.Info("TX TRACING: on block gas limit update", "idHash", fmt.Sprintf("%x", mt.Tx.IDHash), "senderId", mt.Tx.SenderID, "nonce", mt.Tx.Nonce, "subPool", mt.currentSubPool, "updated", updated)
+			}
+
+			if !updated {
+				return true
+			}
+
+			switch mt.currentSubPool {
+			case PendingSubPool:
+				p.pending.Updated(mt)
+			case BaseFeeSubPool:
+				p.baseFee.Updated(mt)
+			case QueuedSubPool:
+				p.queued.Updated(mt)
+			}
+			return true
+		})
+	}
 
 	for i, txn := range unwindBlobTxs.Txs {
 		if txn.Type == types.BlobTxType {

erigon-lib/txpool/pool_test.go

@@ -1048,3 +1048,76 @@ func TestDropRemoteAtNoGossip(t *testing.T) {
 	// no announcement because unprocessedRemoteTxs is already empty
 	assert.True(checkAnnouncementEmpty())
 }
+
+func TestGasLimitChanged(t *testing.T) {
+	assert, require := assert.New(t), require.New(t)
+	ch := make(chan types.Announcements, 100)
+	db, coreDB := memdb.NewTestPoolDB(t), memdb.NewTestDB(t)
+
+	cfg := txpoolcfg.DefaultConfig
+	sendersCache := kvcache.New(kvcache.DefaultCoherentConfig)
+	pool, err := New(ch, coreDB, cfg, sendersCache, *u256.N1, nil, nil, nil, fixedgas.DefaultMaxBlobsPerBlock, nil, log.New())
+	assert.NoError(err)
+	require.True(pool != nil)
+	ctx := context.Background()
+	var stateVersionID uint64 = 0
+	pendingBaseFee := uint64(200000)
+	// start blocks from 0, set empty hash - then kvcache will also work on this
+	h1 := gointerfaces.ConvertHashToH256([32]byte{})
+	var addr [20]byte
+	addr[0] = 1
+	v := make([]byte, types.EncodeSenderLengthForStorage(2, *uint256.NewInt(1 * common.Ether)))
+	types.EncodeSender(2, *uint256.NewInt(1 * common.Ether), v)
+	tx, err := db.BeginRw(ctx)
+	require.NoError(err)
+	defer tx.Rollback()
+
+	change := &remote.StateChangeBatch{
+		StateVersionId:      stateVersionID,
+		PendingBlockBaseFee: pendingBaseFee,
+		BlockGasLimit:       50_000,
+		ChangeBatch: []*remote.StateChange{
+			{BlockHeight: 0, BlockHash: h1},
+		},
+	}
+	change.ChangeBatch[0].Changes = append(change.ChangeBatch[0].Changes, &remote.AccountChange{
+		Action:  remote.Action_UPSERT,
+		Address: gointerfaces.ConvertAddressToH160(addr),
+		Data:    v,
+	})
+	err = pool.OnNewBlock(ctx, change, types.TxSlots{}, types.TxSlots{}, types.TxSlots{}, tx)
+	assert.NoError(err)
+
+	var txSlots types.TxSlots
+	txSlot1 := &types.TxSlot{
+		Tip:    *uint256.NewInt(300000),
+		FeeCap: *uint256.NewInt(300000),
+		Gas:    100_000,
+		Nonce:  3,
+	}
+	txSlot1.IDHash[0] = 1
+	txSlots.Append(txSlot1, addr[:], true)
+
+	reasons, err := pool.AddLocalTxs(ctx, txSlots, tx)
+	assert.NoError(err)
+	for _, reason := range reasons {
+		assert.Equal(txpoolcfg.Success, reason, reason.String())
+	}
+
+	mtx, ok := pool.byHash[string(txSlot1.IDHash[:])]
+	assert.True(ok)
+	assert.Zero(mtx.subPool&NotTooMuchGas, "Should be insufficient block space for the tx")
+
+	change.ChangeBatch[0].Changes = nil
+	change.BlockGasLimit = 150_000
+	err = pool.OnNewBlock(ctx, change, types.TxSlots{}, types.TxSlots{}, types.TxSlots{}, tx)
+	assert.NoError(err)
+
+	assert.NotZero(mtx.subPool&NotTooMuchGas, "Should now have block space for the tx")
+
+	change.BlockGasLimit = 50_000
+	err = pool.OnNewBlock(ctx, change, types.TxSlots{}, types.TxSlots{}, types.TxSlots{}, tx)
+	assert.NoError(err)
+
+	assert.Zero(mtx.subPool&NotTooMuchGas, "Should now have block space (again) for the tx")
+}