Merge branch 'fix-mempool-spam' into test/abci-routing

internal/mempool/reactor.go

@@ -195,6 +195,8 @@ func (r *Reactor) broadcastTxRoutine(ctx context.Context, peerID types.NodeID) {
 
 		memTx := nextGossipTx.Value.(*WrappedTx)
 
+		// We expect the peer to send tx back once it gets it, and that's
+		// when we will mark it as seen.
 		// NOTE: Transaction batching was disabled due to:
 		// https://github.com/tendermint/tendermint/issues/5796
 		if !memTx.HasPeer(peerMempoolID) {
@@ -205,9 +207,6 @@ func (r *Reactor) broadcastTxRoutine(ctx context.Context, peerID types.NodeID) {
 				r.logger.Error("failed to gossip transaction", "peerID", peerID, "error", err)
 				return
 			}
-			// This is optimistic approach, we assume that the transaction was successfully
-			// gossiped to the peer.
-			memTx.SetPeer(peerMempoolID)
 
 			r.logger.Debug("gossiped tx to peer",
 				"tx", tmstrings.LazySprintf("%X", memTx.tx.Hash()),

internal/p2p/channel_params.go

@@ -80,9 +80,9 @@ func ChannelDescriptors(cfg *config.Config) map[ChannelID]*ChannelDescriptor {
 			RecvMessageCapacity: mempoolBatchSize(cfg.Mempool.MaxTxBytes),
 			RecvBufferCapacity:  1000,
 			Name:                "mempool",
-			// SendRateLimit:       5,
-			// SendRateBurst:       20,
-			EnqueueTimeout: 10 * time.Millisecond,
+			SendRateLimit:       5,
+			SendRateBurst:       20,
+			EnqueueTimeout:      10 * time.Millisecond,
 		},
 	}
 

internal/p2p/conn/connection.go

@@ -622,6 +622,8 @@ type ChannelDescriptor struct {
 	Name string
 
 	// Timeout for enqueue operations on the incoming queue.
+	// It is applied to all messages received from remote peer
+	// and delievered to this channel.
 	// When timeout expires, messages will be silently dropped.
 	//
 	// If zero, enqueue operations will not time out.