Skip to content

Supernova TxPool Part2: Fixed eviction consistency #7352

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
danielradu10 merged 8 commits into from
Oct 22, 2025
Merged

Supernova TxPool Part2: Fixed eviction consistency #7352

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
090ee54
fix eviction consistency
danielradu10 Oct 17, 2025
4216d71
fixes and bug fix
danielradu10 Oct 20, 2025
427145b
added a TODO
danielradu10 Oct 20, 2025
bf23bf6
small fix
danielradu10 Oct 20, 2025
dd489da
Merge branch 'feat/mempool-supernova-part2' into fix-eviction-consist…
danielradu10 Oct 20, 2025
f7686e9
added nil check
danielradu10 Oct 21, 2025
7afad8a
fixed race conditions
danielradu10 Oct 21, 2025
3a5548b
added unit tests
danielradu10 Oct 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 13 additions & 11 deletions txcache/autoClean.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,14 +79,11 @@ func (cache *TxCache) RemoveSweepableTxs(accountsProvider common.AccountNoncePro
// we want to remove transactions with nonces < lastCommittedNonce
targetNonce := accountNonce - 1

evicted = append(evicted, sender.removeSweepableTransactionsReturnHashes(targetNonce)...)
evicted = append(evicted, sender.removeSweepableTransactionsReturnHashes(targetNonce, cache.tracker)...)
}

if len(evicted) > 0 {
txs := cache.txByHash.GetTxsBulk(evicted)

untrackedEvicted := cache.tracker.GetBulkOfUntrackedTransactions(txs)
cache.txByHash.RemoveTxsBulk(untrackedEvicted)
cache.txByHash.RemoveTxsBulk(evicted)
}

logRemove.Debug("TxCache.RemoveSweepableTxs end",
Expand Down Expand Up @@ -134,7 +131,7 @@ func shuffleSendersAddresses(senders []string, randomness uint64) {
})
}

func (listForSender *txListForSender) removeSweepableTransactionsReturnHashes(targetNonce uint64) [][]byte {
func (listForSender *txListForSender) removeSweepableTransactionsReturnHashes(targetNonce uint64, tracker *selectionTracker) [][]byte {
txHashesToEvict := make([][]byte, 0)

// We don't allow concurrent goroutines to mutate a given sender's list
Expand All @@ -144,6 +141,7 @@ func (listForSender *txListForSender) removeSweepableTransactionsReturnHashes(ta
for element := listForSender.items.Front(); element != nil; {
// finds transactions with lower nonces
tx := element.Value.(*WrappedTransaction)

txNonce := tx.Tx.GetNonce()

// nonces are sorted ascending, so we can stop as soon as we find a nonce that is higher
Expand All @@ -158,12 +156,16 @@ func (listForSender *txListForSender) removeSweepableTransactionsReturnHashes(ta
)

nextElement := element.Next()
_ = listForSender.items.Remove(element)
listForSender.onRemovedListElement(element)
element = nextElement

// Keep track of removed transactions
txHashesToEvict = append(txHashesToEvict, tx.TxHash)
if !tracker.IsTransactionTracked(tx) {
_ = listForSender.items.Remove(element)
Copy link
Contributor Author

@danielradu10 danielradu10 Oct 20, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

remove only if the transaction is not tracked

listForSender.onRemovedListElement(element)

// Keep track of removed transactions
txHashesToEvict = append(txHashesToEvict, tx.TxHash)
}

element = nextElement
Comment on lines +160 to +168
Copy link

Copilot AI Oct 20, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Similar to the previous issue, this logic only removes untracked transactions but doesn't handle tracked transactions that should be swept. If a transaction with a lower nonce is tracked, it will remain in the list even though it should be removed based on the sweeping criteria.

Copilot uses AI. Check for mistakes.
Copy link
Contributor Author

@danielradu10 danielradu10 Oct 20, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, added already a TODO for this discussion.

}

return txHashesToEvict
Expand Down
13 changes: 7 additions & 6 deletions txcache/autoClean_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,16 +88,17 @@ func TestTxCache_GetDeterministicallyShuffledSenders_Dummy(t *testing.T) {

func Test_RemoveSweepableTransactionsReturnHashes_Dummy(t *testing.T) {
t.Run("with lower nonces", func(t *testing.T) {
txCache := newCacheToTest(maxNumBytesPerSenderUpperBoundTest, math.MaxUint32)
list := newUnconstrainedListToTest()
list.AddTx(createTx([]byte("a"), ".", 1))
list.AddTx(createTx([]byte("b"), ".", 3))
list.AddTx(createTx([]byte("c"), ".", 4))
list.AddTx(createTx([]byte("d"), ".", 2))
list.AddTx(createTx([]byte("e"), ".", 5))
list.AddTx(createTx([]byte("a"), ".", 1), txCache.tracker)
list.AddTx(createTx([]byte("b"), ".", 3), txCache.tracker)
list.AddTx(createTx([]byte("c"), ".", 4), txCache.tracker)
list.AddTx(createTx([]byte("d"), ".", 2), txCache.tracker)
list.AddTx(createTx([]byte("e"), ".", 5), txCache.tracker)

hashesBeforeEviction := list.getTxHashesAsStrings()

list.removeSweepableTransactionsReturnHashes(uint64(3))
list.removeSweepableTransactionsReturnHashes(uint64(3), txCache.tracker)
Copy link

@mradian1 mradian1 Oct 21, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

shouldn't we also check autoclean for tracked transactions and also consistency between the 2 maps?

hashesAfterEviction := list.getTxHashesAsStrings()
require.Equal(t, []string{"c", "e"}, hashesAfterEviction)

Expand Down
15 changes: 6 additions & 9 deletions txcache/eviction.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,8 +126,10 @@ func (cache *TxCache) evictLeastLikelyToSelectTransactions() *evictionJournal {
break
}

transactionsToEvict = append(transactionsToEvict, item.currentTransaction)
transactionsToEvictHashes = append(transactionsToEvictHashes, item.currentTransaction.TxHash)
if !cache.tracker.IsTransactionTracked(item.currentTransaction) {
transactionsToEvict = append(transactionsToEvict, item.currentTransaction)
transactionsToEvictHashes = append(transactionsToEvictHashes, item.currentTransaction.TxHash)
}

// If there are more transactions in the same bunch (same sender as the popped item),
// add the next one to the heap (to compete with the others in being "the worst").
Expand Down Expand Up @@ -157,17 +159,12 @@ func (cache *TxCache) evictLeastLikelyToSelectTransactions() *evictionJournal {
}

// Remove those transactions from "txByHash".
txs := cache.txByHash.GetTxsBulk(transactionsToEvictHashes)

logRemove.Debug("evictLeastLikelyToSelectTransactions", "pass", pass, "num evicted", len(transactionsToEvict))

untrackedTransactionsToEvictHashes := cache.tracker.GetBulkOfUntrackedTransactions(txs)
_ = cache.txByHash.RemoveTxsBulk(untrackedTransactionsToEvictHashes)
_ = cache.txByHash.RemoveTxsBulk(transactionsToEvictHashes)

journal.numEvictedByPass = append(journal.numEvictedByPass, len(transactionsToEvict))
journal.numEvicted += len(transactionsToEvict)

logRemove.Debug("evictLeastLikelyToSelectTransactions", "pass", pass, "num untracked evicted", len(untrackedTransactionsToEvictHashes))
logRemove.Debug("evictLeastLikelyToSelectTransactions", "pass", pass, "num evicted", len(transactionsToEvict))
}

return journal
Expand Down
22 changes: 3 additions & 19 deletions txcache/selectionTracker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -555,25 +555,9 @@ func (st *selectionTracker) getVirtualNonceOfAccountWithRootHash(
return breadcrumb.lastNonce.Value + 1, st.latestRootHash, nil
}

// GetBulkOfUntrackedTransactions returns the hashes of the untracked transactions
Copy link
Contributor Author

@danielradu10 danielradu10 Oct 20, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

not used anymore

func (st *selectionTracker) GetBulkOfUntrackedTransactions(transactions []*WrappedTransaction) [][]byte {
untrackedTransactions := make([][]byte, 0)
for _, tx := range transactions {
if tx == nil || tx.Tx == nil {
continue
}

if !st.isTransactionTracked(tx) {
untrackedTransactions = append(untrackedTransactions, tx.TxHash)
}
}

return untrackedTransactions
}

// isTransactionTracked checks if a transaction is still in the tracked blocks of the SelectionTracker.
// However, in the case of forks, isTransactionTracked might return inaccurate results.
func (st *selectionTracker) isTransactionTracked(transaction *WrappedTransaction) bool {
// IsTransactionTracked checks if a transaction is still in the tracked blocks of the SelectionTracker.
// However, in the case of forks, IsTransactionTracked might return inaccurate results.
func (st *selectionTracker) IsTransactionTracked(transaction *WrappedTransaction) bool {
if transaction == nil || transaction.Tx == nil {
return false
}
Expand Down
26 changes: 16 additions & 10 deletions txcache/selectionTracker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1304,8 +1304,8 @@ func Test_isTransactionTracked(t *testing.T) {
tx1 := createTx([]byte("txHash1"), "alice", 11)
tx2 := createTx([]byte("txHash6"), "eve", 11)

require.True(t, txCache.tracker.isTransactionTracked(tx1))
require.True(t, txCache.tracker.isTransactionTracked(tx2))
require.True(t, txCache.tracker.IsTransactionTracked(tx1))
require.True(t, txCache.tracker.IsTransactionTracked(tx2))
})

t.Run("should return false because out of range", func(t *testing.T) {
Expand All @@ -1314,8 +1314,8 @@ func Test_isTransactionTracked(t *testing.T) {
tx1 := createTx([]byte("txHashX"), "alice", 16)
tx2 := createTx([]byte("txHashX"), "eve", 12)

require.False(t, txCache.tracker.isTransactionTracked(tx1))
require.False(t, txCache.tracker.isTransactionTracked(tx2))
require.False(t, txCache.tracker.IsTransactionTracked(tx1))
require.False(t, txCache.tracker.IsTransactionTracked(tx2))

})

Expand All @@ -1324,27 +1324,27 @@ func Test_isTransactionTracked(t *testing.T) {

tx1 := createTx([]byte("txHashX"), "alice", 16)

require.False(t, txCache.tracker.isTransactionTracked(tx1))
require.False(t, txCache.tracker.IsTransactionTracked(tx1))
})

t.Run("should return false because account is not tracked at all", func(t *testing.T) {
t.Parallel()

tx1 := createTx([]byte("txHash2"), "carol", 12)

require.False(t, txCache.tracker.isTransactionTracked(tx1))
require.False(t, txCache.tracker.IsTransactionTracked(tx1))
})

t.Run("should return true for any transaction of sender with a tracked nonce", func(t *testing.T) {
t.Parallel()

tx1 := createTx([]byte("txHash7"), "eve", 12)

require.False(t, txCache.tracker.isTransactionTracked(tx1))
require.False(t, txCache.tracker.IsTransactionTracked(tx1))
})
}

func TestSelectionTracker_GetBulkOfUntrackedTransactions(t *testing.T) {
func TestSelectionTracker_IsTransactionTracked(t *testing.T) {
t.Parallel()

txCache := newCacheToTest(maxNumBytesPerSenderUpperBoundTest, 6)
Expand Down Expand Up @@ -1384,6 +1384,7 @@ func TestSelectionTracker_GetBulkOfUntrackedTransactions(t *testing.T) {
[]byte("txHash1"),
[]byte("txHash2"),
[]byte("txHash3"),
[]byte("txHash6"),
},
},
},
Expand All @@ -1398,8 +1399,13 @@ func TestSelectionTracker_GetBulkOfUntrackedTransactions(t *testing.T) {
)
require.Nil(t, err)

bulk := tracker.GetBulkOfUntrackedTransactions(txs)
require.Len(t, bulk, 4)
require.True(t, txCache.tracker.IsTransactionTracked(txs[0]))
require.True(t, txCache.tracker.IsTransactionTracked(txs[1]))
require.True(t, txCache.tracker.IsTransactionTracked(txs[2]))
require.False(t, txCache.tracker.IsTransactionTracked(txs[3]))
require.False(t, txCache.tracker.IsTransactionTracked(txs[4]))
require.True(t, txCache.tracker.IsTransactionTracked(txs[5]))
require.True(t, txCache.tracker.IsTransactionTracked(txs[6]))
}

func TestSelectionTracker_ResetTracker(t *testing.T) {
Expand Down
9 changes: 2 additions & 7 deletions txcache/txCache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ func (cache *TxCache) AddTx(tx *WrappedTransaction) (ok bool, added bool) {

cache.mutTxOperation.Lock()
addedInByHash := cache.txByHash.addTx(tx)
addedInBySender, evicted := cache.txListBySender.addTxReturnEvicted(tx)
addedInBySender, evicted := cache.txListBySender.addTxReturnEvicted(tx, cache.tracker)
cache.mutTxOperation.Unlock()
if addedInByHash != addedInBySender {
// This can happen when two go-routines concur to add the same transaction:
Expand All @@ -91,12 +91,7 @@ func (cache *TxCache) AddTx(tx *WrappedTransaction) (ok bool, added bool) {

if len(evicted) > 0 {
logRemove.Trace("TxCache.AddTx with eviction", "sender", tx.Tx.GetSndAddr(), "num evicted txs", len(evicted))
txs := cache.txByHash.GetTxsBulk(evicted)

untrackedEvicted := cache.tracker.GetBulkOfUntrackedTransactions(txs)
_ = cache.txByHash.RemoveTxsBulk(untrackedEvicted)

logRemove.Trace("TxCache.AddTx with eviction and tracking check", "sender", tx.Tx.GetSndAddr(), "num evicted txs with tracking check", len(untrackedEvicted))
_ = cache.txByHash.RemoveTxsBulk(evicted)
}

// The return value "added" is true even if transaction added, but then removed due to limits be sender.
Expand Down
88 changes: 74 additions & 14 deletions txcache/txCache_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,21 +117,81 @@ func Test_AddNilTx_DoesNothing(t *testing.T) {
}

func Test_AddTx_AppliesSizeConstraintsPerSenderForNumTransactions(t *testing.T) {
cache := newCacheToTest(maxNumBytesPerSenderUpperBoundTest, 3)
t.Run("with untracked txs", func(t *testing.T) {
t.Parallel()

cache := newCacheToTest(maxNumBytesPerSenderUpperBoundTest, 3)

cache.AddTx(createTx([]byte("tx-alice-1"), "alice", 1))
cache.AddTx(createTx([]byte("tx-alice-2"), "alice", 2))
cache.AddTx(createTx([]byte("tx-alice-4"), "alice", 4))
cache.AddTx(createTx([]byte("tx-bob-1"), "bob", 1))
cache.AddTx(createTx([]byte("tx-bob-2"), "bob", 2))
require.Equal(t, []string{"tx-alice-1", "tx-alice-2", "tx-alice-4"}, cache.getHashesForSender("alice"))
require.Equal(t, []string{"tx-bob-1", "tx-bob-2"}, cache.getHashesForSender("bob"))
require.True(t, cache.areInternalMapsConsistent())

cache.AddTx(createTx([]byte("tx-alice-3"), "alice", 3))
require.Equal(t, []string{"tx-alice-1", "tx-alice-2", "tx-alice-3"}, cache.getHashesForSender("alice"))
require.Equal(t, []string{"tx-bob-1", "tx-bob-2"}, cache.getHashesForSender("bob"))
require.True(t, cache.areInternalMapsConsistent())
})

cache.AddTx(createTx([]byte("tx-alice-1"), "alice", 1))
cache.AddTx(createTx([]byte("tx-alice-2"), "alice", 2))
cache.AddTx(createTx([]byte("tx-alice-4"), "alice", 4))
cache.AddTx(createTx([]byte("tx-bob-1"), "bob", 1))
cache.AddTx(createTx([]byte("tx-bob-2"), "bob", 2))
require.Equal(t, []string{"tx-alice-1", "tx-alice-2", "tx-alice-4"}, cache.getHashesForSender("alice"))
require.Equal(t, []string{"tx-bob-1", "tx-bob-2"}, cache.getHashesForSender("bob"))
require.True(t, cache.areInternalMapsConsistent())
t.Run("with tracked txs", func(t *testing.T) {
t.Parallel()

cache.AddTx(createTx([]byte("tx-alice-3"), "alice", 3))
require.Equal(t, []string{"tx-alice-1", "tx-alice-2", "tx-alice-3"}, cache.getHashesForSender("alice"))
require.Equal(t, []string{"tx-bob-1", "tx-bob-2"}, cache.getHashesForSender("bob"))
require.True(t, cache.areInternalMapsConsistent())
cache := newCacheToTest(maxNumBytesPerSenderUpperBoundTest, 3)

accountsProvider := &txcachemocks.AccountNonceAndBalanceProviderMock{
GetAccountNonceAndBalanceCalled: func(address []byte) (uint64, *big.Int, bool, error) {
return 1, big.NewInt(3 * 1500000 * oneBillion), true, nil
},
}

cache.AddTx(createTx([]byte("tx-alice-1"), "alice", 1).withGasLimit(50000))
cache.AddTx(createTx([]byte("tx-alice-2"), "alice", 2).withGasLimit(1500000))
cache.AddTx(createTx([]byte("tx-alice-4"), "alice", 3).withGasLimit(1500000))
require.Equal(t, []string{"tx-alice-1", "tx-alice-2", "tx-alice-4"}, cache.getHashesForSender("alice"))
require.True(t, cache.areInternalMapsConsistent())

err := cache.OnProposedBlock(
[]byte("blockHash1"),
&block.Body{
MiniBlocks: []*block.MiniBlock{
{
TxHashes: [][]byte{
[]byte("tx-alice-1"),
[]byte("tx-alice-2"),
[]byte("tx-alice-4"),
},
},
},
},
&block.Header{
Nonce: 1,
PrevHash: []byte("blockHash0"),
},
accountsProvider,
[]byte("blockHash0"),
)
require.Nil(t, err)

// TODO analyze if this behaviour is ok. Even if the limit of txs per sender is exceeded, no tx is removed because the specific nonce is tracked.
cache.AddTx(createTx([]byte("tx-alice-3"), "alice", 3).withGasLimit(1500000))
require.Equal(t, []string{"tx-alice-1", "tx-alice-2", "tx-alice-3", "tx-alice-4"}, cache.getHashesForSender("alice"))
Copy link
Contributor Author

@danielradu10 danielradu10 Oct 20, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

because the nonce=3 is tracked for alice, no tx is removed from the list.

require.True(t, cache.areInternalMapsConsistent())

err = cache.OnExecutedBlock(
&block.Header{
Nonce: 1,
PrevHash: []byte("blockHash0"),
})
require.Nil(t, err)

cache.AddTx(createTx([]byte("tx-alice-5"), "alice", 3).withGasLimit(1500000))
require.Equal(t, []string{"tx-alice-1", "tx-alice-2", "tx-alice-3"}, cache.getHashesForSender("alice"))
require.True(t, cache.areInternalMapsConsistent())
})
}

func Test_AddTx_AppliesSizeConstraintsPerSenderForNumBytes(t *testing.T) {
Expand Down Expand Up @@ -490,7 +550,7 @@ func TestTxCache_TransactionIsAdded_EvenWhenInternalMapsAreInconsistent(t *testi
cache.Clear()

// Setup inconsistency: transaction already exists in map by sender, but not in map by hash
cache.txListBySender.addTxReturnEvicted(createTx([]byte("alice-x"), "alice", 42))
cache.txListBySender.addTxReturnEvicted(createTx([]byte("alice-x"), "alice", 42), cache.tracker)

require.False(t, cache.Has([]byte("alice-x")))
ok, added = cache.AddTx(createTx([]byte("alice-x"), "alice", 42))
Expand Down
4 changes: 2 additions & 2 deletions txcache/txListBySenderMap.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,11 +30,11 @@ func newTxListBySenderMap(

// addTxReturnEvicted adds a transaction in the map, in the corresponding list (selected by its sender).
// This function returns a boolean indicating whether the transaction was added, and a slice of evicted transaction hashes (upon applying sender-level constraints).
func (txMap *txListBySenderMap) addTxReturnEvicted(tx *WrappedTransaction) (bool, [][]byte) {
func (txMap *txListBySenderMap) addTxReturnEvicted(tx *WrappedTransaction, tracker *selectionTracker) (bool, [][]byte) {
sender := string(tx.Tx.GetSndAddr())
listForSender := txMap.getOrAddListForSender(sender)

added, evictedHashes := listForSender.AddTx(tx)
added, evictedHashes := listForSender.AddTx(tx, tracker)
return added, evictedHashes
}

Expand Down
Loading
Loading