mempool,rpc: add removetx rpc method

CHANGELOG_PENDING.md

@@ -22,6 +22,8 @@ Special thanks to external contributors on this release:
 
 ### FEATURES
 
+- [mempool, rpc] \#7041  Add removeTx operation to the RPC layer. (@tychoish)
+
 ### IMPROVEMENTS
 
 ### BUG FIXES

internal/consensus/replay_stubs.go

@@ -24,6 +24,7 @@ func (emptyMempool) Size() int { return 0 }
 func (emptyMempool) CheckTx(_ context.Context, _ types.Tx, _ func(*abci.Response), _ mempool.TxInfo) error {
 	return nil
 }
+func (emptyMempool) RemoveTxByKey(txKey types.TxKey) error   { return nil }
 func (emptyMempool) ReapMaxBytesMaxGas(_, _ int64) types.Txs { return types.Txs{} }
 func (emptyMempool) ReapMaxTxs(n int) types.Txs              { return types.Txs{} }
 func (emptyMempool) Update(

internal/mempool/cache.go

@@ -31,14 +31,14 @@ var _ TxCache = (*LRUTxCache)(nil)
 type LRUTxCache struct {
 	mtx      tmsync.Mutex
 	size     int
-	cacheMap map[[TxKeySize]byte]*list.Element
+	cacheMap map[types.TxKey]*list.Element
 	list     *list.List
 }
 
 func NewLRUTxCache(cacheSize int) *LRUTxCache {
 	return &LRUTxCache{
 		size:     cacheSize,
-		cacheMap: make(map[[TxKeySize]byte]*list.Element, cacheSize),
+		cacheMap: make(map[types.TxKey]*list.Element, cacheSize),
 		list:     list.New(),
 	}
 }
@@ -53,15 +53,15 @@ func (c *LRUTxCache) Reset() {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
 
-	c.cacheMap = make(map[[TxKeySize]byte]*list.Element, c.size)
+	c.cacheMap = make(map[types.TxKey]*list.Element, c.size)
 	c.list.Init()
 }
 
 func (c *LRUTxCache) Push(tx types.Tx) bool {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
 
-	key := TxKey(tx)
+	key := tx.Key()
 
 	moved, ok := c.cacheMap[key]
 	if ok {
@@ -72,7 +72,7 @@ func (c *LRUTxCache) Push(tx types.Tx) bool {
 	if c.list.Len() >= c.size {
 		front := c.list.Front()
 		if front != nil {
-			frontKey := front.Value.([TxKeySize]byte)
+			frontKey := front.Value.(types.TxKey)
 			delete(c.cacheMap, frontKey)
 			c.list.Remove(front)
 		}
@@ -88,7 +88,7 @@ func (c *LRUTxCache) Remove(tx types.Tx) {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
 
-	key := TxKey(tx)
+	key := tx.Key()
 	e := c.cacheMap[key]
 	delete(c.cacheMap, key)
 

internal/mempool/mempool.go

@@ -32,6 +32,10 @@ type Mempool interface {
 	// its validity and whether it should be added to the mempool.
 	CheckTx(ctx context.Context, tx types.Tx, callback func(*abci.Response), txInfo TxInfo) error
 
+	// RemoveTxByKey removes a transaction, identified by its key,
+	// from the mempool.
+	RemoveTxByKey(txKey types.TxKey) error
+
 	// ReapMaxBytesMaxGas reaps transactions from the mempool up to maxBytes
 	// bytes total with the condition that the total gasWanted must be less than
 	// maxGas.

internal/mempool/mock/mempool.go

@@ -20,6 +20,7 @@ func (Mempool) Size() int { return 0 }
 func (Mempool) CheckTx(_ context.Context, _ types.Tx, _ func(*abci.Response), _ mempool.TxInfo) error {
 	return nil
 }
+func (Mempool) RemoveTxByKey(txKey types.TxKey) error   { return nil }
 func (Mempool) ReapMaxBytesMaxGas(_, _ int64) types.Txs { return types.Txs{} }
 func (Mempool) ReapMaxTxs(n int) types.Txs              { return types.Txs{} }
 func (Mempool) Update(

internal/mempool/tx.go

@@ -1,24 +1,9 @@
 package mempool
 
 import (
-	"crypto/sha256"
-
 	"github.com/tendermint/tendermint/types"
 )
 
-// TxKeySize defines the size of the transaction's key used for indexing.
-const TxKeySize = sha256.Size
-
-// TxKey is the fixed length array key used as an index.
-func TxKey(tx types.Tx) [TxKeySize]byte {
-	return sha256.Sum256(tx)
-}
-
-// TxHashFromBytes returns the hash of a transaction from raw bytes.
-func TxHashFromBytes(tx []byte) []byte {
-	return types.Tx(tx).Hash()
-}
-
 // TxInfo are parameters that get passed when attempting to add a tx to the
 // mempool.
 type TxInfo struct {

internal/mempool/v0/cache_test.go

@@ -61,7 +61,7 @@ func TestCacheAfterUpdate(t *testing.T) {
 			require.NotEqual(t, len(tc.txsInCache), counter,
 				"cache larger than expected on testcase %d", tcIndex)
 
-			nodeVal := node.Value.([sha256.Size]byte)
+			nodeVal := node.Value.(types.TxKey)
 			expectedBz := sha256.Sum256([]byte{byte(tc.txsInCache[len(tc.txsInCache)-counter-1])})
 			// Reference for reading the errors:
 			// >>> sha256('\x00').hexdigest()
@@ -71,7 +71,7 @@ func TestCacheAfterUpdate(t *testing.T) {
 			// >>> sha256('\x02').hexdigest()
 			// 'dbc1b4c900ffe48d575b5da5c638040125f65db0fe3e24494b76ea986457d986'
 
-			require.Equal(t, expectedBz, nodeVal, "Equality failed on index %d, tc %d", counter, tcIndex)
+			require.EqualValues(t, expectedBz, nodeVal, "Equality failed on index %d, tc %d", counter, tcIndex)
 			counter++
 			node = node.Next()
 		}

internal/mempool/v0/clist_mempool.go

@@ -3,6 +3,7 @@ package v0
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -240,7 +241,7 @@ func (mem *CListMempool) CheckTx(
 		// Note it's possible a tx is still in the cache but no longer in the mempool
 		// (eg. after committing a block, txs are removed from mempool but not cache),
 		// so we only record the sender for txs still in the mempool.
-		if e, ok := mem.txsMap.Load(mempool.TxKey(tx)); ok {
+		if e, ok := mem.txsMap.Load(tx.Key()); ok {
 			memTx := e.(*clist.CElement).Value.(*mempoolTx)
 			_, loaded := memTx.senders.LoadOrStore(txInfo.SenderID, true)
 			// TODO: consider punishing peer for dups,
@@ -327,7 +328,7 @@ func (mem *CListMempool) reqResCb(
 //  - resCbFirstTime (lock not held) if tx is valid
 func (mem *CListMempool) addTx(memTx *mempoolTx) {
 	e := mem.txs.PushBack(memTx)
-	mem.txsMap.Store(mempool.TxKey(memTx.tx), e)
+	mem.txsMap.Store(memTx.tx.Key(), e)
 	atomic.AddInt64(&mem.txsBytes, int64(len(memTx.tx)))
 	mem.metrics.TxSizeBytes.Observe(float64(len(memTx.tx)))
 }
@@ -338,7 +339,7 @@ func (mem *CListMempool) addTx(memTx *mempoolTx) {
 func (mem *CListMempool) removeTx(tx types.Tx, elem *clist.CElement, removeFromCache bool) {
 	mem.txs.Remove(elem)
 	elem.DetachPrev()
-	mem.txsMap.Delete(mempool.TxKey(tx))
+	mem.txsMap.Delete(tx.Key())
 	atomic.AddInt64(&mem.txsBytes, int64(-len(tx)))
 
 	if removeFromCache {
@@ -347,13 +348,16 @@ func (mem *CListMempool) removeTx(tx types.Tx, elem *clist.CElement, removeFromC
 }
 
 // RemoveTxByKey removes a transaction from the mempool by its TxKey index.
-func (mem *CListMempool) RemoveTxByKey(txKey [mempool.TxKeySize]byte, removeFromCache bool) {
+func (mem *CListMempool) RemoveTxByKey(txKey types.TxKey) error {
 	if e, ok := mem.txsMap.Load(txKey); ok {
 		memTx := e.(*clist.CElement).Value.(*mempoolTx)
 		if memTx != nil {
-			mem.removeTx(memTx.tx, e.(*clist.CElement), removeFromCache)
+			mem.removeTx(memTx.tx, e.(*clist.CElement), false)
+			return nil
 		}
+		return errors.New("transaction not found")
 	}
+	return errors.New("invalid transaction found")
 }
 
 func (mem *CListMempool) isFull(txSize int) error {
@@ -409,7 +413,7 @@ func (mem *CListMempool) resCbFirstTime(
 			mem.addTx(memTx)
 			mem.logger.Debug(
 				"added good transaction",
-				"tx", mempool.TxHashFromBytes(tx),
+				"tx", types.Tx(tx).Hash(),
 				"res", r,
 				"height", memTx.height,
 				"total", mem.Size(),
@@ -419,7 +423,7 @@ func (mem *CListMempool) resCbFirstTime(
 			// ignore bad transaction
 			mem.logger.Debug(
 				"rejected bad transaction",
-				"tx", mempool.TxHashFromBytes(tx),
+				"tx", types.Tx(tx).Hash(),
 				"peerID", peerP2PID,
 				"res", r,
 				"err", postCheckErr,
@@ -460,7 +464,7 @@ func (mem *CListMempool) resCbRecheck(req *abci.Request, res *abci.Response) {
 			// Good, nothing to do.
 		} else {
 			// Tx became invalidated due to newly committed block.
-			mem.logger.Debug("tx is no longer valid", "tx", mempool.TxHashFromBytes(tx), "res", r, "err", postCheckErr)
+			mem.logger.Debug("tx is no longer valid", "tx", types.Tx(tx).Hash(), "res", r, "err", postCheckErr)
 			// NOTE: we remove tx from the cache because it might be good later
 			mem.removeTx(tx, mem.recheckCursor, !mem.config.KeepInvalidTxsInCache)
 		}
@@ -598,7 +602,7 @@ func (mem *CListMempool) Update(
 		// Mempool after:
 		//   100
 		// https://github.com/tendermint/tendermint/issues/3322.
-		if e, ok := mem.txsMap.Load(mempool.TxKey(tx)); ok {
+		if e, ok := mem.txsMap.Load(tx.Key()); ok {
 			mem.removeTx(tx, e.(*clist.CElement), false)
 		}
 	}

internal/mempool/v0/clist_mempool_test.go

@@ -544,9 +544,9 @@ func TestMempoolTxsBytes(t *testing.T) {
 	err = mp.CheckTx(context.Background(), []byte{0x06}, nil, mempool.TxInfo{})
 	require.NoError(t, err)
 	assert.EqualValues(t, 9, mp.SizeBytes())
-	mp.RemoveTxByKey(mempool.TxKey([]byte{0x07}), true)
+	assert.Error(t, mp.RemoveTxByKey(types.Tx([]byte{0x07}).Key()))
 	assert.EqualValues(t, 9, mp.SizeBytes())
-	mp.RemoveTxByKey(mempool.TxKey([]byte{0x06}), true)
+	assert.NoError(t, mp.RemoveTxByKey(types.Tx([]byte{0x06}).Key()))
 	assert.EqualValues(t, 8, mp.SizeBytes())
 
 }

internal/mempool/v0/reactor.go

@@ -171,7 +171,7 @@ func (r *Reactor) handleMempoolMessage(envelope p2p.Envelope) error {
 
 		for _, tx := range protoTxs {
 			if err := r.mempool.CheckTx(context.Background(), types.Tx(tx), nil, txInfo); err != nil {
-				logger.Error("checktx failed for tx", "tx", fmt.Sprintf("%X", mempool.TxHashFromBytes(tx)), "err", err)
+				logger.Error("checktx failed for tx", "tx", fmt.Sprintf("%X", types.Tx(tx).Hash()), "err", err)
 			}
 		}
 
@@ -378,7 +378,7 @@ func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer)
 			}
 			r.Logger.Debug(
 				"gossiped tx to peer",
-				"tx", fmt.Sprintf("%X", mempool.TxHashFromBytes(memTx.tx)),
+				"tx", fmt.Sprintf("%X", memTx.tx.Hash()),
 				"peer", peerID,
 			)
 		}

internal/mempool/v1/mempool.go

@@ -3,6 +3,7 @@ package v1
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"sync/atomic"
 	"time"
@@ -256,7 +257,7 @@ func (txmp *TxMempool) CheckTx(
 		return err
 	}
 
-	txHash := mempool.TxKey(tx)
+	txHash := tx.Key()
 
 	// We add the transaction to the mempool's cache and if the transaction already
 	// exists, i.e. false is returned, then we check if we've seen this transaction
@@ -304,6 +305,19 @@ func (txmp *TxMempool) CheckTx(
 	return nil
 }
 
+func (txmp *TxMempool) RemoveTxByKey(txKey types.TxKey) error {
+	txmp.Lock()
+	defer txmp.Unlock()
+
+	// remove the committed transaction from the transaction store and indexes
+	if wtx := txmp.txStore.GetTxByHash(txKey); wtx != nil {
+		txmp.removeTx(wtx, false)
+		return nil
+	}
+
+	return errors.New("transaction not found")
+}
+
 // Flush flushes out the mempool. It acquires a read-lock, fetches all the
 // transactions currently in the transaction store and removes each transaction
 // from the store and all indexes and finally resets the cache.
@@ -451,7 +465,7 @@ func (txmp *TxMempool) Update(
 		}
 
 		// remove the committed transaction from the transaction store and indexes
-		if wtx := txmp.txStore.GetTxByHash(mempool.TxKey(tx)); wtx != nil {
+		if wtx := txmp.txStore.GetTxByHash(tx.Key()); wtx != nil {
 			txmp.removeTx(wtx, false)
 		}
 	}
@@ -629,7 +643,7 @@ func (txmp *TxMempool) defaultTxCallback(req *abci.Request, res *abci.Response)
 		tx := req.GetCheckTx().Tx
 		wtx := txmp.recheckCursor.Value.(*WrappedTx)
 		if !bytes.Equal(tx, wtx.tx) {
-			panic(fmt.Sprintf("re-CheckTx transaction mismatch; got: %X, expected: %X", wtx.tx.Hash(), mempool.TxKey(tx)))
+			panic(fmt.Sprintf("re-CheckTx transaction mismatch; got: %X, expected: %X", wtx.tx.Hash(), types.Tx(tx).Key()))
 		}
 
 		// Only evaluate transactions that have not been removed. This can happen
@@ -647,7 +661,7 @@ func (txmp *TxMempool) defaultTxCallback(req *abci.Request, res *abci.Response)
 				txmp.logger.Debug(
 					"existing transaction no longer valid; failed re-CheckTx callback",
 					"priority", wtx.priority,
-					"tx", fmt.Sprintf("%X", mempool.TxHashFromBytes(wtx.tx)),
+					"tx", fmt.Sprintf("%X", wtx.tx.Hash()),
 					"err", err,
 					"code", checkTxRes.CheckTx.Code,
 				)
@@ -784,13 +798,13 @@ func (txmp *TxMempool) removeTx(wtx *WrappedTx, removeFromCache bool) {
 // the height and time based indexes.
 func (txmp *TxMempool) purgeExpiredTxs(blockHeight int64) {
 	now := time.Now()
-	expiredTxs := make(map[[mempool.TxKeySize]byte]*WrappedTx)
+	expiredTxs := make(map[types.TxKey]*WrappedTx)
 
 	if txmp.config.TTLNumBlocks > 0 {
 		purgeIdx := -1
 		for i, wtx := range txmp.heightIndex.txs {
 			if (blockHeight - wtx.height) > txmp.config.TTLNumBlocks {
-				expiredTxs[mempool.TxKey(wtx.tx)] = wtx
+				expiredTxs[wtx.tx.Key()] = wtx
 				purgeIdx = i
 			} else {
 				// since the index is sorted, we know no other txs can be be purged
@@ -807,7 +821,7 @@ func (txmp *TxMempool) purgeExpiredTxs(blockHeight int64) {
 		purgeIdx := -1
 		for i, wtx := range txmp.timestampIndex.txs {
 			if now.Sub(wtx.timestamp) > txmp.config.TTLDuration {
-				expiredTxs[mempool.TxKey(wtx.tx)] = wtx
+				expiredTxs[wtx.tx.Key()] = wtx
 				purgeIdx = i
 			} else {
 				// since the index is sorted, we know no other txs can be be purged

internal/mempool/v1/mempool_test.go

@@ -226,10 +226,10 @@ func TestTxMempool_ReapMaxBytesMaxGas(t *testing.T) {
 	require.Equal(t, len(tTxs), txmp.Size())
 	require.Equal(t, int64(5690), txmp.SizeBytes())
 
-	txMap := make(map[[mempool.TxKeySize]byte]testTx)
+	txMap := make(map[types.TxKey]testTx)
 	priorities := make([]int64, len(tTxs))
 	for i, tTx := range tTxs {
-		txMap[mempool.TxKey(tTx.tx)] = tTx
+		txMap[tTx.tx.Key()] = tTx
 		priorities[i] = tTx.priority
 	}
 
@@ -241,7 +241,7 @@ func TestTxMempool_ReapMaxBytesMaxGas(t *testing.T) {
 	ensurePrioritized := func(reapedTxs types.Txs) {
 		reapedPriorities := make([]int64, len(reapedTxs))
 		for i, rTx := range reapedTxs {
-			reapedPriorities[i] = txMap[mempool.TxKey(rTx)].priority
+			reapedPriorities[i] = txMap[rTx.Key()].priority
 		}
 
 		require.Equal(t, priorities[:len(reapedPriorities)], reapedPriorities)
@@ -276,10 +276,10 @@ func TestTxMempool_ReapMaxTxs(t *testing.T) {
 	require.Equal(t, len(tTxs), txmp.Size())
 	require.Equal(t, int64(5690), txmp.SizeBytes())
 
-	txMap := make(map[[mempool.TxKeySize]byte]testTx)
+	txMap := make(map[types.TxKey]testTx)
 	priorities := make([]int64, len(tTxs))
 	for i, tTx := range tTxs {
-		txMap[mempool.TxKey(tTx.tx)] = tTx
+		txMap[tTx.tx.Key()] = tTx
 		priorities[i] = tTx.priority
 	}
 
@@ -291,7 +291,7 @@ func TestTxMempool_ReapMaxTxs(t *testing.T) {
 	ensurePrioritized := func(reapedTxs types.Txs) {
 		reapedPriorities := make([]int64, len(reapedTxs))
 		for i, rTx := range reapedTxs {
-			reapedPriorities[i] = txMap[mempool.TxKey(rTx)].priority
+			reapedPriorities[i] = txMap[rTx.Key()].priority
 		}
 
 		require.Equal(t, priorities[:len(reapedPriorities)], reapedPriorities)