go/consensus/tendermint: Expire txes when CheckTx is disabled

When CheckTx is disabled (for debug purposes only, e.g. in E2E tests), we still
need to periodically remove old transactions as otherwise the mempool will fill
up. Keep track of transactions were added and invalidate them when they expire.

.changelog/2720.internal.md

@@ -0,0 +1,6 @@
+go/consensus/tendermint: Expire txes when CheckTx is disabled
+
+When CheckTx is disabled (for debug purposes only, e.g. in E2E tests), we
+still need to periodically remove old transactions as otherwise the mempool
+will fill up. Keep track of transactions were added and invalidate them when
+they expire.

go/consensus/tendermint/abci/mux.go

@@ -36,6 +36,9 @@ const (
 	stateKeyInitChainEvents = "OasisInitChainEvents"
 
 	metricsUpdateInterval = 10 * time.Second
+
+	// debugTxLifetime is the transaction mempool lifetime when CheckTx is disabled (debug only).
+	debugTxLifetime = 1 * time.Minute
 )
 
 var (
@@ -302,6 +305,9 @@ type abciMux struct {
 	// invalidatedTxs maps transaction hashes (hash.Hash) to a subscriber
 	// waiting for that transaction to become invalid.
 	invalidatedTxs sync.Map
+	// debugExpiringTxs maps transaction hashes to the time at which they were created. This is only
+	// used in case CheckTx is disabled (for debug purposes only).
+	debugExpiringTxs map[hash.Hash]time.Time
 }
 
 type invalidatedTxSubscription struct {
@@ -682,9 +688,46 @@ func (mux *abciMux) EstimateGas(caller signature.PublicKey, tx *transaction.Tran
 	return ctx.Gas().GasUsed(), nil
 }
 
+func (mux *abciMux) notifyInvalidatedCheckTx(txHash hash.Hash, err error) {
+	if item, exists := mux.invalidatedTxs.Load(txHash); exists {
+		// Notify subscriber.
+		sub := item.(*invalidatedTxSubscription)
+		select {
+		case sub.resultCh <- err:
+		default:
+		}
+		close(sub.resultCh)
+
+		mux.invalidatedTxs.Delete(txHash)
+	}
+}
+
 func (mux *abciMux) CheckTx(req types.RequestCheckTx) types.ResponseCheckTx {
 	if mux.state.disableCheckTx {
-		// Blindly accept all transactions if configured to do so.
+		// Blindly accept all transactions if configured to do so. We still need to periodically
+		// remove old transactions as otherwise the mempool will fill up, so keep track of when
+		// transactions were added and invalidate them after the configured interval.
+		var txHash hash.Hash
+		txHash.FromBytes(req.Tx)
+
+		if req.Type == types.CheckTxType_Recheck {
+			// Check timestamp.
+			if ts, ok := mux.debugExpiringTxs[txHash]; ok && mux.currentTime.Sub(ts) > debugTxLifetime {
+				delete(mux.debugExpiringTxs, txHash)
+
+				err := fmt.Errorf("mux: transaction expired (debug only)")
+				mux.notifyInvalidatedCheckTx(txHash, err)
+
+				return types.ResponseCheckTx{
+					Codespace: errors.UnknownModule,
+					Code:      1,
+					Log:       err.Error(),
+				}
+			}
+		} else {
+			mux.debugExpiringTxs[txHash] = mux.currentTime
+		}
+
 		return types.ResponseCheckTx{
 			Code: types.CodeTypeOK,
 		}
@@ -706,17 +749,7 @@ func (mux *abciMux) CheckTx(req types.RequestCheckTx) types.ResponseCheckTx {
 			var txHash hash.Hash
 			txHash.FromBytes(req.Tx)
 
-			if item, exists := mux.invalidatedTxs.Load(txHash); exists {
-				// Notify subscriber.
-				sub := item.(*invalidatedTxSubscription)
-				select {
-				case sub.resultCh <- err:
-				default:
-				}
-				close(sub.resultCh)
-
-				mux.invalidatedTxs.Delete(txHash)
-			}
+			mux.notifyInvalidatedCheckTx(txHash, err)
 		}
 
 		return types.ResponseCheckTx{
@@ -911,6 +944,11 @@ func newABCIMux(ctx context.Context, upgrader upgrade.Backend, cfg *ApplicationC
 		lastBeginBlock: -1,
 	}
 
+	// Create a map of expiring transactions if CheckTx is disabled (debug only).
+	if state.disableCheckTx {
+		mux.debugExpiringTxs = make(map[hash.Hash]time.Time)
+	}
+
 	mux.logger.Debug("ABCI multiplexer initialized",
 		"block_height", state.BlockHeight(),
 		"block_hash", hex.EncodeToString(state.BlockHash()),