Repository: celestiaorg/celestia-core (go)

Issue:
Time based TTL should only happen in Update method
Currently, using a time based TTL, leads to deletions in the mempool at random times. This is problematic because the chechTxState in the app hasn't been updated with that eviction and thus is out of sync with the transactions in the mempool. We should only allow deletions to happen during the `Update` method when the checkTxState is reset and we replay all transactions in the mempool

Pull request:
fix: only purge expired txs during update
Fixes #2195 



Patch:
diff --git a/mempool/cat/pool.go b/mempool/cat/pool.go
index a26d3802f..ecd7373f4 100644
--- a/mempool/cat/pool.go
+++ b/mempool/cat/pool.go
@@ -204,26 +204,6 @@ func (txmp *TxPool) WasRecentlyRejected(txKey types.TxKey) bool {
 	return txmp.rejectedTxCache.Has(txKey)
 }
 
-// CheckToPurgeExpiredTxs checks if there has been adequate time since the last time
-// the txpool looped through all transactions and if so, performs a purge of any transaction
-// that has expired according to the TTLDuration. This is thread safe.
-func (txmp *TxPool) CheckToPurgeExpiredTxs() {
-	txmp.mtx.Lock()
-	defer txmp.mtx.Unlock()
-	if txmp.config.TTLDuration > 0 && time.Since(txmp.lastPurgeTime) > txmp.config.TTLDuration {
-		expirationAge := time.Now().Add(-txmp.config.TTLDuration)
-		// A height of 0 means no transactions will be removed because of height
-		// (in other words, no transaction has a height less than 0)
-		purgedTxs, numExpired := txmp.store.purgeExpiredTxs(0, expirationAge)
-		// Add the purged transactions to the evicted cache
-		for _, tx := range purgedTxs {
-			txmp.evictedTxCache.Push(tx.key())
-		}
-		txmp.metrics.EvictedTxs.Add(float64(numExpired))
-		txmp.lastPurgeTime = time.Now()
-	}
-}
-
 // CheckTx adds the given transaction to the mempool if it fits and passes the
 // application's ABCI CheckTx method. This should be viewed as the entry method for new transactions
 // into the network. In practice this happens via an RPC endpoint
@@ -500,6 +480,8 @@ func (txmp *TxPool) Update(
 		txmp.removeTxByKey(tx.Key())
 	}
 
+	// Purge expired transactions based on TTL. This is the only place where
+	// TTL purging should happen to maintain checkTxState consistency.
 	txmp.purgeExpiredTxs(blockHeight)
 
 	// If there any uncommitted transactions left in the mempool, we either
diff --git a/mempool/cat/reactor.go b/mempool/cat/reactor.go
index 65a97dea1..9b0e0223a 100644
--- a/mempool/cat/reactor.go
+++ b/mempool/cat/reactor.go
@@ -121,20 +121,6 @@ func (memR *Reactor) OnStart() error {
 	} else {
 		memR.Logger.Info("Tx broadcasting is disabled")
 	}
-	// run a separate go routine to check for time based TTLs
-	if memR.mempool.config.TTLDuration > 0 {
-		go func() {
-			ticker := time.NewTicker(memR.mempool.config.TTLDuration)
-			for {
-				select {
-				case <-ticker.C:
-					memR.mempool.CheckToPurgeExpiredTxs()
-				case <-memR.Quit():
-					return
-				}
-			}
-		}()
-	}
 
 	return nil
 }
diff --git a/mempool/priority/mempool.go b/mempool/priority/mempool.go
index 8adc5df8c..21abb7b0b 100644
--- a/mempool/priority/mempool.go
+++ b/mempool/priority/mempool.go
@@ -67,7 +67,6 @@ func NewTxMempool(
 	height int64,
 	options ...TxMempoolOption,
 ) *TxMempool {
-
 	txmp := &TxMempool{
 		logger:       logger,
 		config:       cfg,
@@ -446,6 +445,8 @@ func (txmp *TxMempool) Update(
 		_ = txmp.removeTxByKey(tx.Key())
 	}
 
+	// Purge expired transactions based on TTL. This is the only place where
+	// TTL purging should happen to maintain checkTxState consistency.
 	txmp.purgeExpiredTxs(blockHeight)
 
 	// If there any uncommitted transactions left in the mempool, we either
@@ -735,17 +736,6 @@ func (txmp *TxMempool) canAddTx(wtx *WrappedTx) error {
 	return nil
 }
 
-// CheckToPurgeExpiredTxs checks if there has been adequate time since the last time
-// the txpool looped through all transactions and if so, performs a purge of any transaction
-// that has expired according to the TTLDuration. This is thread safe.
-func (txmp *TxMempool) CheckToPurgeExpiredTxs() {
-	txmp.mtx.Lock()
-	defer txmp.mtx.Unlock()
-	if txmp.config.TTLDuration > 0 && time.Since(txmp.lastPurgeTime) > txmp.config.TTLDuration {
-		txmp.purgeExpiredTxs(txmp.height)
-	}
-}
-
 // purgeExpiredTxs removes all transactions from the mempool that have exceeded
 // their respective height or time-based limits as of the given blockHeight.
 // Transactions removed by this operation are not removed from the cache.
diff --git a/mempool/priority/reactor.go b/mempool/priority/reactor.go
index 39688eb70..880b11dbe 100644
--- a/mempool/priority/reactor.go
+++ b/mempool/priority/reactor.go
@@ -116,21 +116,6 @@ func (memR *Reactor) OnStart() error {
 		memR.Logger.Info("Tx broadcasting is disabled")
 	}
 
-	// run a separate go routine to check for time based TTLs
-	if memR.mempool.config.TTLDuration > 0 {
-		go func() {
-			ticker := time.NewTicker(memR.mempool.config.TTLDuration)
-			for {
-				select {
-				case <-ticker.C:
-					memR.mempool.CheckToPurgeExpiredTxs()
-				case <-memR.Quit():
-					return
-				}
-			}
-		}()
-	}
-
 	return nil
 }
 


Test patch:
diff --git a/mempool/cat/reactor_test.go b/mempool/cat/reactor_test.go
index a3edff435..f08c3c7bc 100644
--- a/mempool/cat/reactor_test.go
+++ b/mempool/cat/reactor_test.go
@@ -229,34 +229,6 @@ func TestMempoolVectors(t *testing.T) {
 	}
 }
 
-func TestReactorEventuallyRemovesExpiredTransaction(t *testing.T) {
-	reactor, _ := setupReactor(t)
-	reactor.mempool.config.TTLDuration = 200 * time.Millisecond
-
-	tx := newDefaultTx("hello")
-	key := tx.Key()
-	txMsg := &protomem.Txs{Txs: [][]byte{tx}}
-
-	peer := genPeer()
-	require.NoError(t, reactor.Start())
-	reactor.InitPeer(peer)
-	reactor.Receive(
-		p2p.Envelope{
-			ChannelID: mempool.MempoolChannel,
-			Message:   txMsg,
-			Src:       peer,
-		},
-	)
-	require.True(t, reactor.mempool.Has(key))
-
-	// wait for the transaction to expire
-	require.Eventually(t,
-		func() bool { return !reactor.mempool.Has(key) },
-		4*reactor.mempool.config.TTLDuration,
-		50*time.Millisecond,
-		"transaction was not removed after TTL expired")
-}
-
 func TestLegacyReactorReceiveBasic(t *testing.T) {
 	config := cfg.TestConfig()
 	// if there were more than two reactors, the order of transactions could not be
diff --git a/mempool/priority/mempool_test.go b/mempool/priority/mempool_test.go
index 548802154..6761e2435 100644
--- a/mempool/priority/mempool_test.go
+++ b/mempool/priority/mempool_test.go
@@ -793,7 +793,7 @@ func TestConcurrentCheckTxDataRace(t *testing.T) {
 		// simultaneously, which should trigger the race condition
 		numGoroutines := 100
 		var startWg sync.WaitGroup
-		var startCh = make(chan struct{})
+		startCh := make(chan struct{})
 
 		for i := 0; i < numGoroutines; i++ {
 			startWg.Add(1)
diff --git a/mempool/priority/reactor_test.go b/mempool/priority/reactor_test.go
index 3a663be99..13f3dcdce 100644
--- a/mempool/priority/reactor_test.go
+++ b/mempool/priority/reactor_test.go
@@ -95,41 +95,6 @@ func TestMempoolVectors(t *testing.T) {
 	}
 }
 
-func TestReactorEventuallyRemovesExpiredTransaction(t *testing.T) {
-	config := cfg.TestConfig()
-	config.Mempool.TTLDuration = 200 * time.Millisecond
-	const N = 1
-	reactor := makeAndConnectReactors(config, N)[0]
-
-	tx := types.Tx("0x00=0x00=123")
-	key := tx.Key()
-	txMsg := &memproto.Txs{Txs: [][]byte{tx}}
-
-	peer := mock.NewPeer(nil)
-	reactor.InitPeer(peer)
-	reactor.Receive(p2p.Envelope{
-		Src:       peer,
-		Message:   txMsg,
-		ChannelID: mempool.MempoolChannel,
-	})
-	reactor.mempool.Lock()
-	_, has := reactor.mempool.txByKey[key]
-	reactor.mempool.Unlock()
-	require.True(t, has)
-
-	// wait for the transaction to expire
-	require.Eventually(t,
-		func() bool {
-			reactor.mempool.Lock()
-			_, has := reactor.mempool.txByKey[key]
-			reactor.mempool.Unlock()
-			return has
-		},
-		4*reactor.mempool.config.TTLDuration,
-		50*time.Millisecond,
-		"transaction was not removed after TTL expired")
-}
-
 func TestLegacyReactorReceiveBasic(t *testing.T) {
 	config := cfg.TestConfig()
 	// if there were more than two reactors, the order of transactions could not be
@@ -177,7 +142,6 @@ func makeAndConnectReactors(config *cfg.Config, n int) []*Reactor {
 	p2p.MakeConnectedSwitches(config.P2P, n, func(i int, s *p2p.Switch) *p2p.Switch {
 		s.AddReactor("MEMPOOL", reactors[i])
 		return s
-
 	}, p2p.Connect2Switches)
 	return reactors
 }