From 2b48bdcf4cc1408cebdfb8b4fd8e60c896ffcd9b Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Wed, 26 Aug 2026 11:39:31 -0400 Subject: [PATCH 1/2] fix(loadtest): bound parallel nonce fetch by --concurrency FetchNoncesInParallel spawned one goroutine per account and issued every eth_getTransactionCount at once, so --concurrency did not apply to the heaviest read burst the tool produces. With a 10,000-account --sending-accounts-file the endpoint received ~10k concurrent requests in the first seconds of a run, which load-balanced and managed endpoints answer with 429s / 5xx, aborting the run before anything is sent. Acquire a semaphore slot before spawning each fetch so both in-flight requests and live goroutines stay capped at min(--concurrency, N); values <= 0 clamp to 1. Each fetch now retries up to 5 times with exponential backoff (250ms -> 4s) so transient load shedding no longer fails the run, and both the acquire loop and the backoff select on ctx.Done() so Ctrl+C returns promptly. --sequential-nonce-fetch is unchanged; its help text now describes what the two paths actually do. Co-Authored-By: Claude Opus 5 (1M context) --- cmd/loadtest/cmd.go | 2 +- doc/polycli_loadtest.md | 2 +- loadtest/account.go | 91 +++++++++++++- loadtest/account_test.go | 265 +++++++++++++++++++++++++++++++++++++++ loadtest/runner.go | 3 +- 5 files changed, 356 insertions(+), 7 deletions(-) create mode 100644 loadtest/account_test.go diff --git a/cmd/loadtest/cmd.go b/cmd/loadtest/cmd.go index 721ad99ec..3611c3fef 100644 --- a/cmd/loadtest/cmd.go +++ b/cmd/loadtest/cmd.go @@ -172,7 +172,7 @@ func initFlags() { f.StringVar(&cfg.SendingAccountsFile, "sending-accounts-file", "", "file with sending account private keys, one per line (avoids pool queue and preserves accounts across runs)") f.StringVar(&cfg.DumpSendingAccountsFile, "dump-sending-accounts-file", "", "file path to dump generated private keys when using --sending-accounts-count") f.Uint64Var(&cfg.AccountsPerFundingTx, "accounts-per-funding-tx", 400, "number of accounts to fund per multicall3 transaction") - f.BoolVar(&cfg.SequentialNonceFetch, "sequential-nonce-fetch", false, "fetch nonces sequentially instead of in parallel (use if hitting rate limits)") + f.BoolVar(&cfg.SequentialNonceFetch, "sequential-nonce-fetch", false, "fetch nonces one at a time through the rate limiter instead of in parallel bounded by --concurrency") f.Uint64Var(&cfg.MaxBaseFeeWei, "max-base-fee-wei", 0, "maximum base fee in wei (pause sending new transactions when exceeded, useful during network congestion)") f.StringSliceVarP(&cfg.Modes, "mode", "m", []string{"t"}, `testing mode (can specify multiple like "d,t"): 2, erc20 - send ERC20 tokens diff --git a/doc/polycli_loadtest.md b/doc/polycli_loadtest.md index fda9531eb..bf0b99153 100644 --- a/doc/polycli_loadtest.md +++ b/doc/polycli_loadtest.md @@ -200,7 +200,7 @@ The codebase has a contract that used for load testing. It's written in Solidity --send-rpc-url string secondary RPC endpoint used only to broadcast transactions (eth_sendRawTransaction / eth_sendRawTransactionPrivate); all other calls use --rpc-url --sending-accounts-count uint number of sending accounts to use (avoids pool account queue) --sending-accounts-file string file with sending account private keys, one per line (avoids pool queue and preserves accounts across runs) - --sequential-nonce-fetch fetch nonces sequentially instead of in parallel (use if hitting rate limits) + --sequential-nonce-fetch fetch nonces one at a time through the rate limiter instead of in parallel bounded by --concurrency --stop-on-insufficient-funds stop sending from account when it encounters insufficient funds error --store-data-size uint number of bytes to store in contract for store mode (default 1024) --summarize produce execution summary after load test (can take a long time for large tests) diff --git a/loadtest/account.go b/loadtest/account.go index 950909bd2..4bf012807 100644 --- a/loadtest/account.go +++ b/loadtest/account.go @@ -35,6 +35,9 @@ type AccountPoolConfig struct { AccountsPerFundingTx uint64 SequentialNonceFetch bool StopOnInsufficientFunds bool + // Concurrency bounds the number of in-flight requests the parallel nonce + // sweep issues. Values <= 0 fall back to a single request at a time. + Concurrency int64 // Gas override settings ForceGasPrice uint64 ForcePriorityGasPrice uint64 @@ -238,8 +241,21 @@ func (ap *AccountPool) AllAccountsReady() (bool, int, int) { return rdyCount == len(ap.accounts), rdyCount, len(ap.accounts) } -// FetchNoncesInParallel fetches nonces for all accounts that aren't ready yet, -// in parallel without rate limiting. This is the default behavior unless SequentialNonceFetch is enabled. +// nonceFetchMaxAttempts is how many times the parallel nonce sweep tries a +// single eth_getTransactionCount before giving up on that account. Managed and +// load-balanced endpoints shed load with 429s / 5xx under a large sweep, so a +// bounded retry lets the run survive transient rejections. +const nonceFetchMaxAttempts = 5 + +// nonceFetchInitialBackoff is the delay before the second attempt; it doubles +// on every subsequent attempt. +const nonceFetchInitialBackoff = 250 * time.Millisecond + +// FetchNoncesInParallel fetches nonces for all accounts that aren't ready yet. +// Fetches run concurrently but no more than Concurrency requests are in flight +// at once, so a large --sending-accounts-file doesn't hit the endpoint with one +// request per account simultaneously. This is the default behavior unless +// SequentialNonceFetch is enabled. func (ap *AccountPool) FetchNoncesInParallel(ctx context.Context) error { ap.mu.Lock() // Collect accounts that need nonce fetching @@ -255,16 +271,41 @@ func (ap *AccountPool) FetchNoncesInParallel(ctx context.Context) error { return nil } - log.Info().Int("count", len(accountsToFetch)).Msg("Fetching nonces in parallel") + concurrency := int(ap.cfg.Concurrency) + if concurrency <= 0 { + concurrency = 1 + } + if concurrency > len(accountsToFetch) { + concurrency = len(accountsToFetch) + } + + log.Info(). + Int("count", len(accountsToFetch)). + Int("concurrency", concurrency). + Msg("Fetching nonces in parallel") var wg sync.WaitGroup errCh := make(chan error, len(accountsToFetch)) + // sem bounds in-flight requests; acquiring a slot before spawning keeps the + // number of live goroutines bounded too. + sem := make(chan struct{}, concurrency) + var canceled bool for _, acc := range accountsToFetch { + select { + case sem <- struct{}{}: + case <-ctx.Done(): + canceled = true + } + if canceled { + break + } + wg.Add(1) go func(a *Account) { defer wg.Done() - nonce, err := ap.client.NonceAt(ctx, a.address, nil) + defer func() { <-sem }() + nonce, err := ap.fetchNonceWithRetry(ctx, a.address) if err != nil { errCh <- fmt.Errorf("failed to get nonce for %s: %w", a.address.Hex(), err) return @@ -280,6 +321,10 @@ func (ap *AccountPool) FetchNoncesInParallel(ctx context.Context) error { wg.Wait() close(errCh) + if canceled { + return ctx.Err() + } + // Collect errors var errs []error for err := range errCh { @@ -293,6 +338,44 @@ func (ap *AccountPool) FetchNoncesInParallel(ctx context.Context) error { return nil } +// fetchNonceWithRetry gets the nonce for an address, retrying with exponential +// backoff so a transiently overloaded endpoint doesn't abort the whole run. +func (ap *AccountPool) fetchNonceWithRetry(ctx context.Context, address common.Address) (uint64, error) { + backoff := nonceFetchInitialBackoff + var lastErr error + for attempt := 1; attempt <= nonceFetchMaxAttempts; attempt++ { + nonce, err := ap.client.NonceAt(ctx, address, nil) + if err == nil { + return nonce, nil + } + lastErr = err + if ctx.Err() != nil { + return 0, ctx.Err() + } + if attempt == nonceFetchMaxAttempts { + break + } + + log.Warn(). + Err(err). + Stringer("addr", address). + Int("attempt", attempt). + Dur("backoff", backoff). + Msg("Failed to get nonce for account, retrying") + + timer := time.NewTimer(backoff) + select { + case <-timer.C: + case <-ctx.Done(): + timer.Stop() + return 0, ctx.Err() + } + timer.Stop() + backoff *= 2 + } + return 0, fmt.Errorf("failed after %d attempts: %w", nonceFetchMaxAttempts, lastErr) +} + // StopAccount marks an account as stopped so it won't be used for further transactions. func (ap *AccountPool) StopAccount(address common.Address) error { ap.mu.Lock() diff --git a/loadtest/account_test.go b/loadtest/account_test.go new file mode 100644 index 000000000..69689e59a --- /dev/null +++ b/loadtest/account_test.go @@ -0,0 +1,265 @@ +package loadtest + +import ( + "context" + "encoding/json" + "fmt" + "math/big" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/ethclient" +) + +// rpcRequest is the subset of the JSON-RPC request the fake server needs. +type rpcRequest struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` +} + +// fakeRPC is a minimal JSON-RPC server that serves the calls the account pool +// makes during construction and nonce fetching. It records the peak number of +// concurrent eth_getTransactionCount calls and can fail the first N of them. +type fakeRPC struct { + server *httptest.Server + + mu sync.Mutex + inFlight int + maxInFlight int + nonceCalls int + failFirstN int + nonceLatency time.Duration + + totalNonceCalls atomic.Int64 +} + +func newFakeRPC(t *testing.T, failFirstN int, nonceLatency time.Duration) *fakeRPC { + t.Helper() + f := &fakeRPC{failFirstN: failFirstN, nonceLatency: nonceLatency} + f.server = httptest.NewServer(http.HandlerFunc(f.handle)) + t.Cleanup(f.server.Close) + return f +} + +func (f *fakeRPC) handle(w http.ResponseWriter, r *http.Request) { + var req rpcRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + writeResult := func(result string) { + w.Header().Set("Content-Type", "application/json") + fmt.Fprintf(w, `{"jsonrpc":"2.0","id":%s,"result":%q}`, req.ID, result) + } + + switch req.Method { + case "eth_chainId": + writeResult("0x89") + case "eth_blockNumber": + writeResult("0x1") + case "eth_getTransactionCount": + f.totalNonceCalls.Add(1) + + f.mu.Lock() + f.nonceCalls++ + shouldFail := f.nonceCalls <= f.failFirstN + f.inFlight++ + if f.inFlight > f.maxInFlight { + f.maxInFlight = f.inFlight + } + f.mu.Unlock() + + defer func() { + f.mu.Lock() + f.inFlight-- + f.mu.Unlock() + }() + + if f.nonceLatency > 0 { + time.Sleep(f.nonceLatency) + } + + if shouldFail { + // Mimic a load-balanced endpoint shedding load. + http.Error(w, "the server encountered an error", http.StatusInternalServerError) + return + } + writeResult("0x2a") + default: + http.Error(w, "unexpected method "+req.Method, http.StatusBadRequest) + } +} + +func (f *fakeRPC) peakConcurrency() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.maxInFlight +} + +// newTestPool builds an account pool pointed at the fake RPC with n accounts +// that have no nonce yet, so FetchNoncesInParallel has work to do. +func newTestPool(t *testing.T, f *fakeRPC, concurrency int64, n int) *AccountPool { + t.Helper() + ctx := context.Background() + + client, err := ethclient.DialContext(ctx, f.server.URL) + if err != nil { + t.Fatalf("failed to dial fake rpc: %v", err) + } + t.Cleanup(client.Close) + + fundingKey, err := crypto.GenerateKey() + if err != nil { + t.Fatalf("failed to generate funding key: %v", err) + } + + ap, err := NewAccountPool(ctx, client, &AccountPoolConfig{ + FundingPrivateKey: fundingKey, + FundingAmount: big.NewInt(0), + Concurrency: concurrency, + }) + if err != nil { + t.Fatalf("failed to create account pool: %v", err) + } + + for range n { + key, err := crypto.GenerateKey() + if err != nil { + t.Fatalf("failed to generate account key: %v", err) + } + if err := ap.Add(ctx, key, nil); err != nil { + t.Fatalf("failed to add account: %v", err) + } + } + + return ap +} + +func TestFetchNoncesInParallelRespectsConcurrency(t *testing.T) { + const accounts = 50 + const concurrency = 4 + + f := newFakeRPC(t, 0, 5*time.Millisecond) + ap := newTestPool(t, f, concurrency, accounts) + + if err := ap.FetchNoncesInParallel(context.Background()); err != nil { + t.Fatalf("FetchNoncesInParallel failed: %v", err) + } + + if peak := f.peakConcurrency(); peak > concurrency { + t.Errorf("peak in-flight nonce requests = %d, want <= %d", peak, concurrency) + } + + ready, rdyCount, total := ap.AllAccountsReady() + if !ready { + t.Errorf("accounts not all ready: %d/%d", rdyCount, total) + } + for _, acc := range ap.accounts { + if acc.nonce != 0x2a || acc.startNonce != 0x2a { + t.Errorf("account %s nonce = %d/%d, want 42/42", acc.address, acc.nonce, acc.startNonce) + } + } +} + +func TestFetchNoncesInParallelClampsConcurrency(t *testing.T) { + tests := []struct { + name string + concurrency int64 + }{ + {"zero falls back to one", 0}, + {"negative falls back to one", -5}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + f := newFakeRPC(t, 0, time.Millisecond) + ap := newTestPool(t, f, tt.concurrency, 5) + + if err := ap.FetchNoncesInParallel(context.Background()); err != nil { + t.Fatalf("FetchNoncesInParallel failed: %v", err) + } + if peak := f.peakConcurrency(); peak != 1 { + t.Errorf("peak in-flight nonce requests = %d, want 1", peak) + } + }) + } +} + +func TestFetchNoncesInParallelRetriesTransientFailures(t *testing.T) { + // Fail the first two calls so the single account needs retries to succeed. + f := newFakeRPC(t, 2, 0) + ap := newTestPool(t, f, 1, 1) + + if err := ap.FetchNoncesInParallel(context.Background()); err != nil { + t.Fatalf("FetchNoncesInParallel failed despite retries: %v", err) + } + if calls := f.totalNonceCalls.Load(); calls != 3 { + t.Errorf("nonce calls = %d, want 3 (two failures plus one success)", calls) + } + if ready, rdyCount, total := ap.AllAccountsReady(); !ready { + t.Errorf("accounts not all ready: %d/%d", rdyCount, total) + } +} + +func TestFetchNoncesInParallelFailsAfterMaxAttempts(t *testing.T) { + f := newFakeRPC(t, nonceFetchMaxAttempts, 0) + ap := newTestPool(t, f, 1, 1) + + err := ap.FetchNoncesInParallel(context.Background()) + if err == nil { + t.Fatal("expected error after exhausting retries, got nil") + } + if calls := f.totalNonceCalls.Load(); calls != int64(nonceFetchMaxAttempts) { + t.Errorf("nonce calls = %d, want %d", calls, nonceFetchMaxAttempts) + } +} + +func TestFetchNoncesInParallelHonorsCancellation(t *testing.T) { + // Every request fails, so the fetch is stuck in backoff when we cancel. + f := newFakeRPC(t, 1000, 0) + ap := newTestPool(t, f, 1, 20) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + done := make(chan error, 1) + go func() { done <- ap.FetchNoncesInParallel(ctx) }() + + select { + case err := <-done: + if err == nil { + t.Fatal("expected error on cancellation, got nil") + } + case <-time.After(10 * time.Second): + t.Fatal("FetchNoncesInParallel did not return after context cancellation") + } +} + +func TestFetchNoncesInParallelNoop(t *testing.T) { + f := newFakeRPC(t, 0, 0) + // A pool whose accounts all have a forced start nonce needs no fetching. + ap := newTestPool(t, f, 4, 0) + + ctx := context.Background() + key, err := crypto.GenerateKey() + if err != nil { + t.Fatalf("failed to generate key: %v", err) + } + startNonce := uint64(7) + if err := ap.Add(ctx, key, &startNonce); err != nil { + t.Fatalf("failed to add account: %v", err) + } + + if err := ap.FetchNoncesInParallel(ctx); err != nil { + t.Fatalf("FetchNoncesInParallel failed: %v", err) + } + if calls := f.totalNonceCalls.Load(); calls != 0 { + t.Errorf("nonce calls = %d, want 0", calls) + } +} diff --git a/loadtest/runner.go b/loadtest/runner.go index 81731b8d6..314612529 100644 --- a/loadtest/runner.go +++ b/loadtest/runner.go @@ -294,6 +294,7 @@ func (r *Runner) initAccountPool(ctx context.Context) error { AccountsPerFundingTx: r.cfg.AccountsPerFundingTx, SequentialNonceFetch: r.cfg.SequentialNonceFetch, StopOnInsufficientFunds: r.cfg.StopOnInsufficientFunds, + Concurrency: r.cfg.Concurrency, ForceGasPrice: r.cfg.ForceGasPrice, ForcePriorityGasPrice: r.cfg.ForcePriorityGasPrice, GasPriceMultiplier: r.cfg.BigGasPriceMultiplier, @@ -356,7 +357,7 @@ func (r *Runner) initAccountPool(ctx context.Context) error { // Wait for all accounts to be ready if !r.cfg.SequentialNonceFetch { - // Fetch nonces in parallel without rate limiting + // Fetch nonces in parallel, bounded by --concurrency in-flight requests if err := r.accountPool.FetchNoncesInParallel(ctx); err != nil { return errors.New("failed to fetch nonces in parallel: " + err.Error()) } From 56edc5d64935e451403dc109ff4b764b10852a86 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Wed, 26 Aug 2026 16:44:03 -0400 Subject: [PATCH 2/2] test(loadtest): check fmt.Fprintf error in fake rpc handler Satisfies errcheck. Co-Authored-By: Claude Opus 5 (1M context) --- loadtest/account_test.go | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/loadtest/account_test.go b/loadtest/account_test.go index d0d7c8499..c7be5af1b 100644 --- a/loadtest/account_test.go +++ b/loadtest/account_test.go @@ -28,6 +28,10 @@ type rpcRequest struct { // makes during construction and nonce fetching. It records the peak number of // concurrent eth_getTransactionCount calls and can fail the first N of them. type fakeRPC struct { + // t reports write failures from handler goroutines. Safe because the + // server is closed during test cleanup, which waits for outstanding + // requests before the test finishes. + t *testing.T server *httptest.Server mu sync.Mutex @@ -42,7 +46,7 @@ type fakeRPC struct { func newFakeRPC(t *testing.T, failFirstN int, nonceLatency time.Duration) *fakeRPC { t.Helper() - f := &fakeRPC{failFirstN: failFirstN, nonceLatency: nonceLatency} + f := &fakeRPC{t: t, failFirstN: failFirstN, nonceLatency: nonceLatency} f.server = httptest.NewServer(http.HandlerFunc(f.handle)) t.Cleanup(f.server.Close) return f @@ -57,7 +61,9 @@ func (f *fakeRPC) handle(w http.ResponseWriter, r *http.Request) { writeResult := func(result string) { w.Header().Set("Content-Type", "application/json") - fmt.Fprintf(w, `{"jsonrpc":"2.0","id":%s,"result":%q}`, req.ID, result) + if _, err := fmt.Fprintf(w, `{"jsonrpc":"2.0","id":%s,"result":%q}`, req.ID, result); err != nil { + f.t.Errorf("failed to write rpc response: %v", err) + } } switch req.Method {