sender lock to help prevent incorrect nonce checks (#1643)

* sender lock to help prevent incorrect nonce checks

helps to ensure the pool has finished working with a sent transaction before allowing any nonce checks to be performed.  Only applies to pending nonce, latest won't use this lock.

* sender lock updates post PR review

cmd/rpcdaemon/commands/eth_accounts.go

@@ -4,13 +4,15 @@ import (
 	"context"
 	"fmt"
 	"math/big"
+	"time"
 
 	libcommon "github.com/gateway-fm/cdk-erigon-lib/common"
 	"github.com/gateway-fm/cdk-erigon-lib/common/hexutility"
 	"github.com/gateway-fm/cdk-erigon-lib/gointerfaces"
 	"google.golang.org/grpc"
 
 	"github.com/ledgerwatch/erigon/turbo/rpchelper"
+	"github.com/ledgerwatch/log/v3"
 
 	txpool_proto "github.com/gateway-fm/cdk-erigon-lib/gointerfaces/txpool"
 
@@ -62,6 +64,22 @@ func (api *APIImpl) GetTransactionCount(ctx context.Context, address libcommon.A
 	}
 
 	if blockNrOrHash.BlockNumber != nil && *blockNrOrHash.BlockNumber == rpc.PendingBlockNumber {
+		// first lets check for any locks on this account before continuing - but don't wait too long
+		waits := 0
+		for {
+			lock := api.SenderLocks.GetLock(address)
+			if lock > 0 {
+				time.Sleep(2 * time.Millisecond)
+				waits++
+				if waits > 250 {
+					log.Debug("waiting too long for transaction processing, returning default behaviour for nonce", "address", address.String())
+					break
+				}
+				continue
+			}
+			break
+		}
+
 		reply, err := api.txPool.Nonce(ctx, &txpool_proto.NonceRequest{
 			Address: gointerfaces.ConvertAddressToH160(address),
 		}, &grpc.EmptyCallOption{})

cmd/rpcdaemon/commands/eth_api.go

@@ -359,6 +359,7 @@ type APIImpl struct {
 	VirtualCountersSmtReduction   float64
 	RejectLowGasPriceTransactions bool
 	RejectLowGasPriceTolerance    float64
+	SenderLocks                   *SenderLock
 }
 
 // NewEthAPI returns APIImpl instance
@@ -388,6 +389,7 @@ func NewEthAPI(base *BaseAPI, db kv.RoDB, eth rpchelper.ApiBackend, txPool txpoo
 		VirtualCountersSmtReduction:   ethCfg.VirtualCountersSmtReduction,
 		RejectLowGasPriceTransactions: ethCfg.RejectLowGasPriceTransactions,
 		RejectLowGasPriceTolerance:    ethCfg.RejectLowGasPriceTolerance,
+		SenderLocks:                   NewSenderLock(),
 	}
 }
 

cmd/rpcdaemon/commands/send_transaction.go

@@ -50,6 +50,15 @@ func (api *APIImpl) SendRawTransaction(ctx context.Context, encodedTx hexutility
 		return common.Hash{}, err
 	}
 
+	// now get the sender and put a lock in place for them
+	signer := types.MakeSigner(cc, chainId.Uint64())
+	sender, err := txn.Sender(*signer)
+	if err != nil {
+		return common.Hash{}, err
+	}
+	api.SenderLocks.AddLock(sender)
+	defer api.SenderLocks.ReleaseLock(sender)
+
 	// check if the transaction price is lower than the current network price and
 	// reject the tx if it is
 	if api.RejectLowGasPriceTransactions && !api.BaseAPI.gasless {

cmd/rpcdaemon/commands/zkevm_sender_locks.go

@@ -0,0 +1,47 @@
+package commands
+
+import (
+	"sync"
+
+	"github.com/gateway-fm/cdk-erigon-lib/common"
+)
+
+// SenderLock is a map of sender addresses to the number of locks they have
+// This is used to ensure that any calls for an account nonce will wait until all
+// pending transactions for that account have been processed.  Without this you can
+// get strange race behaviour where a nonce will come back too low if the pool is taking
+// a long time to process a transaction.
+type SenderLock struct {
+	mtx   sync.RWMutex
+	locks map[common.Address]uint64
+}
+
+func NewSenderLock() *SenderLock {
+	return &SenderLock{
+		locks: make(map[common.Address]uint64),
+	}
+}
+
+func (sl *SenderLock) GetLock(sender common.Address) uint64 {
+	sl.mtx.Lock()
+	defer sl.mtx.Unlock()
+	return sl.locks[sender]
+}
+
+func (sl *SenderLock) ReleaseLock(sender common.Address) {
+	sl.mtx.Lock()
+	defer sl.mtx.Unlock()
+	if current, ok := sl.locks[sender]; ok {
+		if current <= 1 {
+			delete(sl.locks, sender)
+		} else {
+			sl.locks[sender] = current - 1
+		}
+	}
+}
+
+func (sl *SenderLock) AddLock(sender common.Address) {
+	sl.mtx.Lock()
+	defer sl.mtx.Unlock()
+	sl.locks[sender]++
+}

cmd/rpcdaemon/commands/zkevm_sender_locks_test.go

@@ -0,0 +1,140 @@
+package commands
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/gateway-fm/cdk-erigon-lib/common"
+)
+
+func Test_SenderLocks(t *testing.T) {
+	sl := NewSenderLock()
+	addr := common.HexToAddress("0x1")
+
+	// ensure 0 to start
+	lock := sl.GetLock(addr)
+	if lock != 0 {
+		t.Fatalf("expected lock to be 0, got %d", lock)
+	}
+
+	// add a lock and check it shows
+	sl.AddLock(addr)
+	lock = sl.GetLock(addr)
+	if lock != 1 {
+		t.Fatalf("expected lock to be 1, got %d", lock)
+	}
+
+	// add another lock and check it shows
+	sl.AddLock(addr)
+	lock = sl.GetLock(addr)
+	if lock != 2 {
+		t.Fatalf("expected lock to be 2, got %d", lock)
+	}
+
+	// now release one and check it shows
+	sl.ReleaseLock(addr)
+	lock = sl.GetLock(addr)
+	if lock != 1 {
+		t.Fatalf("expected lock to be 1, got %d", lock)
+	}
+
+	// now release the last one and check it shows
+	sl.ReleaseLock(addr)
+	lock = sl.GetLock(addr)
+	if lock != 0 {
+		t.Fatalf("expected lock to be 0, got %d", lock)
+	}
+	if len(sl.locks) != 0 {
+		t.Fatalf("expected lock to be 0, got %d", len(sl.locks))
+	}
+}
+
+func Test_SenderLocks_Concurrency(t *testing.T) {
+	sl := NewSenderLock()
+	addr := common.HexToAddress("0x1")
+
+	wg := sync.WaitGroup{}
+	wg.Add(1000)
+	for i := 0; i < 1000; i++ {
+		go func() {
+			defer wg.Done()
+			sl.AddLock(addr)
+		}()
+	}
+	wg.Wait()
+
+	lock := sl.GetLock(addr)
+	if lock != 1000 {
+		t.Fatalf("expected lock to be 1000, got %d", lock)
+	}
+
+	// now release all the locks concurrently
+	wg.Add(1000)
+	for i := 0; i < 1000; i++ {
+		go func() {
+			defer wg.Done()
+			sl.ReleaseLock(addr)
+		}()
+	}
+	wg.Wait()
+
+	lock = sl.GetLock(addr)
+	if lock != 0 {
+		t.Fatalf("expected lock to be 0, got %d", lock)
+	}
+	if len(sl.locks) != 0 {
+		t.Fatalf("expected lock to be 0, got %d", len(sl.locks))
+	}
+}
+
+func Test_SenderLocks_MultipleAccounts(t *testing.T) {
+	sl := NewSenderLock()
+	addr1 := common.HexToAddress("0x1")
+	addr2 := common.HexToAddress("0x2")
+
+	sl.AddLock(addr1)
+	sl.AddLock(addr2)
+
+	lock1 := sl.GetLock(addr1)
+	lock2 := sl.GetLock(addr2)
+	if lock1 != 1 {
+		t.Fatalf("expected lock to be 1, got %d", lock1)
+	}
+	if lock2 != 1 {
+		t.Fatalf("expected lock to be 1, got %d", lock2)
+	}
+
+	sl.ReleaseLock(addr1)
+
+	lock1 = sl.GetLock(addr1)
+	lock2 = sl.GetLock(addr2)
+	if lock1 != 0 {
+		t.Fatalf("expected lock to be 1, got %d", lock1)
+	}
+	if lock2 != 1 {
+		t.Fatalf("expected lock to be 1, got %d", lock2)
+	}
+
+	sl.ReleaseLock(addr2)
+
+	lock1 = sl.GetLock(addr1)
+	lock2 = sl.GetLock(addr2)
+	if lock1 != 0 {
+		t.Fatalf("expected lock to be 1, got %d", lock1)
+	}
+	if lock2 != 0 {
+		t.Fatalf("expected lock to be 1, got %d", lock2)
+	}
+}
+
+func Test_SenderLocks_ReleaseWhenEmpty(t *testing.T) {
+	sl := NewSenderLock()
+	addr := common.HexToAddress("0x1")
+
+	sl.ReleaseLock(addr)
+
+	lock := sl.GetLock(addr)
+	if lock != 0 {
+		t.Fatalf("expected lock to be 0, got %d", lock)
+	}
+}