mempool no gossip back

CHANGELOG_PENDING.md

@@ -28,4 +28,5 @@ Friendly reminder, we have a [bug bounty program](https://hackerone.com/tendermi
 
 - [crypto/merkle] [\#2756](https://github.com/tendermint/tendermint/issues/2756) Fix crypto/merkle ProofOperators.Verify to check bounds on keypath parts.
 - [mempool] fix a bug where we create a WAL despite `wal_dir` being empty
-- [p2p] \#2771 Fix `peer-id` label name in prometheus metrics
+- [mempool] \#2778 No longer send txs back to peers who sent it to you
+- [p2p] \#2771 Fix `peer-id` label name in prometheus metrics\

abci/client/grpc_client.go

@@ -111,8 +111,8 @@ func (cli *grpcClient) Error() error {
 // NOTE: callback may get internally generated flush responses.
 func (cli *grpcClient) SetResponseCallback(resCb Callback) {
 	cli.mtx.Lock()
-	defer cli.mtx.Unlock()
 	cli.resCb = resCb
+	cli.mtx.Unlock()
 }
 
 //----------------------------------------

abci/client/local_client.go

@@ -30,8 +30,8 @@ func NewLocalClient(mtx *sync.Mutex, app types.Application) *localClient {
 
 func (app *localClient) SetResponseCallback(cb Callback) {
 	app.mtx.Lock()
-	defer app.mtx.Unlock()
 	app.Callback = cb
+	app.mtx.Unlock()
 }
 
 // TODO: change types.Application to include Error()?

docs/spec/reactors/mempool/reactor.md

@@ -12,3 +12,5 @@ for details.
 
 Sending incorrectly encoded data or data exceeding `maxMsgSize` will result
 in stopping the peer.
+
+The mempool will not send a tx back to any peer which it received it from.

mempool/bench_test.go

@@ -25,6 +25,18 @@ func BenchmarkReap(b *testing.B) {
 	}
 }
 
+func BenchmarkCheckTx(b *testing.B) {
+	app := kvstore.NewKVStoreApplication()
+	cc := proxy.NewLocalClientCreator(app)
+	mempool := newMempoolWithApp(cc)
+
+	for i := 0; i < b.N; i++ {
+		tx := make([]byte, 8)
+		binary.BigEndian.PutUint64(tx, uint64(i))
+		mempool.CheckTx(tx, nil)
+	}
+}
+
 func BenchmarkCacheInsertTime(b *testing.B) {
 	cache := newMapTxCache(b.N)
 	txs := make([][]byte, b.N)
@@ -34,7 +46,7 @@ func BenchmarkCacheInsertTime(b *testing.B) {
 	}
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		cache.Push(txs[i])
+		cache.Push(txs[i], 0)
 	}
 }
 
@@ -46,7 +58,7 @@ func BenchmarkCacheRemoveTime(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		txs[i] = make([]byte, 8)
 		binary.BigEndian.PutUint64(txs[i], uint64(i))
-		cache.Push(txs[i])
+		cache.Push(txs[i], 0)
 	}
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {

mempool/mempool.go

@@ -9,6 +9,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/tendermint/tendermint/abci/client"
+
 	"github.com/pkg/errors"
 
 	amino "github.com/tendermint/go-amino"
@@ -177,7 +179,7 @@ func NewMempool(
 	} else {
 		mempool.cache = nopTxCache{}
 	}
-	proxyAppConn.SetResponseCallback(mempool.resCb)
+	proxyAppConn.SetResponseCallback(mempool.resCb(0))
 	for _, option := range options {
 		option(mempool)
 	}
@@ -296,6 +298,10 @@ func (mem *Mempool) TxsWaitChan() <-chan struct{} {
 //     It gets called from another goroutine.
 // CONTRACT: Either cb will get called, or err returned.
 func (mem *Mempool) CheckTx(tx types.Tx, cb func(*abci.Response)) (err error) {
+	return mem.checkTxFromPeer(tx, cb, 0)
+}
+
+func (mem *Mempool) checkTxFromPeer(tx types.Tx, cb func(*abci.Response), peerID uint16) (err error) {
 	mem.proxyMtx.Lock()
 	defer mem.proxyMtx.Unlock()
 
@@ -310,7 +316,7 @@ func (mem *Mempool) CheckTx(tx types.Tx, cb func(*abci.Response)) (err error) {
 	}
 
 	// CACHE
-	if !mem.cache.Push(tx) {
+	if !mem.cache.Push(tx, peerID) {
 		return ErrTxInCache
 	}
 	// END CACHE
@@ -333,6 +339,7 @@ func (mem *Mempool) CheckTx(tx types.Tx, cb func(*abci.Response)) (err error) {
 	if err = mem.proxyAppConn.Error(); err != nil {
 		return err
 	}
+	mem.proxyAppConn.SetResponseCallback(mem.resCb(peerID))
 	reqRes := mem.proxyAppConn.CheckTxAsync(tx)
 	if cb != nil {
 		reqRes.SetCallback(cb)
@@ -342,17 +349,19 @@ func (mem *Mempool) CheckTx(tx types.Tx, cb func(*abci.Response)) (err error) {
 }
 
 // ABCI callback function
-func (mem *Mempool) resCb(req *abci.Request, res *abci.Response) {
-	if mem.recheckCursor == nil {
-		mem.resCbNormal(req, res)
-	} else {
-		mem.metrics.RecheckTimes.Add(1)
-		mem.resCbRecheck(req, res)
+func (mem *Mempool) resCb(peerID uint16) abcicli.Callback {
+	return func(req *abci.Request, res *abci.Response) {
+		if mem.recheckCursor == nil {
+			mem.resCbNormal(req, res, peerID)
+		} else {
+			mem.metrics.RecheckTimes.Add(1)
+			mem.resCbRecheck(req, res)
+		}
+		mem.metrics.Size.Set(float64(mem.Size()))
 	}
-	mem.metrics.Size.Set(float64(mem.Size()))
 }
 
-func (mem *Mempool) resCbNormal(req *abci.Request, res *abci.Response) {
+func (mem *Mempool) resCbNormal(req *abci.Request, res *abci.Response, peerID uint16) {
 	switch r := res.Value.(type) {
 	case *abci.Response_CheckTx:
 		tx := req.GetCheckTx().Tx
@@ -366,7 +375,9 @@ func (mem *Mempool) resCbNormal(req *abci.Request, res *abci.Response) {
 				counter:   mem.counter,
 				height:    mem.height,
 				gasWanted: r.CheckTx.GasWanted,
-				tx:        tx,
+				// senders:   []uint16{mem.cache.GetPeerID(tx)},
+				senders: []uint16{peerID},
+				tx:      tx,
 			}
 			mem.txs.PushBack(memTx)
 			mem.logger.Info("Added good transaction",
@@ -607,6 +618,7 @@ type mempoolTx struct {
 	counter   int64    // a simple incrementing counter
 	height    int64    // height that this tx had been validated in
 	gasWanted int64    // amount of gas this tx states it will require
+	senders   []uint16 // ids of peers who've sent us this tx
 	tx        types.Tx //
 }
 
@@ -619,7 +631,7 @@ func (memTx *mempoolTx) Height() int64 {
 
 type txCache interface {
 	Reset()
-	Push(tx types.Tx) bool
+	Push(tx types.Tx, peerID uint16) bool
 	Remove(tx types.Tx)
 }
 
@@ -653,14 +665,27 @@ func (cache *mapTxCache) Reset() {
 
 // Push adds the given tx to the cache and returns true. It returns false if tx
 // is already in the cache.
-func (cache *mapTxCache) Push(tx types.Tx) bool {
+func (cache *mapTxCache) Push(tx types.Tx, peerID uint16) bool {
 	cache.mtx.Lock()
 	defer cache.mtx.Unlock()
 
 	// Use the tx hash in the cache
 	txHash := sha256.Sum256(tx)
 	if moved, exists := cache.map_[txHash]; exists {
 		cache.list.MoveToFront(moved)
+		memTx, succ := moved.Value.(*mempoolTx)
+		// succ would be false if tx was reaped
+		if succ {
+			for i := 0; i < len(memTx.senders); i++ {
+				if peerID == memTx.senders[i] {
+					// TODO: consider punishing peer for dups,
+					// its non-trivial since invalid txs can become valid,
+					// but they can spam the same tx with little cost to them atm.
+					return false
+				}
+			}
+			memTx.senders = append(memTx.senders, peerID)
+		}
 		return false
 	}
 
@@ -694,6 +719,6 @@ type nopTxCache struct{}
 
 var _ txCache = (*nopTxCache)(nil)
 
-func (nopTxCache) Reset()             {}
-func (nopTxCache) Push(types.Tx) bool { return true }
-func (nopTxCache) Remove(types.Tx)    {}
+func (nopTxCache) Reset()                     {}
+func (nopTxCache) Push(types.Tx, uint16) bool { return true }
+func (nopTxCache) Remove(types.Tx)            {}

mempool/mempool_test.go

@@ -55,7 +55,7 @@ func ensureFire(t *testing.T, ch <-chan struct{}, timeoutMS int) {
 	}
 }
 
-func checkTxs(t *testing.T, mempool *Mempool, count int) types.Txs {
+func checkTxs(t *testing.T, mempool *Mempool, count int, peerID uint16) types.Txs {
 	txs := make(types.Txs, count)
 	for i := 0; i < count; i++ {
 		txBytes := make([]byte, 20)
@@ -64,7 +64,7 @@ func checkTxs(t *testing.T, mempool *Mempool, count int) types.Txs {
 		if err != nil {
 			t.Error(err)
 		}
-		if err := mempool.CheckTx(txBytes, nil); err != nil {
+		if err := mempool.checkTxFromPeer(txBytes, nil, peerID); err != nil {
 			// Skip invalid txs.
 			// TestMempoolFilters will fail otherwise. It asserts a number of txs
 			// returned.
@@ -83,7 +83,7 @@ func TestReapMaxBytesMaxGas(t *testing.T) {
 	mempool := newMempoolWithApp(cc)
 
 	// Ensure gas calculation behaves as expected
-	checkTxs(t, mempool, 1)
+	checkTxs(t, mempool, 1, 0)
 	tx0 := mempool.TxsFront().Value.(*mempoolTx)
 	// assert that kv store has gas wanted = 1.
 	require.Equal(t, app.CheckTx(tx0.tx).GasWanted, int64(1), "KVStore had a gas value neq to 1")
@@ -117,7 +117,7 @@ func TestReapMaxBytesMaxGas(t *testing.T) {
 		{20, 20000, 30, 20},
 	}
 	for tcIndex, tt := range tests {
-		checkTxs(t, mempool, tt.numTxsToCreate)
+		checkTxs(t, mempool, tt.numTxsToCreate, 0)
 		got := mempool.ReapMaxBytesMaxGas(tt.maxBytes, tt.maxGas)
 		assert.Equal(t, tt.expectedNumTxs, len(got), "Got %d txs, expected %d, tc #%d",
 			len(got), tt.expectedNumTxs, tcIndex)
@@ -157,7 +157,7 @@ func TestMempoolFilters(t *testing.T) {
 	}
 	for tcIndex, tt := range tests {
 		mempool.Update(1, emptyTxArr, tt.preFilter, tt.postFilter)
-		checkTxs(t, mempool, tt.numTxsToCreate)
+		checkTxs(t, mempool, tt.numTxsToCreate, 0)
 		require.Equal(t, tt.expectedNumTxs, mempool.Size(), "mempool had the incorrect size, on test case %d", tcIndex)
 		mempool.Flush()
 	}
@@ -175,7 +175,7 @@ func TestTxsAvailable(t *testing.T) {
 	ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
 
 	// send a bunch of txs, it should only fire once
-	txs := checkTxs(t, mempool, 100)
+	txs := checkTxs(t, mempool, 100, 0)
 	ensureFire(t, mempool.TxsAvailable(), timeoutMS)
 	ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
 
@@ -190,7 +190,7 @@ func TestTxsAvailable(t *testing.T) {
 	ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
 
 	// send a bunch more txs. we already fired for this height so it shouldnt fire again
-	moreTxs := checkTxs(t, mempool, 50)
+	moreTxs := checkTxs(t, mempool, 50, 0)
 	ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
 
 	// now call update with all the txs. it should not fire as there are no txs left
@@ -201,7 +201,7 @@ func TestTxsAvailable(t *testing.T) {
 	ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
 
 	// send a bunch more txs, it should only fire once
-	checkTxs(t, mempool, 100)
+	checkTxs(t, mempool, 100, 0)
 	ensureFire(t, mempool.TxsAvailable(), timeoutMS)
 	ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
 }
@@ -324,7 +324,7 @@ func TestCacheRemove(t *testing.T) {
 		txBytes := make([]byte, 32)
 		rand.Read(txBytes)
 		txs[i] = txBytes
-		cache.Push(txBytes)
+		cache.Push(txBytes, 0)
 		// make sure its added to both the linked list and the map
 		require.Equal(t, i+1, len(cache.map_))
 		require.Equal(t, i+1, cache.list.Len())

mempool/reactor.go

@@ -3,6 +3,7 @@ package mempool
 import (
 	"fmt"
 	"reflect"
+	"sync"
 	"time"
 
 	amino "github.com/tendermint/go-amino"
@@ -23,17 +24,27 @@ const (
 )
 
 // MempoolReactor handles mempool tx broadcasting amongst peers.
+// It maintains a map from peer ID to counter, to prevent gossiping txs to the
+// peers you received it from.
 type MempoolReactor struct {
 	p2p.BaseReactor
 	config  *cfg.MempoolConfig
 	Mempool *Mempool
+
+	idMtx   *sync.RWMutex
+	nextID  uint16 // assumes that a node will never have over 65536 peers
+	peerMap map[p2p.ID]uint16
 }
 
 // NewMempoolReactor returns a new MempoolReactor with the given config and mempool.
 func NewMempoolReactor(config *cfg.MempoolConfig, mempool *Mempool) *MempoolReactor {
 	memR := &MempoolReactor{
 		config:  config,
 		Mempool: mempool,
+
+		idMtx:   &sync.RWMutex{},
+		peerMap: make(map[p2p.ID]uint16),
+		nextID:  1, // reserve 0 for mempoolReactor.BroadcastTx
 	}
 	memR.BaseReactor = *p2p.NewBaseReactor("MempoolReactor", memR)
 	return memR
@@ -67,11 +78,18 @@ func (memR *MempoolReactor) GetChannels() []*p2p.ChannelDescriptor {
 // AddPeer implements Reactor.
 // It starts a broadcast routine ensuring all txs are forwarded to the given peer.
 func (memR *MempoolReactor) AddPeer(peer p2p.Peer) {
+	memR.idMtx.Lock()
+	memR.peerMap[peer.ID()] = memR.nextID
+	memR.nextID++
+	memR.idMtx.Unlock()
 	go memR.broadcastTxRoutine(peer)
 }
 
 // RemovePeer implements Reactor.
 func (memR *MempoolReactor) RemovePeer(peer p2p.Peer, reason interface{}) {
+	memR.idMtx.Lock()
+	delete(memR.peerMap, peer.ID())
+	memR.idMtx.Unlock()
 	// broadcast routine checks if peer is gone and returns
 }
 
@@ -88,7 +106,10 @@ func (memR *MempoolReactor) Receive(chID byte, src p2p.Peer, msgBytes []byte) {
 
 	switch msg := msg.(type) {
 	case *TxMessage:
-		err := memR.Mempool.CheckTx(msg.Tx, nil)
+		memR.idMtx.RLock()
+		peerID := memR.peerMap[src.ID()]
+		memR.idMtx.RUnlock()
+		err := memR.Mempool.checkTxFromPeer(msg.Tx, nil, peerID)
 		if err != nil {
 			memR.Logger.Info("Could not check tx", "tx", TxID(msg.Tx), "err", err)
 		}
@@ -114,6 +135,9 @@ func (memR *MempoolReactor) broadcastTxRoutine(peer p2p.Peer) {
 		return
 	}
 
+	memR.idMtx.RLock()
+	peerID := memR.peerMap[peer.ID()]
+	memR.idMtx.RUnlock()
 	var next *clist.CElement
 	for {
 		// This happens because the CElement we were looking at got garbage
@@ -143,6 +167,18 @@ func (memR *MempoolReactor) broadcastTxRoutine(peer p2p.Peer) {
 				continue
 			}
 		}
+		// ensure peer hasn't already sent us this tx
+		skip := false
+		for i := 0; i < len(memTx.senders); i++ {
+			if peerID == memTx.senders[i] {
+				next = next.Next()
+				skip = true
+				break
+			}
+		}
+		if skip {
+			continue
+		}
 		// send memTx
 		msg := &TxMessage{Tx: memTx.tx}
 		success := peer.Send(MempoolChannel, cdc.MustMarshalBinaryBare(msg))