Merge pull request #2987 from jorgemmsilva/fix/ttl-access-nodes

fix: force mempool TTL cleanup on access nodes
iotaledger · Oct 24, 2023 · b3d25f5 · b3d25f5
2 parents bb15580 + d531e98
commit b3d25f5
Showing 1 changed file with 16 additions and 0 deletions.
diff --git a/packages/chain/mempool/mempool.go b/packages/chain/mempool/mempool.go
@@ -73,6 +73,7 @@ const (
 	distShareMaxMsgsPerTick = 100
 	distShareRePublishTick  = 5 * time.Second
 	waitRequestCleanupEvery = 10
+	forceCleanMempoolTick   = 1 * time.Minute
 )
 
 // Partial interface for providing chain events to the outside.
@@ -377,6 +378,7 @@ func (mpi *mempoolImpl) run(ctx context.Context, cleanupFunc context.CancelFunc)
 	debugTicker := time.NewTicker(distShareDebugTick)
 	timeTicker := time.NewTicker(distShareTimeTick)
 	rePublishTicker := time.NewTicker(distShareRePublishTick)
+	forceCleanMempoolTicker := time.NewTicker(forceCleanMempoolTick)
 	for {
 		select {
 		case recv, ok := <-serverNodesUpdatedPipeOutCh:
@@ -397,6 +399,7 @@ func (mpi *mempoolImpl) run(ctx context.Context, cleanupFunc context.CancelFunc)
 				break
 			}
 			mpi.handleConsensusProposal(recv)
+			forceCleanMempoolTicker.Reset(forceCleanMempoolTick) // mempool will be forcebly cleanup if this ticker triggers
 		case recv, ok := <-reqConsensusRequestsPipeOutCh:
 			if !ok {
 				reqConsensusRequestsPipeOutCh = nil
@@ -444,6 +447,8 @@ func (mpi *mempoolImpl) run(ctx context.Context, cleanupFunc context.CancelFunc)
 			mpi.handleDistSyncTimeTick()
 		case <-rePublishTicker.C:
 			mpi.handleRePublishTimeTick()
+		case <-forceCleanMempoolTicker.C:
+			mpi.handleForceCleanMempool()
 		case <-ctx.Done():
 			// mpi.serverNodesUpdatedPipe.Close() // TODO: Causes panic: send on closed channel
 			// mpi.accessNodesUpdatedPipe.Close()
@@ -893,6 +898,17 @@ func (mpi *mempoolImpl) handleRePublishTimeTick() {
 	// })
 }
 
+func (mpi *mempoolImpl) handleForceCleanMempool() {
+	mpi.offLedgerPool.Iterate(func(account string, entries []*OrderedPoolEntry[isc.OffLedgerRequest]) {
+		for _, e := range entries {
+			if time.Since(e.ts) > mpi.ttl && !lo.Some(mpi.consensusInstances, e.proposedFor) {
+				mpi.log.Debugf("handleForceCleanMempool, request TTL expired, removing: %s", e.req.ID().String())
+				mpi.offLedgerPool.Remove(e.req)
+			}
+		}
+	})
+}
+
 func (mpi *mempoolImpl) tryReAddRequest(req isc.Request) {
 	switch req := req.(type) {
 	case isc.OnLedgerRequest: