feat(nonce-cache): read from nonce cache when we flush tx pool or ret…

…urn 0 by sender feat(nonce-cache): rm print lines feat(nonce-cache): rm print lines 2 feat(nonce-cache): rm print lines 3 feat(nonce-cache): rm print lines 4
elliothllm · Dec 17, 2024 · 1c0acc3 · 1c0acc3
1 parent e09961a
commit 1c0acc3
Show file tree

Hide file tree

Showing 8 changed files with 241 additions and 72 deletions.
diff --git a/turbo/jsonrpc/eth_accounts.go b/turbo/jsonrpc/eth_accounts.go
@@ -9,16 +9,11 @@ import (
 
 	libcommon "github.com/ledgerwatch/erigon-lib/common"
 	"github.com/ledgerwatch/erigon-lib/common/hexutility"
-	"github.com/ledgerwatch/erigon-lib/gointerfaces"
-	"google.golang.org/grpc"
 
 	"github.com/ledgerwatch/erigon/turbo/rpchelper"
 
-	txpool_proto "github.com/ledgerwatch/erigon-lib/gointerfaces/txpool"
-
 	"github.com/ledgerwatch/erigon/common"
 	"github.com/ledgerwatch/erigon/rpc"
-	"github.com/ledgerwatch/erigon/zk/sequencer"
 )
 
 // GetBalance implements eth_getBalance. Returns the balance of an account for a given address.
@@ -45,63 +40,6 @@ func (api *APIImpl) GetBalance(ctx context.Context, address libcommon.Address, b
 	return (*hexutil.Big)(acc.Balance.ToBig()), nil
 }
 
-// GetTransactionCount implements eth_getTransactionCount. Returns the number of transactions sent from an address (the nonce).
-func (api *APIImpl) GetTransactionCount(ctx context.Context, address libcommon.Address, blockNrOrHash *rpc.BlockNumberOrHash) (*hexutil.Uint64, error) {
-	// zkevm: forward requests to the sequencer
-	if !sequencer.IsSequencer() {
-		res, err := api.sendGetTransactionCountToSequencer(api.l2RpcUrl, address, blockNrOrHash)
-		if err != nil {
-			return nil, err
-		}
-		return res, nil
-	}
-
-	// if not set, use latest
-	if blockNrOrHash == nil {
-		tmp := rpc.BlockNumberOrHashWithNumber(rpc.LatestBlockNumber)
-		blockNrOrHash = &tmp
-	}
-
-	if blockNrOrHash.BlockNumber != nil && *blockNrOrHash.BlockNumber == rpc.PendingBlockNumber {
-		reply, err := api.txPool.Nonce(ctx, &txpool_proto.NonceRequest{
-			Address: gointerfaces.ConvertAddressToH160(address),
-		}, &grpc.EmptyCallOption{})
-		if err != nil {
-			return nil, err
-		}
-		if reply.Found {
-			reply.Nonce++
-			return (*hexutil.Uint64)(&reply.Nonce), nil
-		}
-	}
-	tx, err1 := api.db.BeginRo(ctx)
-	if err1 != nil {
-		return nil, fmt.Errorf("getTransactionCount cannot open tx: %w", err1)
-	}
-	defer tx.Rollback()
-
-	latestExecutedBlockNumber, err := rpchelper.GetLatestExecutedBlockNumber(tx)
-	if err != nil {
-		return nil, fmt.Errorf("getTransactionCount cannot get latest executed block number: %w", err)
-	}
-
-	if blockNrOrHash.BlockNumber != nil && *blockNrOrHash.BlockNumber == rpc.BlockNumber(latestExecutedBlockNumber) {
-		blockNumber := rpc.BlockNumber(rpc.LatestExecutedBlockNumber)
-		blockNrOrHash.BlockNumber = &blockNumber
-	}
-
-	reader, err := rpchelper.CreateStateReader(ctx, tx, *blockNrOrHash, 0, api.filters, api.stateCache, api.historyV3(tx), "")
-	if err != nil {
-		return nil, err
-	}
-	nonce := hexutil.Uint64(0)
-	acc, err := reader.ReadAccountData(address)
-	if acc == nil || err != nil {
-		return &nonce, err
-	}
-	return (*hexutil.Uint64)(&acc.Nonce), err
-}
-
 // GetCode implements eth_getCode. Returns the byte code at a given address (if it's a smart contract).
 func (api *APIImpl) GetCode(ctx context.Context, address libcommon.Address, blockNrOrHash rpc.BlockNumberOrHash) (hexutility.Bytes, error) {
 	tx, err1 := api.db.BeginRo(ctx)

diff --git a/turbo/jsonrpc/eth_accounts_zk.go b/turbo/jsonrpc/eth_accounts_zk.go
@@ -1,14 +1,20 @@
 package jsonrpc
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
 	libcommon "github.com/ledgerwatch/erigon-lib/common"
 	"github.com/ledgerwatch/erigon-lib/common/hexutil"
+	"github.com/ledgerwatch/erigon-lib/gointerfaces"
+	txpool_proto "github.com/ledgerwatch/erigon-lib/gointerfaces/txpool"
 	"github.com/ledgerwatch/erigon/rpc"
+	"github.com/ledgerwatch/erigon/turbo/rpchelper"
+	"github.com/ledgerwatch/erigon/zk/sequencer"
 	"github.com/ledgerwatch/erigon/zkevm/hex"
 	"github.com/ledgerwatch/erigon/zkevm/jsonrpc/client"
+	"google.golang.org/grpc"
 )
 
 func (api *APIImpl) sendGetTransactionCountToSequencer(rpcUrl string, address libcommon.Address, blockNrOrHash *rpc.BlockNumberOrHash) (*hexutil.Uint64, error) {
@@ -46,3 +52,74 @@ func (api *APIImpl) sendGetTransactionCountToSequencer(rpcUrl string, address li
 
 	return &result, nil
 }
+
+var nonceGoneBackwards = false
+var nonce = uint64(0)
+var lastNonce = uint64(0)
+
+// GetTransactionCount implements eth_getTransactionCount. Returns the number of transactions sent from an address (the nonce).
+func (api *APIImpl) GetTransactionCount(ctx context.Context, address libcommon.Address, blockNrOrHash *rpc.BlockNumberOrHash) (*hexutil.Uint64, error) {
+	// zkevm: forward requests to the sequencer
+	if !sequencer.IsSequencer() {
+		res, err := api.sendGetTransactionCountToSequencer(api.l2RpcUrl, address, blockNrOrHash)
+		if err != nil {
+			return nil, err
+		}
+		return res, nil
+	}
+
+	// if not set, use latest
+	if blockNrOrHash == nil {
+		tmp := rpc.BlockNumberOrHashWithNumber(rpc.LatestBlockNumber)
+		blockNrOrHash = &tmp
+	}
+
+	if blockNrOrHash.BlockNumber != nil && *blockNrOrHash.BlockNumber == rpc.PendingBlockNumber {
+		reply, err := api.txPool.Nonce(ctx, &txpool_proto.NonceRequest{
+			Address: gointerfaces.ConvertAddressToH160(address),
+		}, &grpc.EmptyCallOption{})
+		if err != nil {
+			return nil, err
+		}
+
+		nonce = reply.Nonce
+		nonceGoneBackwards = nonce < lastNonce
+		if nonceGoneBackwards {
+			// trigger breakpoint
+			fmt.Println("nonceGoneBackwards", nonceGoneBackwards, nonce, lastNonce, reply.Found, reply.Nonce)
+		}
+		lastNonce = nonce
+
+		if reply.Found {
+			reply.Nonce++
+			return (*hexutil.Uint64)(&reply.Nonce), nil
+		}
+	}
+	tx, err1 := api.db.BeginRo(ctx)
+	if err1 != nil {
+		return nil, fmt.Errorf("getTransactionCount cannot open tx: %w", err1)
+	}
+	defer tx.Rollback()
+
+	latestExecutedBlockNumber, err := rpchelper.GetLatestExecutedBlockNumber(tx)
+	if err != nil {
+		return nil, fmt.Errorf("getTransactionCount cannot get latest executed block number: %w", err)
+	}
+
+	if blockNrOrHash.BlockNumber != nil && *blockNrOrHash.BlockNumber == rpc.BlockNumber(latestExecutedBlockNumber) {
+		blockNumber := rpc.BlockNumber(rpc.LatestExecutedBlockNumber)
+		blockNrOrHash.BlockNumber = &blockNumber
+	}
+
+	reader, err := rpchelper.CreateStateReader(ctx, tx, *blockNrOrHash, 0, api.filters, api.stateCache, api.historyV3(tx), "")
+	if err != nil {
+		return nil, err
+	}
+	nonce := hexutil.Uint64(0)
+	acc, err := reader.ReadAccountData(address)
+	if acc == nil || err != nil {
+		return &nonce, err
+	}
+
+	return (*hexutil.Uint64)(&acc.Nonce), err
+}
diff --git a/turbo/jsonrpc/eth_api.go b/turbo/jsonrpc/eth_api.go
@@ -379,8 +379,8 @@ type APIImpl struct {
 	gasTracker                    RpcL1GasPriceTracker
 	RejectLowGasPriceTransactions bool
 	RejectLowGasPriceTolerance    float64
- 
-	logLevel                    utils.LogLevel
+
+	logLevel utils.LogLevel
 }
 
 // NewEthAPI returns APIImpl instance

diff --git a/zk/nonce_cache/nonce_cache.go b/zk/nonce_cache/nonce_cache.go
@@ -0,0 +1,60 @@
+package nonce_cache
+
+import (
+	"sync"
+
+	"github.com/ledgerwatch/erigon-lib/common"
+)
+
+const MaxNonceCacheSize = 1000
+
+type NonceCache struct {
+	highestNoncesBySender map[common.Address]uint64
+	mu                    sync.RWMutex
+	maxSize               uint64
+}
+
+func NewNonceCache(cacheSize uint64) *NonceCache {
+	return &NonceCache{
+		highestNoncesBySender: make(map[common.Address]uint64),
+		mu:                    sync.RWMutex{},
+		maxSize:               cacheSize,
+	}
+}
+
+func (nc *NonceCache) GetHighestNonceForSender(sender common.Address) (uint64, bool) {
+	nc.mu.RLock()
+	defer nc.mu.RUnlock()
+	nonce, ok := nc.highestNoncesBySender[sender]
+	return nonce, ok
+}
+
+func (nc *NonceCache) TrySetHighestNonceForSender(sender common.Address, nonce uint64) bool {
+	nc.mu.Lock()
+	defer nc.mu.Unlock()
+
+	if nc.highestNoncesBySender[sender] < nonce {
+		nc.highestNoncesBySender[sender] = nonce
+		if uint64(len(nc.highestNoncesBySender)) > nc.maxSize {
+			var oldestSender common.Address
+			var oldestNonce uint64
+			for s, n := range nc.highestNoncesBySender {
+				if oldestNonce == 0 || n < oldestNonce {
+					oldestSender = s
+					oldestNonce = n
+				}
+			}
+			delete(nc.highestNoncesBySender, oldestSender)
+		}
+
+		return true
+	}
+
+	return false
+}
+
+func (nc *NonceCache) CacheSize() int {
+	nc.mu.RLock()
+	defer nc.mu.RUnlock()
+	return len(nc.highestNoncesBySender)
+}
diff --git a/zk/nonce_cache/nonce_cache_test.go b/zk/nonce_cache/nonce_cache_test.go
@@ -0,0 +1,79 @@
+package nonce_cache
+
+import (
+	"github.com/ledgerwatch/erigon-lib/common"
+	"testing"
+)
+
+const testCacheSize = 4
+
+func Test_NonceCacheSize(t *testing.T) {
+	scenarios := map[string]struct {
+		setupActions   []func(*NonceCache)
+		expectedSize   int
+		expectedExists map[string]bool
+	}{
+		"Initial state with 4 entries": {
+			setupActions: []func(*NonceCache){
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x1"), 1) },
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x2"), 2) },
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x3"), 3) },
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x4"), 4) },
+			},
+			expectedSize: 4,
+			expectedExists: map[string]bool{
+				"0x1": true, "0x2": true, "0x3": true, "0x4": true,
+			},
+		},
+		"Evict when adding 5th entry": {
+			setupActions: []func(*NonceCache){
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x1"), 1) },
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x2"), 2) },
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x3"), 3) },
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x4"), 4) },
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x5"), 5) },
+			},
+			expectedSize: testCacheSize,
+			expectedExists: map[string]bool{
+				"0x1": false, "0x2": true, "0x3": true, "0x4": true, "0x5": true,
+			},
+		},
+		"Try write loads of entries": {
+			setupActions: []func(*NonceCache){
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x1"), 1) },
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x2"), 2) },
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x3"), 3) },
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x4"), 4) },
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x5"), 5) },
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x6"), 6) },
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x7"), 7) },
+				func(nc *NonceCache) { nc.TrySetHighestNonceForSender(common.HexToAddress("0x8"), 8) },
+			},
+			expectedSize: testCacheSize,
+			expectedExists: map[string]bool{
+				"0x1": false, "0x2": false, "0x3": false, "0x4": false, "0x5": true, "0x6": true, "0x7": true, "0x8": true,
+			},
+		},
+	}
+
+	for name, scenario := range scenarios {
+		t.Run(name, func(t *testing.T) {
+			cache := NewNonceCache(testCacheSize)
+
+			for _, action := range scenario.setupActions {
+				action(cache)
+			}
+
+			if size := cache.CacheSize(); size != scenario.expectedSize {
+				t.Errorf("Expected cache size %d, got %d", scenario.expectedSize, size)
+			}
+
+			for addr, shouldExist := range scenario.expectedExists {
+				_, exists := cache.GetHighestNonceForSender(common.HexToAddress(addr))
+				if exists != shouldExist {
+					t.Errorf("Expected sender %s to exist: %t, got %t", addr, shouldExist, exists)
+				}
+			}
+		})
+	}
+}
diff --git a/zk/stages/stage_sequence_execute.go b/zk/stages/stage_sequence_execute.go
@@ -592,10 +592,8 @@ func sequencingBatchStep(
 			return err
 		}
 
-		if err := cfg.txPool.RemoveMinedTransactions(ctx, sdb.tx, header.GasLimit, batchState.blockState.builtBlockElements.txSlots); err != nil {
-			return err
-		}
-		if err := cfg.txPool.RemoveMinedTransactions(ctx, sdb.tx, header.GasLimit, batchState.blockState.transactionsToDiscard); err != nil {
+		txsToRemove := append(batchState.blockState.builtBlockElements.txSlots, batchState.blockState.transactionsToDiscard...)
+		if err = cfg.txPool.RemoveMinedTransactions(ctx, sdb.tx, header.GasLimit, txsToRemove); err != nil {
 			return err
 		}