fix(mempool)!: stop accepting TXs in the mempool if we can't keep up …

…with reCheckTX (backport cometbft#3314) (cometbft#3338)

This PR is a combination of ideas from @ValarDragon, @hvanz and
@sergio-mena to alleviate nodes that, while not having their mempool
full "officially", they have too many TXs lingering in the mempool which
causes them to fall behind.
The mechanism works as follows:

* We mark when we start and end reChecking
* If, by the time a new block is decided we are still running the
previous `reCheckTx`, we declare the mempool as rechecktx-full
* Otherwise, we declare the mempool as not rechecktx-full

We have tested this and it fixes the failing nightlies that are blocking
us from cutting `v1.0.0-rc1`.

Some UTs need to be fixed, hence posting as draft for the moment.

---

#### PR checklist

- [ ] Tests written/updated
- [x] Changelog entry added in `.changelog` (we use
[unclog](https://github.com/informalsystems/unclog) to manage our
changelog)
- [ ] Updated relevant documentation (`docs/` or `spec/`) and code
comments
- [x] Title follows the [Conventional
Commits](https://www.conventionalcommits.org/en/v1.0.0/) spec
<hr>This is an automatic backport of pull request cometbft#3314 done by
[Mergify](https://mergify.com).

Co-authored-by: Sergio Mena <[email protected]>
Co-authored-by: Andy Nogueira <[email protected]>
Co-authored-by: Dev Ojha <[email protected]>
Co-authored-by: hvanz <[email protected]>
Co-authored-by: Hernán Vanzetto <[email protected]>

.changelog/unreleased/breaking-changes/3314-mempool-preupdate.md

@@ -0,0 +1,4 @@
+- `[mempool]` Add to the `Mempool` interface a new method `PreUpdate()`. This method should be
+  called before acquiring the mempool lock, to signal that a new update is coming. Also add to
+  `ErrMempoolIsFull` a new field `RecheckFull`.
+  ([\#3314](https://github.com/cometbft/cometbft/pull/3314))

.changelog/unreleased/improvements/3314-mempool-update-consider-full-when-rechecking.md

@@ -0,0 +1,4 @@
+- `[mempool]` Before updating the mempool, consider it as full if rechecking is still in progress.
+  This will stop accepting transactions in the mempool if the node can't keep up with re-CheckTx.
+  This improvement is implemented only in the v0 mempool.
+  ([\#3314](https://github.com/cometbft/cometbft/pull/3314))

blocksync/reactor_test.go

@@ -86,6 +86,7 @@ func newReactor(
 	mp := &mpmocks.Mempool{}
 	mp.On("Lock").Return()
 	mp.On("Unlock").Return()
+	mp.On("PreUpdate").Return()
 	mp.On("FlushAppConn", mock.Anything).Return(nil)
 	mp.On("Update",
 		mock.Anything,

consensus/replay_stubs.go

@@ -17,6 +17,7 @@ var _ mempl.Mempool = emptyMempool{}
 
 func (emptyMempool) Lock()            {}
 func (emptyMempool) Unlock()          {}
+func (emptyMempool) PreUpdate()       {}
 func (emptyMempool) Size() int        { return 0 }
 func (emptyMempool) SizeBytes() int64 { return 0 }
 func (emptyMempool) CheckTx(_ types.Tx, _ func(*abci.Response), _ mempl.TxInfo) error {

mempool/mempool.go

@@ -58,6 +58,10 @@ type Mempool interface {
 	// Unlock unlocks the mempool.
 	Unlock()
 
+	// PreUpdate signals that a new update is coming, before acquiring the mempool lock.
+	// If the mempool is still rechecking at this point, it should be considered full.
+	PreUpdate()
+
 	// Update informs the mempool that the given txs were committed and can be
 	// discarded.
 	//

mempool/nop_mempool.go

@@ -40,6 +40,8 @@ func (*NopMempool) Lock() {}
 // Unlock does nothing.
 func (*NopMempool) Unlock() {}
 
+func (*NopMempool) PreUpdate() {}
+
 // Update does nothing.
 func (*NopMempool) Update(
 	int64,

mempool/v0/clist_mempool.go

@@ -48,6 +48,8 @@ type CListMempool struct {
 	// serial (ie. by abci responses which are called in serial).
 	recheckCursor *clist.CElement // next expected response
 	recheckEnd    *clist.CElement // re-checking stops here
+	isRechecking  atomic.Bool     // true iff the rechecking process has begun and is not yet finished
+	recheckFull   atomic.Bool     // whether rechecking TXs cannot be completed before a new block is decided
 
 	// Map for quick access to txs to record sender in CheckTx.
 	// txsMap: txKey -> CElement
@@ -140,6 +142,15 @@ func (mem *CListMempool) Unlock() {
 	mem.updateMtx.Unlock()
 }
 
+// Safe for concurrent use by multiple goroutines.
+func (mem *CListMempool) PreUpdate() {
+	rechecking := mem.isRechecking.Load()
+	recheckFull := mem.recheckFull.Swap(rechecking)
+	if rechecking != recheckFull {
+		mem.logger.Debug("the state of recheckFull has flipped")
+	}
+}
+
 // Safe for concurrent use by multiple goroutines.
 func (mem *CListMempool) Size() int {
 	return mem.txs.Len()
@@ -350,12 +361,11 @@ func (mem *CListMempool) RemoveTxByKey(txKey types.TxKey) error {
 }
 
 func (mem *CListMempool) isFull(txSize int) error {
-	var (
-		memSize  = mem.Size()
-		txsBytes = mem.SizeBytes()
-	)
+	memSize := mem.Size()
+	txsBytes := mem.SizeBytes()
+	recheckFull := mem.recheckFull.Load()
 
-	if memSize >= mem.config.Size || int64(txSize)+txsBytes > mem.config.MaxTxsBytes {
+	if memSize >= mem.config.Size || int64(txSize)+txsBytes > mem.config.MaxTxsBytes || recheckFull {
 		return mempool.ErrMempoolIsFull{
 			NumTxs:      memSize,
 			MaxTxs:      mem.config.Size,
@@ -477,6 +487,8 @@ func (mem *CListMempool) resCbRecheck(req *abci.Request, res *abci.Response) {
 				// matching the one we received from the ABCI application.
 				// Return without processing any tx.
 				mem.recheckCursor = nil
+				mem.isRechecking.Store(false)
+				mem.recheckFull.Store(false)
 				return
 			}
 
@@ -499,6 +511,8 @@ func (mem *CListMempool) resCbRecheck(req *abci.Request, res *abci.Response) {
 		}
 		if mem.recheckCursor == mem.recheckEnd {
 			mem.recheckCursor = nil
+			mem.isRechecking.Store(false)
+			mem.recheckFull.Store(false)
 		} else {
 			mem.recheckCursor = mem.recheckCursor.Next()
 		}
@@ -665,6 +679,7 @@ func (mem *CListMempool) recheckTxs() {
 
 	mem.recheckCursor = mem.txs.Front()
 	mem.recheckEnd = mem.txs.Back()
+	mem.isRechecking.Store(true)
 
 	// Push txs to proxyAppConn
 	// NOTE: globalCb may be called concurrently.

mempool/v0/reactor_test.go

@@ -96,6 +96,7 @@ func TestReactorConcurrency(t *testing.T) {
 		go func() {
 			defer wg.Done()
 
+			reactors[0].mempool.PreUpdate()
 			reactors[0].mempool.Lock()
 			defer reactors[0].mempool.Unlock()
 
@@ -113,6 +114,7 @@ func TestReactorConcurrency(t *testing.T) {
 		go func() {
 			defer wg.Done()
 
+			reactors[1].mempool.PreUpdate()
 			reactors[1].mempool.Lock()
 			defer reactors[1].mempool.Unlock()
 			err := reactors[1].mempool.Update(1, []types.Tx{}, make([]*abci.ResponseDeliverTx, 0), nil, nil)

mempool/v1/mempool.go

@@ -117,6 +117,9 @@ func (txmp *TxMempool) Lock() { txmp.mtx.Lock() }
 // Unlock releases a write-lock on the mempool.
 func (txmp *TxMempool) Unlock() { txmp.mtx.Unlock() }
 
+// PreUpdate does nothing on the v1 mempool
+func (txmp *TxMempool) PreUpdate() {}
+
 // Size returns the number of valid transactions in the mempool. It is
 // thread-safe.
 func (txmp *TxMempool) Size() int { return txmp.txs.Len() }

state/execution.go

@@ -296,6 +296,7 @@ func (blockExec *BlockExecutor) Commit(
 	block *types.Block,
 	deliverTxResponses []*abci.ResponseDeliverTx,
 ) ([]byte, int64, error) {
+	blockExec.mempool.PreUpdate()
 	blockExec.mempool.Lock()
 	unlockMempool := func() { blockExec.mempool.Unlock() }
 

state/execution_test.go

@@ -50,6 +50,7 @@ func TestApplyBlock(t *testing.T) {
 	mp := &mpmocks.Mempool{}
 	mp.On("Lock").Return()
 	mp.On("Unlock").Return()
+	mp.On("PreUpdate").Return()
 	mp.On("FlushAppConn", mock.Anything).Return(nil)
 	mp.On("Update",
 		mock.Anything,
@@ -222,6 +223,7 @@ func TestBeginBlockByzantineValidators(t *testing.T) {
 	mp := &mpmocks.Mempool{}
 	mp.On("Lock").Return()
 	mp.On("Unlock").Return()
+	mp.On("PreUpdate").Return()
 	mp.On("FlushAppConn", mock.Anything).Return(nil)
 	mp.On("Update",
 		mock.Anything,
@@ -478,6 +480,7 @@ func TestEndBlockValidatorUpdates(t *testing.T) {
 	mp := &mpmocks.Mempool{}
 	mp.On("Lock").Return()
 	mp.On("Unlock").Return()
+	mp.On("PreUpdate").Return()
 	mp.On("FlushAppConn", mock.Anything).Return(nil)
 	mp.On("Update",
 		mock.Anything,
@@ -604,6 +607,7 @@ func TestEmptyPrepareProposal(t *testing.T) {
 	mp := &mpmocks.Mempool{}
 	mp.On("Lock").Return()
 	mp.On("Unlock").Return()
+	mp.On("PreUpdate").Return()
 	mp.On("FlushAppConn", mock.Anything).Return(nil)
 	mp.On("Update",
 		mock.Anything,

state/validation_test.go

@@ -35,6 +35,7 @@ func TestValidateBlockHeader(t *testing.T) {
 	mp := &mpmocks.Mempool{}
 	mp.On("Lock").Return()
 	mp.On("Unlock").Return()
+	mp.On("PreUpdate").Return()
 	mp.On("FlushAppConn", mock.Anything).Return(nil)
 	mp.On("Update",
 		mock.Anything,
@@ -120,6 +121,7 @@ func TestValidateBlockCommit(t *testing.T) {
 	mp := &mpmocks.Mempool{}
 	mp.On("Lock").Return()
 	mp.On("Unlock").Return()
+	mp.On("PreUpdate").Return()
 	mp.On("FlushAppConn", mock.Anything).Return(nil)
 	mp.On("Update",
 		mock.Anything,
@@ -259,6 +261,7 @@ func TestValidateBlockEvidence(t *testing.T) {
 	mp := &mpmocks.Mempool{}
 	mp.On("Lock").Return()
 	mp.On("Unlock").Return()
+	mp.On("PreUpdate").Return()
 	mp.On("FlushAppConn", mock.Anything).Return(nil)
 	mp.On("Update",
 		mock.Anything,