[TxPool] Reject future transactions during high memory usage (#690)

* reject future transactions when the TxPool is low on available slots

txpool/txpool.go

@@ -46,6 +46,7 @@ var (
 	ErrAlreadyKnown            = errors.New("already known")
 	ErrOversizedData           = errors.New("oversized data")
 	ErrMaxEnqueuedLimitReached = errors.New("maximum number of enqueued transactions reached")
+	ErrRejectFutureTx          = errors.New("rejected future tx due to low slots")
 )
 
 // indicates origin of a transaction
@@ -656,6 +657,12 @@ func (p *TxPool) addTx(origin txOrigin, tx *types.Transaction) error {
 
 	if p.gauge.highPressure() {
 		p.signalPruning()
+
+		//	only accept transactions with expected nonce
+		if account := p.accounts.get(tx.From); account != nil &&
+			tx.Nonce > account.getNonce() {
+			return ErrRejectFutureTx
+		}
 	}
 
 	// check for overflow

txpool/txpool_test.go

@@ -447,6 +447,61 @@ func TestAddTxHighPressure(t *testing.T) {
 			//	pick up signal
 			_, ok := <-pool.pruneCh
 			assert.True(t, ok)
+
+			//	unblock the handler (handler would block entire test run)
+			<-pool.enqueueReqCh
+		},
+	)
+
+	t.Run(
+		"reject tx with nonce not matching expected",
+		func(t *testing.T) {
+			t.Parallel()
+
+			pool, err := newTestPool()
+			assert.NoError(t, err)
+			pool.SetSigner(&mockSigner{})
+
+			pool.createAccountOnce(addr1)
+			pool.accounts.get(addr1).nextNonce = 5
+
+			//	mock high pressure
+			slots := 1 + (highPressureMark*pool.gauge.max)/100
+			pool.gauge.increase(slots)
+
+			assert.ErrorIs(t,
+				ErrRejectFutureTx,
+				pool.addTx(local, newTx(addr1, 8, 1)),
+			)
+		},
+	)
+
+	t.Run(
+		"accept tx with expected nonce during high gauge level",
+		func(t *testing.T) {
+			t.Parallel()
+
+			pool, err := newTestPool()
+			assert.NoError(t, err)
+			pool.SetSigner(&mockSigner{})
+
+			pool.createAccountOnce(addr1)
+			pool.accounts.get(addr1).nextNonce = 5
+
+			//	mock high pressure
+			slots := 1 + (highPressureMark*pool.gauge.max)/100
+			println("slots", slots, "max", pool.gauge.max)
+			pool.gauge.increase(slots)
+
+			go func() {
+				assert.NoError(t,
+					pool.addTx(local, newTx(addr1, 5, 1)),
+				)
+			}()
+			enq := <-pool.enqueueReqCh
+
+			_, exists := pool.index.get(enq.tx.Hash)
+			assert.True(t, exists)
 		},
 	)
 }