v0.34.x: Prevent a transaction to appear twice in the mempool (backport

#890) (#927)

* add a test to trigger the issue
* add a fix (in particular, we track the sender when receiving a tx twice)
* add a changelog
* update fix and test wrt. v0.34.x

---------
Co-authored-by: Daniel <daniel.cason@informal.systems>
Co-authored-by: Pierre Sutra <0track@gmail.com>

.changelog/unreleased/bug-fixes/890-mempool-fix-cache.md

@@ -0,0 +1 @@
+- `[mempool/clist_mempool]` \#890 Prevent a transaction to appear twice in the mempool  (@otrack)

mempool/v0/clist_mempool.go

@@ -391,6 +391,20 @@ func (mem *CListMempool) resCbFirstTime(
 				return
 			}
 
+			// Check transaction not already in the mempool
+			if e, ok := mem.txsMap.Load(types.Tx(tx).Key()); ok {
+				memTx := e.(*clist.CElement).Value.(*mempoolTx)
+				memTx.senders.LoadOrStore(peerID, true)
+				mem.logger.Debug(
+					"transaction already there, not adding it again",
+					"tx", types.Tx(tx).Hash(),
+					"res", r,
+					"height", mem.height,
+					"total", mem.Size(),
+				)
+				return
+			}
+
 			memTx := &mempoolTx{
 				height:    mem.height,
 				gasWanted: r.CheckTx.GasWanted,

mempool/v0/clist_mempool_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	mrand "math/rand"
 	"os"
+	"strconv"
 	"testing"
 	"time"
 
@@ -638,6 +639,51 @@ func TestMempoolTxsBytes(t *testing.T) {
 
 }
 
+func TestMempoolNoCacheOverflow(t *testing.T) {
+	sockPath := fmt.Sprintf("unix:///tmp/echo_%v.sock", cmtrand.Str(6))
+	app := kvstore.NewApplication()
+	_, server := newRemoteApp(t, sockPath, app)
+	t.Cleanup(func() {
+		if err := server.Stop(); err != nil {
+			t.Error(err)
+		}
+	})
+	cfg := config.ResetTestRoot("mempool_test")
+	mp, cleanup := newMempoolWithAppAndConfig(proxy.NewRemoteClientCreator(sockPath, "socket", true), cfg)
+	defer cleanup()
+
+	// add tx0
+	var tx0 = types.Tx([]byte{0x01})
+	err := mp.CheckTx(tx0, nil, mempool.TxInfo{})
+	require.NoError(t, err)
+	err = mp.FlushAppConn()
+	require.NoError(t, err)
+
+	// saturate the cache to remove tx0
+	for i := 1; i <= mp.config.CacheSize; i++ {
+		err = mp.CheckTx(types.Tx([]byte(strconv.Itoa(i))), nil, mempool.TxInfo{})
+		require.NoError(t, err)
+	}
+	err = mp.FlushAppConn()
+	require.NoError(t, err)
+	assert.False(t, mp.cache.Has(types.Tx([]byte{0x01})))
+
+	// add again tx0
+	err = mp.CheckTx(tx0, nil, mempool.TxInfo{})
+	require.NoError(t, err)
+	err = mp.FlushAppConn()
+	require.NoError(t, err)
+
+	// tx0 should appear only once in mp.txs
+	found := 0
+	for e := mp.txs.Front(); e != nil; e = e.Next() {
+		if types.Tx.Key(e.Value.(*mempoolTx).tx) == types.Tx.Key(tx0) {
+			found++
+		}
+	}
+	assert.True(t, found == 1)
+}
+
 // This will non-deterministically catch some concurrency failures like
 // https://github.com/tendermint/tendermint/issues/3509
 // TODO: all of the tests should probably also run using the remote proxy app