diff --git a/giga/evmonly/README.md b/giga/evmonly/README.md index 4945662901..5dbc72e65d 100644 --- a/giga/evmonly/README.md +++ b/giga/evmonly/README.md @@ -34,6 +34,8 @@ The `evmonly` package currently provides: contract address, and effective gas price - a map-backed `MemoryState` for tests and early integration - fail-closed custom precompile placeholders plus an SDK-free staking precompile +- a standalone load harness at `giga/evmonly/cmd/evmonly-loadtest` that feeds + generated transfer blocks into the executor with mock state and receipt sinks The executor accepts config for nonce checks, gas-price checks, minimum gas price, chain config, and the custom precompile registry. @@ -148,6 +150,8 @@ closed with `ErrCustomPrecompilesOpen`. - The map-backed `MemoryState` is for tests and early integration; production should provide a durable native state backend. - Historical `BLOCKHASH` lookups beyond the parent block are not wired yet. +- Block-level blob gas accounting and `MaxBlobGasPerBlock` enforcement are not + wired yet, so blob transactions are rejected fail-closed. - The staking precompile models bonding, delegation, redelegation, unbonding, and validator-set updates in usei, but does not model staking rewards, slashing, or jailing. Delegation shares track tokens 1:1 (no slash-driven diff --git a/giga/evmonly/cmd/evmonly-loadtest/README.md b/giga/evmonly/cmd/evmonly-loadtest/README.md new file mode 100644 index 0000000000..351751a32c --- /dev/null +++ b/giga/evmonly/cmd/evmonly-loadtest/README.md @@ -0,0 +1,110 @@ +# evmonly-loadtest + +`evmonly-loadtest` is a standalone executable for feeding synthetic blocks to +the EVM-only executor without Cosmos SDK state, mempool, RPC, or persistence. + +It currently generates pure EVM legacy transfer transactions and ERC20 transfer +transactions. Each generated sender account has exactly one nonce-0 transaction +and is funded in the command's in-memory genesis state before its block is +queued. Recipients are unique by default so the workloads exercise the optimistic +no-overlap case; pass `--recipient=0x...` to force a single recipient. + +Run a bounded test: + +```bash +go run ./giga/evmonly/cmd/evmonly-loadtest --blocks=1000 --txs-per-block=1000 +``` + +Run continuously until interrupted: + +```bash +go run ./giga/evmonly/cmd/evmonly-loadtest --txs-per-block=1000 +``` + +Example local saturation run: + +```bash +go run ./giga/evmonly/cmd/evmonly-loadtest \ + --metrics-addr= \ + --report-interval=5s \ + --blocks=2000 \ + --txs-per-block=1000 \ + --builders=16 \ + --workers=1 \ + --executor-workers=12 \ + --gas-price-wei=0 \ + --min-gas-price-wei=0 \ + --queue-size=512 +``` + +To isolate executor throughput from block generation, prebuild a bounded run +before starting the prepare/recover and executor workers: + +```bash +go run ./giga/evmonly/cmd/evmonly-loadtest \ + --metrics-addr= \ + --report-interval=5s \ + --prebuild-blocks \ + --blocks=400 \ + --txs-per-block=5000 \ + --builders=48 \ + --workers=1 \ + --executor-workers=24 \ + --gas-price-wei=0 \ + --min-gas-price-wei=0 \ + --queue-size=512 +``` + +Prebuilding requires `--blocks > 0` and stores every raw block in memory. +Sender recovery still runs in the measured phase, but it is pipelined ahead of +execution through `--prepare-workers`. + +The zero gas price/min-gas settings keep the transfer workload focused on the +optimistic no-overlap case. Non-zero fees make every transaction update the +same coinbase balance, which is a real intra-block conflict. + +Useful knobs: + +- `--workers`: parallel executor workers. The default is `1`. +- `--executor-workers`: parallel OCC workers inside each executor. The default + is `min(12, GOMAXPROCS)`, following the `sei-v3` OCC worker default. +- `--prepare-workers`: parallel stateless preparation workers used for + transaction RLP decode and sender recovery. The default is `GOMAXPROCS`. +- `--builders`: parallel block builders used to keep the input queue full. The + default is `GOMAXPROCS`. +- `--queue-size`: buffered raw and prepared blocks. The default is `64`. +- `--result-pool-size`: reusable executor result slots. The default `0` + sizes the pool for in-flight executor and async sink results; negative + disables result pooling. +- `--target-blocks-per-sec`: cap block input rate. The default `0` feeds as + fast as block generation and the queue allow. +- `--prebuild-blocks`: generate all bounded blocks before starting executor + workers. This separates build throughput from executor throughput. +- `--metrics-addr`: Prometheus endpoint. The default is + `127.0.0.1:9698`; set it to empty to disable HTTP metrics. +- `--report-interval`: stdout rate reporting interval. The default is `5s`. +- `--gas-price-wei`, `--min-gas-price-wei`, `--sender-balance-wei`, + `--transfer-value-wei`: transaction economics for the generated accounts. +- `--workload`: workload type, either `transfer` or `erc20-transfer`. + +The command reports these saturation signals on stdout and at `/metrics`: + +- block input throughput +- block preparation throughput +- prepared transactions per second +- block finishing throughput +- finished transactions per second +- total gas consumed per second +- prepared blocks queued for execution and cumulative totals + +The executor output is intentionally discarded through mocks: + +- `generatedState` implements `evmonly.StateReader` and supplies generated + genesis balances, nonces, code, and storage. +- `discardStateWriter` implements `evmonly.StateWriter` and sinks the + executor `StateChangeSet`. +- `discardReceiptSink` sinks Ethereum receipts. + +Future workloads should add another workload builder beside `transferWorkload` +and `erc20TransferWorkload`, then reuse the same block +producer/prepare/executor/metrics pipeline. diff --git a/giga/evmonly/cmd/evmonly-loadtest/main.go b/giga/evmonly/cmd/evmonly-loadtest/main.go new file mode 100644 index 0000000000..73b8238500 --- /dev/null +++ b/giga/evmonly/cmd/evmonly-loadtest/main.go @@ -0,0 +1,2166 @@ +package main + +import ( + "bufio" + "context" + "crypto/ecdsa" + "encoding/binary" + "errors" + "flag" + "fmt" + "math" + "math/big" + "net" + "net/http" + "os" + "os/signal" + "path/filepath" + "runtime" + "runtime/pprof" + "runtime/trace" + "strings" + "sync" + "sync/atomic" + "syscall" + "time" + + "github.com/ethereum/go-ethereum/common" + ethtypes "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/rlp" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promhttp" + "golang.org/x/sync/errgroup" + "golang.org/x/time/rate" + + "github.com/sei-protocol/sei-chain/giga/evmonly" +) + +const ( + defaultChainID = "713715" + defaultGasPriceWei = "1000000000" + defaultMinGasPriceWei = "1000000000" + defaultSenderBalance = "1000000000000000000" + defaultTransferValue = "1" + defaultERC20Contract = "0x000000000000000000000000000000000000e20c" + defaultMetricsAddr = "127.0.0.1:9698" + defaultReportInterval = 5 * time.Second + defaultQueueSize = 64 + defaultTxGasLimit = 21_000 + defaultERC20TxGasLimit = 100_000 + defaultTxsPerBlock = 1_000 + defaultPersistBuffer = 4 << 20 + defaultWorkerCount = 1 + defaultCoinbaseAddress = "0x00000000000000000000000000000000000000cb" + workloadTransfer = "transfer" + workloadERC20Transfer = "erc20-transfer" + resultSinkDiscard = "discard" + resultSinkFile = "file" + resultSinkBlock = "block" + resultSinkChangeSet = "changeset" + resultSinkReceipts = "receipts" +) + +type config struct { + blocks uint64 + txsPerBlock int + queueSize int + builders int + prepareWorkers int + workers int + executorWorkers int + targetBlocksPerSec float64 + reportInterval time.Duration + metricsAddr string + resultSink string + resultPoolSize int + persistDir string + persistSync bool + persistBufferSize int + persistQueueSize int + cpuProfile string + heapProfile string + traceProfile string + workload string + chainID *big.Int + gasPrice *big.Int + minGasPrice *big.Int + senderBalance *big.Int + transferValue *big.Int + txGasLimit uint64 + blockGasLimit uint64 + coinbase common.Address + erc20Contract common.Address + fixedRecipient *common.Address + disableGasPriceRule bool + prebuildBlocks bool +} + +type blockEnvelope struct { + number uint64 + request evmonly.BlockRequest +} + +type preparedBlockEnvelope struct { + number uint64 + block evmonly.PreparedBlock +} + +func main() { + cfg, err := parseConfig(os.Args[1:]) + if err != nil { + fmt.Fprintf(os.Stderr, "evmonly-loadtest: %v\n", err) + os.Exit(2) + } + if err := run(cfg); err != nil { + fmt.Fprintf(os.Stderr, "evmonly-loadtest: %v\n", err) + os.Exit(1) + } +} + +func parseConfig(args []string) (config, error) { + cfg := config{} + fs := flag.NewFlagSet("evmonly-loadtest", flag.ContinueOnError) + fs.SetOutput(os.Stderr) + + chainID := fs.String("chain-id", defaultChainID, "EVM chain ID used to sign and execute transactions") + gasPrice := fs.String("gas-price-wei", defaultGasPriceWei, "legacy transaction gas price in wei") + minGasPrice := fs.String("min-gas-price-wei", defaultMinGasPriceWei, "executor minimum gas price in wei") + senderBalance := fs.String("sender-balance-wei", defaultSenderBalance, "generated sender genesis balance in wei") + transferValue := fs.String("transfer-value-wei", defaultTransferValue, "wei or token units transferred by each generated transaction") + coinbase := fs.String("coinbase", defaultCoinbaseAddress, "block coinbase address") + erc20Contract := fs.String("erc20-contract", defaultERC20Contract, "EVM address for the generated ERC20 transfer contract") + recipient := fs.String("recipient", "", "optional fixed transfer recipient; empty creates one recipient per tx") + + fs.Uint64Var(&cfg.blocks, "blocks", 0, "number of blocks to feed; 0 runs until interrupted") + fs.IntVar(&cfg.txsPerBlock, "txs-per-block", defaultTxsPerBlock, "transactions generated per block") + fs.IntVar(&cfg.queueSize, "queue-size", defaultQueueSize, "buffered blocks waiting for executor workers") + fs.IntVar(&cfg.builders, "builders", runtime.GOMAXPROCS(0), "parallel block builder goroutines") + fs.IntVar(&cfg.prepareWorkers, "prepare-workers", defaultPrepareWorkers(), "parallel block preparation workers for transaction decode and sender recovery") + fs.IntVar(&cfg.workers, "workers", defaultWorkerCount, "parallel executor workers") + fs.IntVar(&cfg.executorWorkers, "executor-workers", defaultExecutorWorkers(), "parallel OCC workers inside each executor") + fs.Float64Var(&cfg.targetBlocksPerSec, "target-blocks-per-sec", 0, "input block rate cap; 0 means unlimited") + fs.DurationVar(&cfg.reportInterval, "report-interval", defaultReportInterval, "stdout and rate-gauge reporting interval; 0 disables periodic reports") + fs.StringVar(&cfg.metricsAddr, "metrics-addr", defaultMetricsAddr, "Prometheus listen address; empty disables HTTP metrics") + fs.StringVar(&cfg.resultSink, "result-sink", resultSinkDiscard, "result sink mode: discard or file") + fs.IntVar(&cfg.resultPoolSize, "result-pool-size", 0, "pooled executor BlockResult slots; 0 sizes for in-flight sink results, negative disables pooling") + fs.StringVar(&cfg.persistDir, "persist-dir", "", "directory for --result-sink=file append-only changeset and receipt files") + fs.BoolVar(&cfg.persistSync, "persist-sync", false, "fsync persistent result files from the async sink writer") + fs.IntVar(&cfg.persistBufferSize, "persist-buffer-size", defaultPersistBuffer, "buffer size in bytes for --result-sink=file") + fs.IntVar(&cfg.persistQueueSize, "persist-queue-size", 0, "record queue size for async file persistence; 0 defaults to 2*queue-size") + fs.StringVar(&cfg.cpuProfile, "cpu-profile", "", "write Go CPU profile to this file; starts after prebuild when --prebuild-blocks is set") + fs.StringVar(&cfg.heapProfile, "heap-profile", "", "write Go heap profile to this file after execution") + fs.StringVar(&cfg.traceProfile, "trace-profile", "", "write Go runtime trace to this file; starts after prebuild when --prebuild-blocks is set") + fs.StringVar(&cfg.workload, "workload", workloadTransfer, "workload type: transfer or erc20-transfer") + fs.Uint64Var(&cfg.txGasLimit, "tx-gas-limit", defaultTxGasLimit, "gas limit for each generated transaction") + fs.Uint64Var(&cfg.blockGasLimit, "block-gas-limit", 0, "block gas limit; 0 lets the executor use its maximum") + fs.BoolVar(&cfg.disableGasPriceRule, "disable-gas-price-rule", false, "disable the executor min-gas-price validity rule") + fs.BoolVar(&cfg.prebuildBlocks, "prebuild-blocks", false, "generate all bounded blocks before starting executor workers") + + if err := fs.Parse(args); err != nil { + return config{}, err + } + var err error + if cfg.chainID, err = parsePositiveBig("chain-id", *chainID); err != nil { + return config{}, err + } + if cfg.gasPrice, err = parseNonNegativeBig("gas-price-wei", *gasPrice); err != nil { + return config{}, err + } + if cfg.minGasPrice, err = parseNonNegativeBig("min-gas-price-wei", *minGasPrice); err != nil { + return config{}, err + } + if cfg.senderBalance, err = parseNonNegativeBig("sender-balance-wei", *senderBalance); err != nil { + return config{}, err + } + if cfg.transferValue, err = parseNonNegativeBig("transfer-value-wei", *transferValue); err != nil { + return config{}, err + } + if !common.IsHexAddress(*coinbase) { + return config{}, fmt.Errorf("coinbase must be a hex EVM address") + } + cfg.coinbase = common.HexToAddress(*coinbase) + if *recipient != "" { + if !common.IsHexAddress(*recipient) { + return config{}, fmt.Errorf("recipient must be a hex EVM address") + } + addr := common.HexToAddress(*recipient) + cfg.fixedRecipient = &addr + } + if !common.IsHexAddress(*erc20Contract) { + return config{}, fmt.Errorf("erc20-contract must be a hex EVM address") + } + cfg.erc20Contract = common.HexToAddress(*erc20Contract) + cfg.workload = strings.ToLower(strings.TrimSpace(cfg.workload)) + if cfg.workload != workloadTransfer && cfg.workload != workloadERC20Transfer { + return config{}, fmt.Errorf("unsupported workload %q", cfg.workload) + } + txGasLimitSet := false + fs.Visit(func(f *flag.Flag) { + if f.Name == "tx-gas-limit" { + txGasLimitSet = true + } + }) + if cfg.workload == workloadERC20Transfer && !txGasLimitSet { + cfg.txGasLimit = defaultERC20TxGasLimit + } + if cfg.txsPerBlock <= 0 { + return config{}, fmt.Errorf("txs-per-block must be positive") + } + if cfg.queueSize <= 0 { + return config{}, fmt.Errorf("queue-size must be positive") + } + if cfg.builders <= 0 { + return config{}, fmt.Errorf("builders must be positive") + } + if cfg.prepareWorkers <= 0 { + return config{}, fmt.Errorf("prepare-workers must be positive") + } + if cfg.workers <= 0 { + return config{}, fmt.Errorf("workers must be positive") + } + if cfg.executorWorkers <= 0 { + return config{}, fmt.Errorf("executor-workers must be positive") + } + if cfg.targetBlocksPerSec < 0 { + return config{}, fmt.Errorf("target-blocks-per-sec must be non-negative") + } + if cfg.reportInterval < 0 { + return config{}, fmt.Errorf("report-interval must be non-negative") + } + cfg.resultSink = strings.ToLower(strings.TrimSpace(cfg.resultSink)) + if cfg.resultSink != resultSinkDiscard && cfg.resultSink != resultSinkFile { + return config{}, fmt.Errorf("unsupported result-sink %q", cfg.resultSink) + } + if cfg.persistBufferSize <= 0 { + return config{}, fmt.Errorf("persist-buffer-size must be positive") + } + if cfg.persistQueueSize < 0 { + return config{}, fmt.Errorf("persist-queue-size must be non-negative") + } + if cfg.persistQueueSize == 0 { + cfg.persistQueueSize = 2 * cfg.queueSize + } + if cfg.resultPoolSize < 0 { + cfg.resultPoolSize = 0 + } else if cfg.resultPoolSize == 0 { + cfg.resultPoolSize = defaultResultPoolSize(cfg) + } + if cfg.resultSink == resultSinkFile && strings.TrimSpace(cfg.persistDir) == "" { + return config{}, fmt.Errorf("persist-dir is required when result-sink=file") + } + if cfg.txGasLimit == 0 { + return config{}, fmt.Errorf("tx-gas-limit must be positive") + } + if cfg.transferValue.BitLen() > 256 { + return config{}, fmt.Errorf("transfer-value-wei must fit in uint256") + } + if cfg.prebuildBlocks && cfg.blocks == 0 { + return config{}, fmt.Errorf("prebuild-blocks requires --blocks > 0") + } + if !cfg.disableGasPriceRule && cfg.gasPrice.Cmp(cfg.minGasPrice) < 0 { + return config{}, fmt.Errorf("gas-price-wei must be greater than or equal to min-gas-price-wei unless disable-gas-price-rule is set") + } + requiredBalance := new(big.Int).Mul(new(big.Int).SetUint64(cfg.txGasLimit), cfg.gasPrice) + requiredBalanceReason := "max gas cost" + if cfg.workload == workloadTransfer { + requiredBalance.Add(requiredBalance, cfg.transferValue) + requiredBalanceReason = "transfer value plus max gas cost" + } + if cfg.senderBalance.Cmp(requiredBalance) < 0 { + return config{}, fmt.Errorf("sender-balance-wei must cover %s: need at least %s", requiredBalanceReason, requiredBalance.String()) + } + return cfg, nil +} + +func defaultPrepareWorkers() int { + return runtime.GOMAXPROCS(0) +} + +func defaultExecutorWorkers() int { + workers := runtime.GOMAXPROCS(0) + if workers > 12 { + return 12 + } + return workers +} + +func defaultResultPoolSize(cfg config) int { + size := cfg.workers + 1 + if cfg.resultSink == resultSinkFile { + size += cfg.persistQueueSize + } + return size +} + +func parsePositiveBig(name, raw string) (*big.Int, error) { + v, err := parseBig(name, raw) + if err != nil { + return nil, err + } + if v.Sign() <= 0 { + return nil, fmt.Errorf("%s must be positive", name) + } + return v, nil +} + +func parseNonNegativeBig(name, raw string) (*big.Int, error) { + v, err := parseBig(name, raw) + if err != nil { + return nil, err + } + if v.Sign() < 0 { + return nil, fmt.Errorf("%s must be non-negative", name) + } + return v, nil +} + +func parseBig(name, raw string) (*big.Int, error) { + v, ok := new(big.Int).SetString(strings.TrimSpace(raw), 10) + if !ok { + return nil, fmt.Errorf("%s must be a base-10 integer", name) + } + return v, nil +} + +func run(cfg config) (err error) { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + state := newGeneratedState() + workload, err := newWorkload(cfg, state) + if err != nil { + return err + } + registry := prometheus.NewRegistry() + metrics := newLoadMetrics(registry) + sinks, err := newResultSinks(cfg, metrics) + if err != nil { + return err + } + defer func() { + closeStartedAt := time.Now() + if closeErr := sinks.Close(); closeErr != nil { + if err != nil { + fmt.Fprintf(os.Stderr, "evmonly-loadtest: result sink close: %v\n", closeErr) + return + } + err = closeErr + } + if cfg.resultSink == resultSinkFile { + printResultSinkReport(time.Since(closeStartedAt), metrics.snapshot()) + } + }() + stopSinkSignalCleanup := cleanupSinksOnContextCancel(ctx, sinks) + defer stopSinkSignalCleanup() + + var server *metricsServer + if cfg.metricsAddr != "" { + var err error + server, err = startMetricsServer(cfg.metricsAddr, registry) + if err != nil { + return err + } + defer func() { + if err := server.stop(3 * time.Second); err != nil { + fmt.Fprintf(os.Stderr, "evmonly-loadtest: metrics server shutdown: %v\n", err) + } + }() + fmt.Printf("metrics listening on http://%s/metrics\n", cfg.metricsAddr) + } + + if cfg.prebuildBlocks { + return runPrebuilt(ctx, cfg, state, workload, sinks, metrics) + } + return runStreaming(ctx, cfg, state, workload, sinks, metrics) +} + +type profileSession struct { + cpuFile *os.File + traceFile *os.File + heapPath string + cpuActive bool + traceActive bool + heapComplete bool +} + +func startProfiles(cfg config) (*profileSession, error) { + if cfg.cpuProfile == "" && cfg.heapProfile == "" && cfg.traceProfile == "" { + return nil, nil + } + session := &profileSession{heapPath: cfg.heapProfile} + if cfg.cpuProfile != "" { + file, err := createProfileFile(cfg.cpuProfile) + if err != nil { + return nil, err + } + session.cpuFile = file + if err := pprof.StartCPUProfile(file); err != nil { + _ = session.Close() + return nil, fmt.Errorf("start CPU profile: %w", err) + } + session.cpuActive = true + } + if cfg.traceProfile != "" { + file, err := createProfileFile(cfg.traceProfile) + if err != nil { + _ = session.Close() + return nil, err + } + session.traceFile = file + if err := trace.Start(file); err != nil { + _ = session.Close() + return nil, fmt.Errorf("start runtime trace: %w", err) + } + session.traceActive = true + } + return session, nil +} + +func createProfileFile(path string) (*os.File, error) { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return nil, fmt.Errorf("create profile dir for %s: %w", path, err) + } + file, err := os.Create(path) + if err != nil { + return nil, fmt.Errorf("create profile file %s: %w", path, err) + } + return file, nil +} + +func (p *profileSession) Close() error { + if p == nil { + return nil + } + var errs []error + if p.traceActive { + trace.Stop() + p.traceActive = false + } + if p.cpuActive { + pprof.StopCPUProfile() + p.cpuActive = false + } + if p.cpuFile != nil { + errs = append(errs, p.cpuFile.Close()) + p.cpuFile = nil + } + if p.traceFile != nil { + errs = append(errs, p.traceFile.Close()) + p.traceFile = nil + } + if p.heapPath != "" && !p.heapComplete { + errs = append(errs, writeHeapProfile(p.heapPath)) + p.heapComplete = true + } + return errors.Join(errs...) +} + +func writeHeapProfile(path string) error { + file, err := createProfileFile(path) + if err != nil { + return err + } + defer file.Close() + runtime.GC() + if err := pprof.WriteHeapProfile(file); err != nil { + return fmt.Errorf("write heap profile %s: %w", path, err) + } + return nil +} + +func finishProfiles(session *profileSession, err *error) { + if closeErr := session.Close(); closeErr != nil { + if *err != nil { + fmt.Fprintf(os.Stderr, "evmonly-loadtest: write profile: %v\n", closeErr) + return + } + *err = closeErr + } +} + +func cleanupSinksOnContextCancel(ctx context.Context, sinks *resultSinks) func() { + cleanupCtx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + select { + case <-ctx.Done(): + if err := sinks.Cleanup(); err != nil { + fmt.Fprintf(os.Stderr, "evmonly-loadtest: result sink cleanup: %v\n", err) + } + case <-cleanupCtx.Done(): + } + }() + return func() { + select { + case <-ctx.Done(): + default: + cancel() + } + <-done + } +} + +type blockWorkload interface { + buildBlock(context.Context, uint64) (evmonly.BlockRequest, error) +} + +func newWorkload(cfg config, state *generatedState) (blockWorkload, error) { + switch cfg.workload { + case workloadTransfer: + return newTransferWorkload(cfg, state), nil + case workloadERC20Transfer: + return newERC20TransferWorkload(cfg, state), nil + default: + return nil, fmt.Errorf("unsupported workload %q", cfg.workload) + } +} + +func runStreaming(ctx context.Context, cfg config, state *generatedState, workload blockWorkload, sinks *resultSinks, metrics *loadMetrics) (err error) { + profiles, err := startProfiles(cfg) + if err != nil { + return err + } + + blocks := make(chan blockEnvelope, cfg.queueSize) + preparedBlocks := make(chan preparedBlockEnvelope, cfg.queueSize) + reportCtx, stopReporter := context.WithCancel(ctx) + reportDone := make(chan struct{}) + go func() { + defer close(reportDone) + reportLoop(reportCtx, cfg.reportInterval, metrics, preparedBlocks) + }() + + startedAt := time.Now() + group, groupCtx := errgroup.WithContext(ctx) + executor := evmonly.NewExecutor(evmonly.Config{ + MinGasPrice: new(big.Int).Set(cfg.minGasPrice), + DisableGasPriceCheck: cfg.disableGasPriceRule, + OCCWorkers: cfg.executorWorkers, + BlockResultPoolSize: cfg.resultPoolSize, + }, evmonly.WithState(state), evmonly.WithResultSink(sinks)) + defer executor.Close() + group.Go(func() error { + defer close(blocks) + return produceBlocks(groupCtx, cfg, workload, blocks, metrics) + }) + group.Go(func() error { + defer close(preparedBlocks) + return prepareBlocks(groupCtx, cfg, executor, blocks, preparedBlocks, metrics) + }) + for workerID := 0; workerID < cfg.workers; workerID++ { + workerID := workerID + group.Go(func() error { + return executeBlocks(groupCtx, workerID, executor, preparedBlocks, metrics) + }) + } + + err = group.Wait() + stopReporter() + <-reportDone + + if errors.Is(err, context.Canceled) { + err = nil + } + printFinalReport(startedAt, metrics.snapshot()) + finishProfiles(profiles, &err) + return err +} + +func runPrebuilt(ctx context.Context, cfg config, state *generatedState, workload blockWorkload, sinks *resultSinks, metrics *loadMetrics) (err error) { + prebuildStartedAt := time.Now() + prebuilt, err := prebuildBlockRequests(ctx, cfg, workload) + if err != nil { + return err + } + state.Freeze() + prebuildElapsed := time.Since(prebuildStartedAt) + printPrebuildReport(prebuildElapsed, prebuilt, cfg.txsPerBlock) + + profiles, err := startProfiles(cfg) + if err != nil { + return err + } + + blocks := make(chan blockEnvelope, cfg.queueSize) + preparedBlocks := make(chan preparedBlockEnvelope, cfg.queueSize) + reportCtx, stopReporter := context.WithCancel(ctx) + reportDone := make(chan struct{}) + go func() { + defer close(reportDone) + reportLoop(reportCtx, cfg.reportInterval, metrics, preparedBlocks) + }() + + startedAt := time.Now() + group, groupCtx := errgroup.WithContext(ctx) + executor := evmonly.NewExecutor(evmonly.Config{ + MinGasPrice: new(big.Int).Set(cfg.minGasPrice), + DisableGasPriceCheck: cfg.disableGasPriceRule, + OCCWorkers: cfg.executorWorkers, + BlockResultPoolSize: cfg.resultPoolSize, + }, evmonly.WithState(state), evmonly.WithResultSink(sinks)) + defer executor.Close() + group.Go(func() error { + defer close(blocks) + return feedPrebuiltBlocks(groupCtx, prebuilt, blocks, metrics) + }) + group.Go(func() error { + defer close(preparedBlocks) + return prepareBlocks(groupCtx, cfg, executor, blocks, preparedBlocks, metrics) + }) + for workerID := 0; workerID < cfg.workers; workerID++ { + workerID := workerID + group.Go(func() error { + return executeBlocks(groupCtx, workerID, executor, preparedBlocks, metrics) + }) + } + + err = group.Wait() + stopReporter() + <-reportDone + + if errors.Is(err, context.Canceled) { + err = nil + } + printFinalReport(startedAt, metrics.snapshot()) + prebuilt = nil + finishProfiles(profiles, &err) + return err +} + +func produceBlocks(ctx context.Context, cfg config, workload blockWorkload, out chan<- blockEnvelope, metrics *loadMetrics) error { + var limiter *rate.Limiter + if cfg.targetBlocksPerSec > 0 { + burst := int(math.Ceil(cfg.targetBlocksPerSec)) + if burst < 1 { + burst = 1 + } + limiter = rate.NewLimiter(rate.Limit(cfg.targetBlocksPerSec), burst) + } + + var nextBlock atomic.Uint64 + group, groupCtx := errgroup.WithContext(ctx) + for builderID := 0; builderID < cfg.builders; builderID++ { + group.Go(func() error { + for { + number := nextBlock.Add(1) + if cfg.blocks != 0 && number > cfg.blocks { + return nil + } + if limiter != nil { + if err := limiter.Wait(groupCtx); err != nil { + return nil + } + } + request, err := workload.buildBlock(groupCtx, number) + if err != nil { + if groupCtx.Err() != nil { + return nil + } + return err + } + block := blockEnvelope{number: number, request: request} + select { + case out <- block: + metrics.recordInput() + case <-groupCtx.Done(): + return nil + } + } + }) + } + return group.Wait() +} + +func prebuildBlockRequests(ctx context.Context, cfg config, workload blockWorkload) ([]blockEnvelope, error) { + if cfg.blocks > uint64(maxInt()) { + return nil, fmt.Errorf("prebuild-blocks cannot allocate %d blocks on this platform", cfg.blocks) + } + prebuilt := make([]blockEnvelope, int(cfg.blocks)) + var nextBlock atomic.Uint64 + group, groupCtx := errgroup.WithContext(ctx) + for builderID := 0; builderID < cfg.builders; builderID++ { + group.Go(func() error { + for { + number := nextBlock.Add(1) + if number > cfg.blocks { + return nil + } + request, err := workload.buildBlock(groupCtx, number) + if err != nil { + if groupCtx.Err() != nil { + return nil + } + return err + } + prebuilt[number-1] = blockEnvelope{ + number: number, + request: request, + } + } + }) + } + if err := group.Wait(); err != nil { + return nil, err + } + return prebuilt, nil +} + +func feedPrebuiltBlocks(ctx context.Context, prebuilt []blockEnvelope, out chan<- blockEnvelope, metrics *loadMetrics) error { + for _, block := range prebuilt { + select { + case out <- block: + metrics.recordInput() + case <-ctx.Done(): + return nil + } + } + return nil +} + +func prepareBlocks( + ctx context.Context, + cfg config, + executor evmonly.PreparedBlockExecutor, + blocks <-chan blockEnvelope, + out chan<- preparedBlockEnvelope, + metrics *loadMetrics, +) error { + unordered := make(chan preparedBlockEnvelope, cfg.queueSize) + group, groupCtx := errgroup.WithContext(ctx) + for workerID := 0; workerID < cfg.prepareWorkers; workerID++ { + workerID := workerID + group.Go(func() error { + for { + select { + case <-groupCtx.Done(): + return nil + case block, ok := <-blocks: + if !ok { + return nil + } + prepared, err := executor.PrepareBlock(groupCtx, block.request) + if err != nil { + metrics.recordPrepareError() + if groupCtx.Err() != nil { + return nil + } + return fmt.Errorf("prepare worker %d prepare block %d: %w", workerID, block.number, err) + } + select { + case unordered <- preparedBlockEnvelope{number: block.number, block: prepared}: + metrics.recordPrepared(len(prepared.Txs)) + case <-groupCtx.Done(): + return nil + } + } + } + }) + } + done := make(chan error, 1) + go func() { + err := group.Wait() + close(unordered) + done <- err + }() + orderErr := forwardPreparedBlocksInOrder(ctx, unordered, out) + workerErr := <-done + if orderErr != nil { + return orderErr + } + return workerErr +} + +func forwardPreparedBlocksInOrder(ctx context.Context, in <-chan preparedBlockEnvelope, out chan<- preparedBlockEnvelope) error { + next := uint64(1) + pending := make(map[uint64]preparedBlockEnvelope) + for { + for { + block, ok := pending[next] + if !ok { + break + } + if err := sendPreparedBlock(ctx, out, block); err != nil { + return err + } + delete(pending, next) + next++ + } + + select { + case <-ctx.Done(): + return nil + case block, ok := <-in: + if !ok { + for len(pending) > 0 { + block, ok := pending[next] + if !ok { + if ctx.Err() != nil { + return nil + } + return fmt.Errorf("prepared block stream closed before block %d", next) + } + if err := sendPreparedBlock(ctx, out, block); err != nil { + return err + } + delete(pending, next) + next++ + } + return nil + } + if block.number < next { + return fmt.Errorf("prepared block %d arrived after block %d", block.number, next-1) + } + if block.number == next { + if err := sendPreparedBlock(ctx, out, block); err != nil { + return err + } + next++ + continue + } + if _, exists := pending[block.number]; exists { + return fmt.Errorf("duplicate prepared block %d", block.number) + } + pending[block.number] = block + } + } +} + +func sendPreparedBlock(ctx context.Context, out chan<- preparedBlockEnvelope, block preparedBlockEnvelope) error { + select { + case out <- block: + return nil + case <-ctx.Done(): + return nil + } +} + +func maxInt() int { + return int(^uint(0) >> 1) +} + +func executeBlocks( + ctx context.Context, + workerID int, + executor evmonly.PreparedBlockExecutor, + blocks <-chan preparedBlockEnvelope, + metrics *loadMetrics, +) error { + for { + select { + case <-ctx.Done(): + return nil + case block, ok := <-blocks: + if !ok { + return nil + } + result, err := executor.ExecutePreparedBlock(ctx, block.block) + if err != nil { + metrics.recordExecutionError() + if ctx.Err() != nil { + return nil + } + return fmt.Errorf("worker %d execute block %d: %w", workerID, block.number, err) + } + metrics.recordFinished(len(result.Txs), result.GasUsed, result.OCCStats) + result.Release() + } + } +} + +type transferWorkload struct { + cfg config + state *generatedState + signer ethtypes.Signer + accountCursor atomic.Uint64 +} + +func newTransferWorkload(cfg config, state *generatedState) *transferWorkload { + return &transferWorkload{ + cfg: cfg, + state: state, + signer: ethtypes.LatestSignerForChainID(cfg.chainID), + } +} + +func (w *transferWorkload) buildBlock(ctx context.Context, number uint64) (evmonly.BlockRequest, error) { + txs := make([][]byte, w.cfg.txsPerBlock) + for i := 0; i < w.cfg.txsPerBlock; i++ { + select { + case <-ctx.Done(): + return evmonly.BlockRequest{}, ctx.Err() + default: + } + accountIndex := w.accountCursor.Add(1) + raw, sender, err := w.buildTransferTx(accountIndex) + if err != nil { + return evmonly.BlockRequest{}, err + } + w.state.SetBalance(sender, w.cfg.senderBalance) + txs[i] = raw + } + return evmonly.BlockRequest{ + Context: blockContext(w.cfg, number), + Txs: txs, + }, nil +} + +func (w *transferWorkload) buildTransferTx(accountIndex uint64) ([]byte, common.Address, error) { + key, err := deterministicPrivateKey(accountIndex) + if err != nil { + return nil, common.Address{}, err + } + sender := crypto.PubkeyToAddress(key.PublicKey) + recipient := w.recipient(accountIndex) + tx := ethtypes.NewTx(ðtypes.LegacyTx{ + Nonce: 0, + GasPrice: new(big.Int).Set(w.cfg.gasPrice), + Gas: w.cfg.txGasLimit, + To: &recipient, + Value: new(big.Int).Set(w.cfg.transferValue), + }) + signed, err := ethtypes.SignTx(tx, w.signer, key) + if err != nil { + return nil, common.Address{}, err + } + raw, err := signed.MarshalBinary() + if err != nil { + return nil, common.Address{}, err + } + return raw, sender, nil +} + +func (w *transferWorkload) recipient(accountIndex uint64) common.Address { + if w.cfg.fixedRecipient != nil { + return *w.cfg.fixedRecipient + } + return addressFromSeed("sei-evmonly-loadtest-recipient", accountIndex) +} + +type erc20TransferWorkload struct { + cfg config + state *generatedState + signer ethtypes.Signer + accountCursor atomic.Uint64 +} + +var ( + erc20TransferSelector = [4]byte{0xa9, 0x05, 0x9c, 0xbb} + // Minimal ERC20-like runtime for transfer(address,uint256), with balances at + // storage slot 0 and a standard Transfer(address,address,uint256) log. + erc20TransferRuntimeCode = common.FromHex("0x60003560e01c63a9059cbb1460145760006000fd5b60243560043533600052600060205260406000208054831060805780548303905580600052600060205260406000208054830190558160005280337fddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef60206000a3600160005260206000f35b60006000fd") +) + +func newERC20TransferWorkload(cfg config, state *generatedState) *erc20TransferWorkload { + state.SetCode(cfg.erc20Contract, erc20TransferRuntimeCode) + return &erc20TransferWorkload{ + cfg: cfg, + state: state, + signer: ethtypes.LatestSignerForChainID(cfg.chainID), + } +} + +func (w *erc20TransferWorkload) buildBlock(ctx context.Context, number uint64) (evmonly.BlockRequest, error) { + txs := make([][]byte, w.cfg.txsPerBlock) + for i := 0; i < w.cfg.txsPerBlock; i++ { + select { + case <-ctx.Done(): + return evmonly.BlockRequest{}, ctx.Err() + default: + } + accountIndex := w.accountCursor.Add(1) + raw, sender, err := w.buildTransferTx(accountIndex) + if err != nil { + return evmonly.BlockRequest{}, err + } + w.state.SetBalance(sender, w.cfg.senderBalance) + w.state.SetState(w.cfg.erc20Contract, erc20BalanceSlot(sender), common.BigToHash(w.cfg.transferValue)) + txs[i] = raw + } + return evmonly.BlockRequest{ + Context: blockContext(w.cfg, number), + Txs: txs, + }, nil +} + +func (w *erc20TransferWorkload) buildTransferTx(accountIndex uint64) ([]byte, common.Address, error) { + key, err := deterministicPrivateKey(accountIndex) + if err != nil { + return nil, common.Address{}, err + } + sender := crypto.PubkeyToAddress(key.PublicKey) + recipient := w.recipient(accountIndex) + tx := ethtypes.NewTx(ðtypes.LegacyTx{ + Nonce: 0, + GasPrice: new(big.Int).Set(w.cfg.gasPrice), + Gas: w.cfg.txGasLimit, + To: &w.cfg.erc20Contract, + Value: new(big.Int), + Data: erc20TransferCalldata(recipient, w.cfg.transferValue), + }) + signed, err := ethtypes.SignTx(tx, w.signer, key) + if err != nil { + return nil, common.Address{}, err + } + raw, err := signed.MarshalBinary() + if err != nil { + return nil, common.Address{}, err + } + return raw, sender, nil +} + +func (w *erc20TransferWorkload) recipient(accountIndex uint64) common.Address { + if w.cfg.fixedRecipient != nil { + return *w.cfg.fixedRecipient + } + return addressFromSeed("sei-evmonly-loadtest-erc20-recipient", accountIndex) +} + +func erc20TransferCalldata(recipient common.Address, amount *big.Int) []byte { + data := make([]byte, 4+32+32) + copy(data[:4], erc20TransferSelector[:]) + copy(data[4+12:36], recipient.Bytes()) + amount.FillBytes(data[36:68]) + return data +} + +func erc20BalanceSlot(owner common.Address) common.Hash { + var encoded [64]byte + copy(encoded[12:32], owner.Bytes()) + return crypto.Keccak256Hash(encoded[:]) +} + +func blockContext(cfg config, number uint64) evmonly.BlockContext { + return evmonly.BlockContext{ + Number: number, + Time: uint64(time.Now().Unix()), + GasLimit: cfg.blockGasLimit, + ChainID: new(big.Int).Set(cfg.chainID), + BaseFee: big.NewInt(0), + Coinbase: cfg.coinbase, + ParentHash: hashFromSeed("sei-evmonly-loadtest-parent", number-1), + BlockHash: hashFromSeed("sei-evmonly-loadtest-block", number), + PrevRandao: hashFromSeed("sei-evmonly-loadtest-randao", number), + } +} + +func deterministicPrivateKey(index uint64) (*ecdsa.PrivateKey, error) { + var buf [16]byte + binary.BigEndian.PutUint64(buf[:8], index) + for attempt := uint64(0); ; attempt++ { + binary.BigEndian.PutUint64(buf[8:], attempt) + key, err := crypto.ToECDSA(crypto.Keccak256([]byte("sei-evmonly-loadtest-sender"), buf[:])) + if err == nil { + return key, nil + } + if attempt == ^uint64(0) { + break + } + } + return nil, fmt.Errorf("could not derive private key for account %d", index) +} + +func addressFromSeed(prefix string, index uint64) common.Address { + hash := hashFromSeed(prefix, index) + return common.BytesToAddress(hash[12:]) +} + +func hashFromSeed(prefix string, index uint64) common.Hash { + var buf [8]byte + binary.BigEndian.PutUint64(buf[:], index) + return crypto.Keccak256Hash([]byte(prefix), buf[:]) +} + +type generatedState struct { + mu sync.RWMutex + frozen atomic.Bool + balances map[common.Address]*big.Int + nonces map[common.Address]uint64 + code map[common.Address][]byte + storage map[common.Address]map[common.Hash]common.Hash +} + +var _ evmonly.StateReader = (*generatedState)(nil) + +var frozenZeroBalance = new(big.Int) + +func newGeneratedState() *generatedState { + return &generatedState{ + balances: map[common.Address]*big.Int{}, + nonces: map[common.Address]uint64{}, + code: map[common.Address][]byte{}, + storage: map[common.Address]map[common.Hash]common.Hash{}, + } +} + +func (s *generatedState) Freeze() { + s.frozen.Store(true) +} + +func (s *generatedState) GetBalance(addr common.Address) *big.Int { + if s.frozen.Load() { + if balance, ok := s.balances[addr]; ok && balance != nil { + return balance + } + return frozenZeroBalance + } + s.mu.RLock() + defer s.mu.RUnlock() + if balance, ok := s.balances[addr]; ok && balance != nil { + return new(big.Int).Set(balance) + } + return new(big.Int) +} + +func (s *generatedState) SetBalance(addr common.Address, balance *big.Int) { + s.requireMutable() + s.mu.Lock() + defer s.mu.Unlock() + if balance == nil { + s.balances[addr] = new(big.Int) + return + } + s.balances[addr] = new(big.Int).Set(balance) +} + +func (s *generatedState) GetNonce(addr common.Address) uint64 { + if s.frozen.Load() { + return s.nonces[addr] + } + s.mu.RLock() + defer s.mu.RUnlock() + return s.nonces[addr] +} + +func (s *generatedState) SetNonce(addr common.Address, nonce uint64) { + s.requireMutable() + s.mu.Lock() + defer s.mu.Unlock() + s.nonces[addr] = nonce +} + +func (s *generatedState) GetCode(addr common.Address) []byte { + if s.frozen.Load() { + return cloneBytes(s.code[addr]) + } + s.mu.RLock() + defer s.mu.RUnlock() + return cloneBytes(s.code[addr]) +} + +func (s *generatedState) SetCode(addr common.Address, code []byte) { + s.requireMutable() + s.mu.Lock() + defer s.mu.Unlock() + s.code[addr] = cloneBytes(code) +} + +func (s *generatedState) GetState(addr common.Address, key common.Hash) common.Hash { + if s.frozen.Load() { + if accountStorage, ok := s.storage[addr]; ok { + return accountStorage[key] + } + return common.Hash{} + } + s.mu.RLock() + defer s.mu.RUnlock() + if accountStorage, ok := s.storage[addr]; ok { + return accountStorage[key] + } + return common.Hash{} +} + +func (s *generatedState) SetState(addr common.Address, key common.Hash, value common.Hash) { + s.requireMutable() + s.mu.Lock() + defer s.mu.Unlock() + accountStorage, ok := s.storage[addr] + if !ok { + accountStorage = map[common.Hash]common.Hash{} + s.storage[addr] = accountStorage + } + if value == (common.Hash{}) { + delete(accountStorage, key) + return + } + accountStorage[key] = value +} + +func (s *generatedState) requireMutable() { + if s.frozen.Load() { + panic("generated state is frozen") + } +} + +func cloneBytes(v []byte) []byte { + if len(v) == 0 { + return nil + } + return append([]byte(nil), v...) +} + +type discardStateWriter struct{} + +var _ evmonly.StateWriter = (*discardStateWriter)(nil) + +func (*discardStateWriter) ApplyChangeSet(evmonly.StateChangeSet) {} + +type changeSetSink interface { + StoreChangeSet(ctx context.Context, height uint64, changeSet evmonly.StateChangeSet) error +} + +type receiptSink interface { + StoreReceipts(ctx context.Context, height uint64, receipts ethtypes.Receipts) error +} + +type resultSinks struct { + changeSets changeSetSink + receipts receiptSink + close func() error + cleanup func() error +} + +var _ evmonly.ResultSink = (*resultSinks)(nil) +var _ evmonly.BlockResultSink = (*resultSinks)(nil) + +func newResultSinks(cfg config, metrics *loadMetrics) (*resultSinks, error) { + switch cfg.resultSink { + case resultSinkDiscard: + return &resultSinks{ + changeSets: discardChangeSetSink{writer: &discardStateWriter{}}, + receipts: discardReceiptSink{}, + }, nil + case resultSinkFile: + return newFileResultSinks(cfg, metrics) + default: + return nil, fmt.Errorf("unsupported result-sink %q", cfg.resultSink) + } +} + +func (s *resultSinks) StoreChangeSet(ctx context.Context, height uint64, changeSet evmonly.StateChangeSet) error { + return s.changeSets.StoreChangeSet(ctx, height, changeSet) +} + +func (s *resultSinks) StoreReceipts(ctx context.Context, height uint64, receipts ethtypes.Receipts) error { + return s.receipts.StoreReceipts(ctx, height, receipts) +} + +func (s *resultSinks) StoreBlockResult(ctx context.Context, height uint64, result *evmonly.BlockResult, release func()) error { + if sink, ok := s.changeSets.(evmonly.BlockResultSink); ok { + return sink.StoreBlockResult(ctx, height, result, release) + } + if err := s.StoreChangeSet(ctx, height, result.ChangeSet); err != nil { + return err + } + if err := s.StoreReceipts(ctx, height, result.Receipts); err != nil { + return err + } + release() + return nil +} + +func (s *resultSinks) Close() error { + var closeErr error + if s.close == nil { + closeErr = nil + } else { + closeErr = s.close() + } + return errors.Join(closeErr, s.Cleanup()) +} + +func (s *resultSinks) Cleanup() error { + if s.cleanup == nil { + return nil + } + return s.cleanup() +} + +type discardChangeSetSink struct { + writer evmonly.StateWriter +} + +func (s discardChangeSetSink) StoreChangeSet(_ context.Context, _ uint64, changeSet evmonly.StateChangeSet) error { + s.writer.ApplyChangeSet(changeSet) + return nil +} + +type discardReceiptSink struct{} + +func (discardReceiptSink) StoreReceipts(context.Context, uint64, ethtypes.Receipts) error { + return nil +} + +type fileResultSinks struct { + changeSetFile *appendRLPFile + receiptFile *appendRLPFile + metrics *loadMetrics + cleanupMu sync.Mutex + paths []string + cleaned map[string]struct{} +} + +func newFileResultSinks(cfg config, metrics *loadMetrics) (*resultSinks, error) { + if err := os.MkdirAll(cfg.persistDir, 0o755); err != nil { + return nil, fmt.Errorf("create persist dir %s: %w", cfg.persistDir, err) + } + changeSetPath := filepath.Join(cfg.persistDir, "changesets.rlp") + receiptPath := filepath.Join(cfg.persistDir, "receipts.rlp") + files := &fileResultSinks{ + metrics: metrics, + paths: []string{changeSetPath, receiptPath}, + cleaned: map[string]struct{}{}, + } + var err error + files.changeSetFile, err = newAppendRLPFile(changeSetPath, cfg.persistBufferSize, cfg.persistSync) + if err != nil { + return nil, err + } + files.receiptFile, err = newAppendRLPFile(receiptPath, cfg.persistBufferSize, cfg.persistSync) + if err != nil { + return nil, errors.Join(err, files.Close()) + } + async := newAsyncFileResultSinks(files, cfg.persistQueueSize, metrics) + return &resultSinks{ + changeSets: async, + receipts: async, + close: async.Close, + cleanup: files.Cleanup, + }, nil +} + +func (s *fileResultSinks) Close() error { + var errs []error + if s.changeSetFile != nil { + errs = append(errs, s.changeSetFile.Close()) + } + if s.receiptFile != nil { + errs = append(errs, s.receiptFile.Close()) + } + errs = append(errs, s.Cleanup()) + return errors.Join(errs...) +} + +func (s *fileResultSinks) Cleanup() error { + s.cleanupMu.Lock() + defer s.cleanupMu.Unlock() + + var errs []error + for _, path := range s.paths { + if _, ok := s.cleaned[path]; ok { + continue + } + if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { + errs = append(errs, fmt.Errorf("remove persist file %s: %w", path, err)) + continue + } + s.cleaned[path] = struct{}{} + } + return errors.Join(errs...) +} + +func (s *fileResultSinks) WriteRecord(kind string, height uint64, value any) error { + var file *appendRLPFile + switch kind { + case resultSinkChangeSet: + file = s.changeSetFile + case resultSinkReceipts: + file = s.receiptFile + default: + return fmt.Errorf("unsupported result sink record kind %q", kind) + } + startedAt := time.Now() + bytes, err := file.WriteRecord(height, value) + elapsed := time.Since(startedAt) + if s.metrics != nil { + s.metrics.recordSinkWrite(kind, bytes, elapsed, err == nil) + } + return err +} + +type resultSinkRecord struct { + kind string + height uint64 + value any + result *evmonly.BlockResult + release func() +} + +type asyncFileResultSinks struct { + files *fileResultSinks + metrics *loadMetrics + records chan resultSinkRecord + done chan struct{} + closeErr error + close sync.Once + errMu sync.Mutex + err error +} + +func newAsyncFileResultSinks(files *fileResultSinks, queueSize int, metrics *loadMetrics) *asyncFileResultSinks { + s := &asyncFileResultSinks{ + files: files, + metrics: metrics, + records: make(chan resultSinkRecord, queueSize), + done: make(chan struct{}), + } + if metrics != nil { + metrics.setSinkQueueCapacity(queueSize) + } + go s.run() + return s +} + +func (s *asyncFileResultSinks) StoreChangeSet(ctx context.Context, height uint64, changeSet evmonly.StateChangeSet) error { + return s.enqueue(ctx, resultSinkRecord{ + kind: resultSinkChangeSet, + height: height, + value: changeSet, + }) +} + +func (s *asyncFileResultSinks) StoreReceipts(ctx context.Context, height uint64, receipts ethtypes.Receipts) error { + return s.enqueue(ctx, resultSinkRecord{ + kind: resultSinkReceipts, + height: height, + value: receipts, + }) +} + +func (s *asyncFileResultSinks) StoreBlockResult(ctx context.Context, height uint64, result *evmonly.BlockResult, release func()) error { + return s.enqueue(ctx, resultSinkRecord{ + kind: resultSinkBlock, + height: height, + result: result, + release: release, + }) +} + +func (s *asyncFileResultSinks) enqueue(ctx context.Context, record resultSinkRecord) error { + if err := s.getErr(); err != nil { + return err + } + select { + case s.records <- record: + s.recordEnqueued(record.kind) + return nil + default: + } + + startedAt := time.Now() + select { + case s.records <- record: + if s.metrics != nil { + s.metrics.recordSinkEnqueueWait(time.Since(startedAt)) + } + s.recordEnqueued(record.kind) + return nil + case <-s.done: + if err := s.getErr(); err != nil { + return err + } + return fmt.Errorf("result sink is closed") + case <-ctx.Done(): + if s.metrics != nil { + s.metrics.recordSinkEnqueueWait(time.Since(startedAt)) + } + return ctx.Err() + } +} + +func (s *asyncFileResultSinks) recordEnqueued(kind string) { + if s.metrics == nil { + return + } + if kind == resultSinkBlock { + s.metrics.recordSinkEnqueued(resultSinkChangeSet) + s.metrics.recordSinkEnqueued(resultSinkReceipts) + s.metrics.setSinkQueued(len(s.records)) + return + } + s.metrics.recordSinkEnqueued(kind) + s.metrics.setSinkQueued(len(s.records)) +} + +func (s *asyncFileResultSinks) run() { + defer close(s.done) + for record := range s.records { + if s.metrics != nil { + s.metrics.setSinkQueued(len(s.records)) + } + if err := s.writeRecord(record); err != nil { + record.releaseResult() + s.setErr(err) + s.releaseQueuedResults() + return + } + record.releaseResult() + if s.metrics != nil { + s.metrics.setSinkQueued(len(s.records)) + } + } + if s.metrics != nil { + s.metrics.setSinkQueued(0) + } +} + +func (s *asyncFileResultSinks) releaseQueuedResults() { + for { + select { + case record, ok := <-s.records: + if !ok { + return + } + record.releaseResult() + default: + return + } + } +} + +func (s *asyncFileResultSinks) writeRecord(record resultSinkRecord) error { + if record.kind == resultSinkBlock { + if err := s.files.WriteRecord(resultSinkChangeSet, record.height, record.result.ChangeSet); err != nil { + return err + } + return s.files.WriteRecord(resultSinkReceipts, record.height, record.result.Receipts) + } + return s.files.WriteRecord(record.kind, record.height, record.value) +} + +func (r resultSinkRecord) releaseResult() { + if r.release != nil { + r.release() + } +} + +func (s *asyncFileResultSinks) Close() error { + s.close.Do(func() { + close(s.records) + <-s.done + if s.metrics != nil { + s.metrics.setSinkQueued(0) + } + s.closeErr = errors.Join(s.getErr(), s.files.Close()) + }) + return s.closeErr +} + +func (s *asyncFileResultSinks) setErr(err error) { + s.errMu.Lock() + defer s.errMu.Unlock() + if s.err == nil { + s.err = err + } +} + +func (s *asyncFileResultSinks) getErr() error { + s.errMu.Lock() + defer s.errMu.Unlock() + return s.err +} + +type appendRLPFile struct { + mu sync.Mutex + file *os.File + writer *bufio.Writer + syncOnWrite bool + closed bool +} + +func newAppendRLPFile(path string, bufferSize int, syncOnWrite bool) (*appendRLPFile, error) { + file, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644) + if err != nil { + return nil, fmt.Errorf("open persist file %s: %w", path, err) + } + return &appendRLPFile{ + file: file, + writer: bufio.NewWriterSize(file, bufferSize), + syncOnWrite: syncOnWrite, + }, nil +} + +func (f *appendRLPFile) WriteRecord(height uint64, value any) (int, error) { + payload, err := rlp.EncodeToBytes(value) + if err != nil { + return 0, fmt.Errorf("encode rlp record for height %d: %w", height, err) + } + var header [16]byte + binary.BigEndian.PutUint64(header[:8], height) + binary.BigEndian.PutUint64(header[8:], uint64(len(payload))) + + f.mu.Lock() + defer f.mu.Unlock() + if f.closed { + return 0, fmt.Errorf("write record for height %d: persist file is closed", height) + } + if _, err := f.writer.Write(header[:]); err != nil { + return 0, fmt.Errorf("write record header for height %d: %w", height, err) + } + if _, err := f.writer.Write(payload); err != nil { + return 0, fmt.Errorf("write record payload for height %d: %w", height, err) + } + if err := f.writer.Flush(); err != nil { + return 0, fmt.Errorf("flush record for height %d: %w", height, err) + } + if f.syncOnWrite { + if err := f.file.Sync(); err != nil { + return 0, fmt.Errorf("sync record for height %d: %w", height, err) + } + } + return len(header) + len(payload), nil +} + +func (f *appendRLPFile) sync() error { + if err := f.writer.Flush(); err != nil { + return err + } + if err := f.file.Sync(); err != nil { + return fmt.Errorf("sync persist file: %w", err) + } + return nil +} + +func (f *appendRLPFile) Close() error { + f.mu.Lock() + defer f.mu.Unlock() + if f.closed { + return nil + } + f.closed = true + return errors.Join(f.sync(), f.file.Close()) +} + +type loadMetrics struct { + inputBlocks atomic.Uint64 + preparedBlocks atomic.Uint64 + preparedTxs atomic.Uint64 + finishedBlocks atomic.Uint64 + finishedTxs atomic.Uint64 + gasConsumed atomic.Uint64 + prepareErrors atomic.Uint64 + executionErrors atomic.Uint64 + occAttempts atomic.Uint64 + occFallbacks atomic.Uint64 + occConflicts atomic.Uint64 + sinkEnqueued atomic.Uint64 + sinkWritten atomic.Uint64 + sinkBytes atomic.Uint64 + sinkWaitNanos atomic.Uint64 + sinkWaitEvents atomic.Uint64 + sinkWriteNanos atomic.Uint64 + sinkQueued atomic.Int64 + + inputBlocksTotal prometheus.Counter + preparedBlocksTotal prometheus.Counter + preparedTxsTotal prometheus.Counter + finishedBlocksTotal prometheus.Counter + finishedTxsTotal prometheus.Counter + gasConsumedTotal prometheus.Counter + prepareErrorsTotal prometheus.Counter + executionErrorsTotal prometheus.Counter + occAttemptsTotal prometheus.Counter + occFallbacksTotal prometheus.Counter + occConflictsTotal prometheus.Counter + + occFallbackReasonTotal *prometheus.CounterVec + occConflictTotal *prometheus.CounterVec + sinkEnqueuedTotal *prometheus.CounterVec + sinkWrittenTotal *prometheus.CounterVec + sinkBytesTotal *prometheus.CounterVec + sinkEnqueueWaitTotal prometheus.Counter + sinkEnqueueWaitEvents prometheus.Counter + sinkWriteSecondsTotal *prometheus.CounterVec + + inputBlockRate prometheus.Gauge + preparedBlockRate prometheus.Gauge + preparedTxRate prometheus.Gauge + finishedBlockRate prometheus.Gauge + txRate prometheus.Gauge + gasRate prometheus.Gauge + queuedBlocks prometheus.Gauge + sinkQueuedRecords prometheus.Gauge + sinkQueueCapacity prometheus.Gauge +} + +type metricsSnapshot struct { + at time.Time + inputBlocks uint64 + preparedBlocks uint64 + preparedTxs uint64 + finishedBlocks uint64 + finishedTxs uint64 + gasConsumed uint64 + prepareErrors uint64 + executionErrors uint64 + occAttempts uint64 + occFallbacks uint64 + occConflicts uint64 + sinkEnqueued uint64 + sinkWritten uint64 + sinkBytes uint64 + sinkWaitNanos uint64 + sinkWaitEvents uint64 + sinkWriteNanos uint64 + sinkQueued int64 +} + +func newLoadMetrics(registry *prometheus.Registry) *loadMetrics { + m := &loadMetrics{ + inputBlocksTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "evmonly_loadtest_block_input_total", + Help: "Total blocks fed to the EVM-only executor input queue.", + }), + preparedBlocksTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "evmonly_loadtest_block_prepared_total", + Help: "Total blocks decoded and sender-recovered before EVM-only executor execution.", + }), + preparedTxsTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "evmonly_loadtest_transactions_prepared_total", + Help: "Total transactions decoded and sender-recovered before EVM-only executor execution.", + }), + finishedBlocksTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "evmonly_loadtest_block_finished_total", + Help: "Total blocks that finished EVM-only executor execution.", + }), + finishedTxsTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "evmonly_loadtest_transactions_finished_total", + Help: "Total transactions that finished EVM-only executor execution.", + }), + gasConsumedTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "evmonly_loadtest_gas_consumed_total", + Help: "Total EVM gas consumed by finished blocks.", + }), + prepareErrorsTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "evmonly_loadtest_prepare_errors_total", + Help: "Total block preparation errors returned while decoding transactions or recovering senders.", + }), + executionErrorsTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "evmonly_loadtest_execution_errors_total", + Help: "Total block execution errors returned by the EVM-only executor.", + }), + occAttemptsTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "evmonly_loadtest_occ_attempts_total", + Help: "Total blocks executed with optimistic concurrency control.", + }), + occFallbacksTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "evmonly_loadtest_occ_fallbacks_total", + Help: "Total OCC blocks that fell back to sequential execution.", + }), + occConflictsTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "evmonly_loadtest_occ_conflicts_total", + Help: "Total OCC conflict accesses observed before sequential fallback.", + }), + occFallbackReasonTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "evmonly_loadtest_occ_fallback_reasons_total", + Help: "OCC fallback count by reason.", + }, []string{"reason"}), + occConflictTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "evmonly_loadtest_occ_conflict_keys_total", + Help: "OCC conflict accesses by access type, state kind, address, and slot.", + }, []string{"access", "kind", "address", "slot"}), + sinkEnqueuedTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "evmonly_loadtest_result_sink_records_enqueued_total", + Help: "Total records accepted by the result sink.", + }, []string{"kind"}), + sinkWrittenTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "evmonly_loadtest_result_sink_records_written_total", + Help: "Total records written by the result sink.", + }, []string{"kind"}), + sinkBytesTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "evmonly_loadtest_result_sink_bytes_written_total", + Help: "Total bytes written by the result sink, including record framing.", + }, []string{"kind"}), + sinkEnqueueWaitTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "evmonly_loadtest_result_sink_enqueue_wait_seconds_total", + Help: "Total time executor workers spent blocked enqueueing result sink records.", + }), + sinkEnqueueWaitEvents: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "evmonly_loadtest_result_sink_enqueue_wait_events_total", + Help: "Total result sink enqueue attempts that had to wait for queue capacity or cancellation.", + }), + sinkWriteSecondsTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "evmonly_loadtest_result_sink_write_seconds_total", + Help: "Total time spent by result sink writers serializing, writing, flushing, and optionally syncing records.", + }, []string{"kind"}), + inputBlockRate: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "evmonly_loadtest_block_input_throughput", + Help: "Most recent measured block input throughput in blocks per second.", + }), + preparedBlockRate: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "evmonly_loadtest_block_prepared_throughput", + Help: "Most recent measured block preparation throughput in blocks per second.", + }), + preparedTxRate: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "evmonly_loadtest_transactions_prepared_per_second", + Help: "Most recent measured transaction decode and sender recovery throughput in transactions per second.", + }), + finishedBlockRate: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "evmonly_loadtest_block_finished_throughput", + Help: "Most recent measured block completion throughput in blocks per second.", + }), + txRate: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "evmonly_loadtest_transactions_per_second", + Help: "Most recent measured transaction execution throughput in transactions per second.", + }), + gasRate: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "evmonly_loadtest_gas_consumed_per_second", + Help: "Most recent measured gas consumption throughput in gas per second.", + }), + queuedBlocks: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "evmonly_loadtest_queued_blocks", + Help: "Blocks currently waiting in the executor input queue.", + }), + sinkQueuedRecords: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "evmonly_loadtest_result_sink_queued_records", + Help: "Persistent result sink records currently waiting for the async writer.", + }), + sinkQueueCapacity: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "evmonly_loadtest_result_sink_queue_capacity", + Help: "Capacity of the persistent result sink async record queue.", + }), + } + registry.MustRegister( + m.inputBlocksTotal, + m.preparedBlocksTotal, + m.preparedTxsTotal, + m.finishedBlocksTotal, + m.finishedTxsTotal, + m.gasConsumedTotal, + m.prepareErrorsTotal, + m.executionErrorsTotal, + m.occAttemptsTotal, + m.occFallbacksTotal, + m.occConflictsTotal, + m.occFallbackReasonTotal, + m.occConflictTotal, + m.sinkEnqueuedTotal, + m.sinkWrittenTotal, + m.sinkBytesTotal, + m.sinkEnqueueWaitTotal, + m.sinkEnqueueWaitEvents, + m.sinkWriteSecondsTotal, + m.inputBlockRate, + m.preparedBlockRate, + m.preparedTxRate, + m.finishedBlockRate, + m.txRate, + m.gasRate, + m.queuedBlocks, + m.sinkQueuedRecords, + m.sinkQueueCapacity, + ) + return m +} + +func (m *loadMetrics) recordInput() { + m.inputBlocks.Add(1) + m.inputBlocksTotal.Inc() +} + +func (m *loadMetrics) recordPrepared(txCount int) { + m.preparedBlocks.Add(1) + m.preparedTxs.Add(uint64(txCount)) + m.preparedBlocksTotal.Inc() + m.preparedTxsTotal.Add(float64(txCount)) +} + +func (m *loadMetrics) recordPrepareError() { + m.prepareErrors.Add(1) + m.prepareErrorsTotal.Inc() +} + +func (m *loadMetrics) recordFinished(txCount int, gasUsed uint64, occStats evmonly.OCCStats) { + m.finishedBlocks.Add(1) + m.finishedTxs.Add(uint64(txCount)) + m.gasConsumed.Add(gasUsed) + m.finishedBlocksTotal.Inc() + m.finishedTxsTotal.Add(float64(txCount)) + m.gasConsumedTotal.Add(float64(gasUsed)) + m.recordOCC(occStats) +} + +func (m *loadMetrics) recordOCC(stats evmonly.OCCStats) { + if !stats.Attempted { + return + } + m.occAttempts.Add(1) + m.occAttemptsTotal.Inc() + if stats.Fallback { + reason := stats.FallbackReason + if reason == "" { + reason = "unknown" + } + m.occFallbacks.Add(1) + m.occFallbacksTotal.Inc() + m.occFallbackReasonTotal.WithLabelValues(reason).Inc() + } + if stats.ConflictCount == 0 { + return + } + m.occConflicts.Add(stats.ConflictCount) + m.occConflictsTotal.Add(float64(stats.ConflictCount)) + for _, conflict := range stats.ConflictSamples { + m.occConflictTotal.WithLabelValues( + conflict.Access, + conflict.Kind, + conflict.Address.Hex(), + conflictSlotLabel(conflict), + ).Add(float64(conflict.Count)) + } +} + +func conflictSlotLabel(conflict evmonly.OCCConflictCount) string { + if conflict.Kind != "storage" { + return "" + } + return conflict.Slot.Hex() +} + +func (m *loadMetrics) recordExecutionError() { + m.executionErrors.Add(1) + m.executionErrorsTotal.Inc() +} + +func (m *loadMetrics) recordSinkEnqueued(kind string) { + m.sinkEnqueued.Add(1) + m.sinkEnqueuedTotal.WithLabelValues(kind).Inc() +} + +func (m *loadMetrics) recordSinkEnqueueWait(elapsed time.Duration) { + if elapsed <= 0 { + return + } + m.sinkWaitNanos.Add(uint64(elapsed.Nanoseconds())) + m.sinkWaitEvents.Add(1) + m.sinkEnqueueWaitTotal.Add(elapsed.Seconds()) + m.sinkEnqueueWaitEvents.Inc() +} + +func (m *loadMetrics) recordSinkWrite(kind string, bytes int, elapsed time.Duration, completed bool) { + if elapsed > 0 { + m.sinkWriteNanos.Add(uint64(elapsed.Nanoseconds())) + m.sinkWriteSecondsTotal.WithLabelValues(kind).Add(elapsed.Seconds()) + } + if !completed { + return + } + m.sinkWritten.Add(1) + m.sinkBytes.Add(uint64(bytes)) + m.sinkWrittenTotal.WithLabelValues(kind).Inc() + m.sinkBytesTotal.WithLabelValues(kind).Add(float64(bytes)) +} + +func (m *loadMetrics) setSinkQueued(records int) { + m.sinkQueued.Store(int64(records)) + m.sinkQueuedRecords.Set(float64(records)) +} + +func (m *loadMetrics) setSinkQueueCapacity(records int) { + m.sinkQueueCapacity.Set(float64(records)) +} + +func (m *loadMetrics) snapshot() metricsSnapshot { + return metricsSnapshot{ + at: time.Now(), + inputBlocks: m.inputBlocks.Load(), + preparedBlocks: m.preparedBlocks.Load(), + preparedTxs: m.preparedTxs.Load(), + finishedBlocks: m.finishedBlocks.Load(), + finishedTxs: m.finishedTxs.Load(), + gasConsumed: m.gasConsumed.Load(), + prepareErrors: m.prepareErrors.Load(), + executionErrors: m.executionErrors.Load(), + occAttempts: m.occAttempts.Load(), + occFallbacks: m.occFallbacks.Load(), + occConflicts: m.occConflicts.Load(), + sinkEnqueued: m.sinkEnqueued.Load(), + sinkWritten: m.sinkWritten.Load(), + sinkBytes: m.sinkBytes.Load(), + sinkWaitNanos: m.sinkWaitNanos.Load(), + sinkWaitEvents: m.sinkWaitEvents.Load(), + sinkWriteNanos: m.sinkWriteNanos.Load(), + sinkQueued: m.sinkQueued.Load(), + } +} + +func (m *loadMetrics) setRates(prev, curr metricsSnapshot, queued int) { + elapsed := curr.at.Sub(prev.at).Seconds() + if elapsed <= 0 { + return + } + inputRate := float64(curr.inputBlocks-prev.inputBlocks) / elapsed + preparedRate := float64(curr.preparedBlocks-prev.preparedBlocks) / elapsed + preparedTxRate := float64(curr.preparedTxs-prev.preparedTxs) / elapsed + finishedRate := float64(curr.finishedBlocks-prev.finishedBlocks) / elapsed + txRate := float64(curr.finishedTxs-prev.finishedTxs) / elapsed + gasRate := float64(curr.gasConsumed-prev.gasConsumed) / elapsed + m.inputBlockRate.Set(inputRate) + m.preparedBlockRate.Set(preparedRate) + m.preparedTxRate.Set(preparedTxRate) + m.finishedBlockRate.Set(finishedRate) + m.txRate.Set(txRate) + m.gasRate.Set(gasRate) + m.queuedBlocks.Set(float64(queued)) +} + +func reportLoop(ctx context.Context, interval time.Duration, metrics *loadMetrics, blocks <-chan preparedBlockEnvelope) { + if interval == 0 { + <-ctx.Done() + return + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + + prev := metrics.snapshot() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + curr := metrics.snapshot() + queued := len(blocks) + metrics.setRates(prev, curr, queued) + elapsed := curr.at.Sub(prev.at).Seconds() + if elapsed <= 0 { + prev = curr + continue + } + sinkWaitSeconds := float64(curr.sinkWaitNanos-prev.sinkWaitNanos) / float64(time.Second) + sinkWriteSeconds := float64(curr.sinkWriteNanos-prev.sinkWriteNanos) / float64(time.Second) + fmt.Printf( + "input_blocks/s=%.2f prepared_blocks/s=%.2f prepared_tx/s=%.2f finished_blocks/s=%.2f tx/s=%.2f gas/s=%.2f queued_blocks=%d sink_queue=%d sink_enqueue_wait/s=%.6f sink_write/s=%.6f totals(input_blocks=%d prepared_blocks=%d prepared_txs=%d finished_blocks=%d txs=%d gas=%d prepare_errors=%d errors=%d occ_attempts=%d occ_fallbacks=%d occ_conflicts=%d sink_enqueued=%d sink_written=%d)\n", + float64(curr.inputBlocks-prev.inputBlocks)/elapsed, + float64(curr.preparedBlocks-prev.preparedBlocks)/elapsed, + float64(curr.preparedTxs-prev.preparedTxs)/elapsed, + float64(curr.finishedBlocks-prev.finishedBlocks)/elapsed, + float64(curr.finishedTxs-prev.finishedTxs)/elapsed, + float64(curr.gasConsumed-prev.gasConsumed)/elapsed, + queued, + curr.sinkQueued, + sinkWaitSeconds/elapsed, + sinkWriteSeconds/elapsed, + curr.inputBlocks, + curr.preparedBlocks, + curr.preparedTxs, + curr.finishedBlocks, + curr.finishedTxs, + curr.gasConsumed, + curr.prepareErrors, + curr.executionErrors, + curr.occAttempts, + curr.occFallbacks, + curr.occConflicts, + curr.sinkEnqueued, + curr.sinkWritten, + ) + prev = curr + } + } +} + +func printFinalReport(startedAt time.Time, snapshot metricsSnapshot) { + elapsed := snapshot.at.Sub(startedAt).Seconds() + if elapsed <= 0 { + elapsed = 1 + } + fmt.Printf( + "complete elapsed=%s input_blocks=%d prepared_blocks=%d prepared_txs=%d finished_blocks=%d txs=%d gas=%d prepare_errors=%d errors=%d occ_attempts=%d occ_fallbacks=%d occ_conflicts=%d sink_queue=%d sink_enqueued=%d sink_written=%d sink_bytes=%d sink_enqueue_wait=%s sink_enqueue_wait_events=%d sink_write=%s avg_input_blocks/s=%.2f avg_prepared_blocks/s=%.2f avg_prepared_tx/s=%.2f avg_finished_blocks/s=%.2f avg_tx/s=%.2f avg_gas/s=%.2f\n", + snapshot.at.Sub(startedAt).Round(time.Millisecond), + snapshot.inputBlocks, + snapshot.preparedBlocks, + snapshot.preparedTxs, + snapshot.finishedBlocks, + snapshot.finishedTxs, + snapshot.gasConsumed, + snapshot.prepareErrors, + snapshot.executionErrors, + snapshot.occAttempts, + snapshot.occFallbacks, + snapshot.occConflicts, + snapshot.sinkQueued, + snapshot.sinkEnqueued, + snapshot.sinkWritten, + snapshot.sinkBytes, + time.Duration(snapshot.sinkWaitNanos).Round(time.Microsecond), + snapshot.sinkWaitEvents, + time.Duration(snapshot.sinkWriteNanos).Round(time.Microsecond), + float64(snapshot.inputBlocks)/elapsed, + float64(snapshot.preparedBlocks)/elapsed, + float64(snapshot.preparedTxs)/elapsed, + float64(snapshot.finishedBlocks)/elapsed, + float64(snapshot.finishedTxs)/elapsed, + float64(snapshot.gasConsumed)/elapsed, + ) +} + +func printResultSinkReport(closeElapsed time.Duration, snapshot metricsSnapshot) { + fmt.Printf( + "result sink close elapsed=%s sink_queue=%d sink_enqueued=%d sink_written=%d sink_bytes=%d sink_enqueue_wait=%s sink_enqueue_wait_events=%d sink_write=%s\n", + closeElapsed.Round(time.Millisecond), + snapshot.sinkQueued, + snapshot.sinkEnqueued, + snapshot.sinkWritten, + snapshot.sinkBytes, + time.Duration(snapshot.sinkWaitNanos).Round(time.Microsecond), + snapshot.sinkWaitEvents, + time.Duration(snapshot.sinkWriteNanos).Round(time.Microsecond), + ) +} + +func printPrebuildReport(elapsed time.Duration, blocks []blockEnvelope, txsPerBlock int) { + seconds := elapsed.Seconds() + if seconds <= 0 { + seconds = 1 + } + txCount := len(blocks) * txsPerBlock + fmt.Printf( + "prebuild complete elapsed=%s blocks=%d txs=%d build_blocks/s=%.2f build_tx/s=%.2f\n", + elapsed.Round(time.Millisecond), + len(blocks), + txCount, + float64(len(blocks))/seconds, + float64(txCount)/seconds, + ) +} + +type metricsServer struct { + server *http.Server + done chan error +} + +func startMetricsServer(addr string, registry *prometheus.Registry) (*metricsServer, error) { + listener, err := net.Listen("tcp", addr) + if err != nil { + return nil, fmt.Errorf("listen for metrics on %s: %w", addr, err) + } + mux := http.NewServeMux() + mux.Handle("/metrics", promhttp.HandlerFor(registry, promhttp.HandlerOpts{})) + mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte("ok\n")) + }) + server := &http.Server{ + Addr: addr, + Handler: mux, + ReadHeaderTimeout: 3 * time.Second, + } + ms := &metricsServer{server: server, done: make(chan error, 1)} + go func() { + err := server.Serve(listener) + if errors.Is(err, http.ErrServerClosed) { + err = nil + } + ms.done <- err + }() + return ms, nil +} + +func (s *metricsServer) stop(timeout time.Duration) error { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + if err := s.server.Shutdown(ctx); err != nil { + return err + } + return <-s.done +} diff --git a/giga/evmonly/cmd/evmonly-loadtest/main_test.go b/giga/evmonly/cmd/evmonly-loadtest/main_test.go new file mode 100644 index 0000000000..9ead8d07e8 --- /dev/null +++ b/giga/evmonly/cmd/evmonly-loadtest/main_test.go @@ -0,0 +1,340 @@ +package main + +import ( + "context" + "encoding/binary" + "errors" + "fmt" + "math/big" + "os" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + ethtypes "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/rlp" + "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/require" + + "github.com/sei-protocol/sei-chain/giga/evmonly" +) + +func TestTransferWorkloadExecutesAgainstEVMOnlyExecutor(t *testing.T) { + cfg, err := parseConfig([]string{ + "--metrics-addr=", + "--txs-per-block=4", + }) + require.NoError(t, err) + + state := newGeneratedState() + workload := newTransferWorkload(cfg, state) + request, err := workload.buildBlock(context.Background(), 1) + require.NoError(t, err) + + executor := evmonly.NewExecutor(evmonly.Config{ + MinGasPrice: cfg.minGasPrice, + }, evmonly.WithState(state)) + result, err := executor.ExecuteBlock(context.Background(), request) + require.NoError(t, err) + + require.Len(t, result.Txs, cfg.txsPerBlock) + require.Len(t, result.Receipts, cfg.txsPerBlock) + require.Equal(t, uint64(cfg.txsPerBlock)*cfg.txGasLimit, result.GasUsed) + for _, tx := range result.Txs { + require.Equal(t, ethtypes.ReceiptStatusSuccessful, tx.Status) + require.NoError(t, tx.Err) + } + + writer := &discardStateWriter{} + writer.ApplyChangeSet(result.ChangeSet) + require.NoError(t, discardReceiptSink{}.StoreReceipts(context.Background(), request.Context.Number, result.Receipts)) +} + +func TestERC20TransferWorkloadExecutesAgainstEVMOnlyExecutor(t *testing.T) { + cfg, err := parseConfig([]string{ + "--metrics-addr=", + "--workload=erc20-transfer", + "--txs-per-block=4", + "--gas-price-wei=0", + "--min-gas-price-wei=0", + }) + require.NoError(t, err) + require.Equal(t, uint64(defaultERC20TxGasLimit), cfg.txGasLimit) + + state := newGeneratedState() + workload, err := newWorkload(cfg, state) + require.NoError(t, err) + request, err := workload.buildBlock(context.Background(), 1) + require.NoError(t, err) + + executor := evmonly.NewExecutor(evmonly.Config{ + MinGasPrice: cfg.minGasPrice, + OCCWorkers: cfg.executorWorkers, + }, evmonly.WithState(state)) + result, err := executor.ExecuteBlock(context.Background(), request) + require.NoError(t, err) + + require.Len(t, result.Txs, cfg.txsPerBlock) + require.Len(t, result.Receipts, cfg.txsPerBlock) + require.NotEmpty(t, result.ChangeSet.Storage) + require.True(t, result.OCCStats.Attempted) + require.False(t, result.OCCStats.Fallback) + for _, tx := range result.Txs { + require.Equal(t, ethtypes.ReceiptStatusSuccessful, tx.Status) + require.NoError(t, tx.Err) + require.Greater(t, tx.GasUsed, uint64(21_000)) + require.Len(t, tx.Logs, 1) + } + for _, receipt := range result.Receipts { + require.Len(t, receipt.Logs, 1) + } + + applyGeneratedStateChangeSet(state, result.ChangeSet) + transferWorkload := workload.(*erc20TransferWorkload) + for i := uint64(1); i <= uint64(cfg.txsPerBlock); i++ { + key, err := deterministicPrivateKey(i) + require.NoError(t, err) + sender := crypto.PubkeyToAddress(key.PublicKey) + recipient := transferWorkload.recipient(i) + require.Equal(t, common.Hash{}, state.GetState(cfg.erc20Contract, erc20BalanceSlot(sender))) + require.Equal(t, common.BigToHash(cfg.transferValue), state.GetState(cfg.erc20Contract, erc20BalanceSlot(recipient))) + } +} + +func applyGeneratedStateChangeSet(state *generatedState, changeSet evmonly.StateChangeSet) { + for _, change := range changeSet.Balances { + state.SetBalance(change.Address, change.Balance) + } + for _, change := range changeSet.Nonces { + state.SetNonce(change.Address, change.Nonce) + } + for _, change := range changeSet.Code { + if change.Delete { + state.SetCode(change.Address, nil) + } else { + state.SetCode(change.Address, change.Code) + } + } + for _, change := range changeSet.Storage { + value := change.Value + if change.Delete { + value = common.Hash{} + } + state.SetState(change.Address, change.Key, value) + } +} + +func TestPrebuildBlocksRequiresBoundedRun(t *testing.T) { + _, err := parseConfig([]string{ + "--prebuild-blocks", + }) + require.ErrorContains(t, err, "prebuild-blocks requires --blocks > 0") +} + +func TestRunPrebuiltBlocks(t *testing.T) { + cfg, err := parseConfig([]string{ + "--metrics-addr=", + "--report-interval=0", + "--prebuild-blocks", + "--blocks=2", + "--txs-per-block=2", + }) + require.NoError(t, err) + require.NoError(t, run(cfg)) +} + +func TestFileResultSinkWritesRLPRecordsAndCleansUpOnCancel(t *testing.T) { + dir := t.TempDir() + cfg, err := parseConfig([]string{ + "--metrics-addr=", + "--result-sink=file", + "--persist-dir=" + dir, + }) + require.NoError(t, err) + metrics := newLoadMetrics(prometheus.NewRegistry()) + sinks, err := newResultSinks(cfg, metrics) + require.NoError(t, err) + + changePath := filepath.Join(dir, "changesets.rlp") + receiptPath := filepath.Join(dir, "receipts.rlp") + writtenChangeSet := evmonly.StateChangeSet{ + Balances: []evmonly.BalanceChange{{ + Address: common.HexToAddress("0x0000000000000000000000000000000000000001"), + Balance: big.NewInt(7), + }}, + } + writtenReceipts := ethtypes.Receipts{{ + Type: ethtypes.LegacyTxType, + Status: ethtypes.ReceiptStatusSuccessful, + TxHash: common.HexToHash("0x01"), + }} + require.NoError(t, sinks.StoreChangeSet(context.Background(), 1, writtenChangeSet)) + require.NoError(t, sinks.StoreReceipts(context.Background(), 1, writtenReceipts)) + require.Eventually(t, func() bool { + return metrics.snapshot().sinkWritten == 2 + }, time.Second, time.Millisecond) + + requireFileExists(t, changePath) + requireFileExists(t, receiptPath) + var changeSet evmonly.StateChangeSet + height := readPersistedRLPRecord(t, changePath, &changeSet) + require.Equal(t, uint64(1), height) + require.NotEmpty(t, changeSet.Balances) + + var receipts ethtypes.Receipts + height = readPersistedRLPRecord(t, receiptPath, &receipts) + require.Equal(t, uint64(1), height) + require.Len(t, receipts, 1) + + ctx, cancel := context.WithCancel(context.Background()) + stopCleanup := cleanupSinksOnContextCancel(ctx, sinks) + cancel() + stopCleanup() + requireNoFileExists(t, changePath) + requireNoFileExists(t, receiptPath) + require.NoError(t, sinks.Close()) + requireNoFileExists(t, changePath) + requireNoFileExists(t, receiptPath) +} + +func TestAsyncFileResultSinkReportsMetrics(t *testing.T) { + dir := t.TempDir() + cfg, err := parseConfig([]string{ + "--metrics-addr=", + "--result-sink=file", + "--persist-dir=" + dir, + "--persist-queue-size=1", + }) + require.NoError(t, err) + metrics := newLoadMetrics(prometheus.NewRegistry()) + sinks, err := newResultSinks(cfg, metrics) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + require.NoError(t, sinks.StoreChangeSet(ctx, 1, evmonly.StateChangeSet{})) + require.NoError(t, sinks.StoreReceipts(ctx, 1, ethtypes.Receipts{})) + require.NoError(t, sinks.Close()) + + snapshot := metrics.snapshot() + require.Equal(t, uint64(2), snapshot.sinkEnqueued) + require.Equal(t, uint64(2), snapshot.sinkWritten) + require.Equal(t, int64(0), snapshot.sinkQueued) + require.Greater(t, snapshot.sinkBytes, uint64(0)) +} + +func TestAsyncFileResultSinkStoresBlockResultAndReleases(t *testing.T) { + dir := t.TempDir() + cfg, err := parseConfig([]string{ + "--metrics-addr=", + "--result-sink=file", + "--persist-dir=" + dir, + "--persist-queue-size=1", + }) + require.NoError(t, err) + metrics := newLoadMetrics(prometheus.NewRegistry()) + sinks, err := newResultSinks(cfg, metrics) + require.NoError(t, err) + + result := &evmonly.BlockResult{ + ChangeSet: evmonly.StateChangeSet{ + Balances: []evmonly.BalanceChange{{ + Address: common.HexToAddress("0x0000000000000000000000000000000000000001"), + Balance: big.NewInt(7), + }}, + }, + Receipts: ethtypes.Receipts{{ + Type: ethtypes.LegacyTxType, + Status: ethtypes.ReceiptStatusSuccessful, + TxHash: common.HexToHash("0x02"), + }}, + } + var released atomic.Bool + require.NoError(t, sinks.StoreBlockResult(context.Background(), 1, result, func() { + released.Store(true) + })) + require.Eventually(t, func() bool { + return metrics.snapshot().sinkWritten == 2 && released.Load() + }, time.Second, time.Millisecond) + require.NoError(t, sinks.Close()) + + snapshot := metrics.snapshot() + require.Equal(t, uint64(2), snapshot.sinkEnqueued) + require.Equal(t, uint64(2), snapshot.sinkWritten) +} + +func TestRunPrebuiltBlocksWithFileResultSinkCleansUp(t *testing.T) { + dir := t.TempDir() + cfg, err := parseConfig([]string{ + "--metrics-addr=", + "--report-interval=0", + "--prebuild-blocks", + "--blocks=2", + "--txs-per-block=2", + "--gas-price-wei=0", + "--min-gas-price-wei=0", + "--result-sink=file", + "--persist-dir=" + dir, + }) + require.NoError(t, err) + require.NoError(t, run(cfg)) + requireNoFileExists(t, filepath.Join(dir, "changesets.rlp")) + requireNoFileExists(t, filepath.Join(dir, "receipts.rlp")) +} + +func readPersistedRLPRecord(t *testing.T, path string, out any) uint64 { + t.Helper() + data, err := os.ReadFile(path) + require.NoError(t, err) + require.GreaterOrEqual(t, len(data), 16) + height := binary.BigEndian.Uint64(data[:8]) + payloadLen := binary.BigEndian.Uint64(data[8:16]) + require.LessOrEqual(t, payloadLen, uint64(len(data)-16)) + payload := data[16 : 16+payloadLen] + require.NoError(t, rlp.DecodeBytes(payload, out), fmt.Sprintf("decode %s", path)) + return height +} + +func requireFileExists(t *testing.T, path string) { + t.Helper() + _, err := os.Stat(path) + require.NoError(t, err) +} + +func requireNoFileExists(t *testing.T, path string) { + t.Helper() + _, err := os.Stat(path) + require.Truef(t, errors.Is(err, os.ErrNotExist), "expected %s to be removed, got %v", path, err) +} + +func BenchmarkExecuteTransferBlock(b *testing.B) { + cfg, err := parseConfig([]string{ + "--metrics-addr=", + "--txs-per-block=1000", + }) + require.NoError(b, err) + + state := newGeneratedState() + workload := newTransferWorkload(cfg, state) + request, err := workload.buildBlock(context.Background(), 1) + require.NoError(b, err) + executor := evmonly.NewExecutor(evmonly.Config{ + MinGasPrice: cfg.minGasPrice, + }, evmonly.WithState(state)) + + b.ReportAllocs() + b.SetBytes(int64(cfg.txsPerBlock)) + b.ResetTimer() + for i := 0; i < b.N; i++ { + result, err := executor.ExecuteBlock(context.Background(), request) + if err != nil { + b.Fatal(err) + } + if len(result.Txs) != cfg.txsPerBlock { + b.Fatalf("expected %d txs, got %d", cfg.txsPerBlock, len(result.Txs)) + } + } +} diff --git a/giga/evmonly/executor.go b/giga/evmonly/executor.go index d9ad18abf5..87291bf4d9 100644 --- a/giga/evmonly/executor.go +++ b/giga/evmonly/executor.go @@ -228,6 +228,9 @@ func (e *Executor) executeTx( baseFee *big.Int, ) (TxResult, *ethtypes.Receipt, error) { tx := p.Tx + if err := validateSupportedTx(tx); err != nil { + return TxResult{Hash: tx.Hash(), Sender: p.Sender, To: tx.To(), Err: err}, nil, err + } if !e.cfg.DisableGasPriceCheck && e.cfg.MinGasPrice != nil { // MinGasPrice is block-validity policy; unlike EVM call failures, it // does not produce a receipt for an otherwise invalid block. @@ -366,6 +369,13 @@ func (e *Executor) chainConfig(ctx BlockContext) *params.ChainConfig { return &cfg } +func validateSupportedTx(tx *ethtypes.Transaction) error { + if tx.Type() == ethtypes.BlobTxType { + return errUnsupportedBlobTx + } + return nil +} + func effectiveGasPrice(tx *ethtypes.Transaction, baseFee *big.Int) *big.Int { if baseFee == nil { return tx.GasPrice() @@ -376,4 +386,7 @@ func effectiveGasPrice(tx *ethtypes.Transaction, baseFee *big.Int) *big.Int { return tx.GasPrice() } -var errInsufficientGasPrice = fmt.Errorf("insufficient gas price") +var ( + errInsufficientGasPrice = fmt.Errorf("insufficient gas price") + errUnsupportedBlobTx = fmt.Errorf("blob transactions require block-level blob gas accounting") +) diff --git a/giga/evmonly/executor_test.go b/giga/evmonly/executor_test.go index 50bf94f4b1..d0a38dc83f 100644 --- a/giga/evmonly/executor_test.go +++ b/giga/evmonly/executor_test.go @@ -13,6 +13,7 @@ import ( "github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/params" + "github.com/holiman/uint256" "github.com/stretchr/testify/require" "github.com/sei-protocol/sei-chain/giga/evmonly/precompiles" @@ -183,6 +184,59 @@ func TestExecutorDynamicFeeTx(t *testing.T) { require.Equal(t, big.NewInt(11), state.GetBalance(recipient)) } +func TestExecutorRejectsBlobTxUntilBlockAccountingIsWired(t *testing.T) { + chainID := big.NewInt(713715) + key, err := crypto.GenerateKey() + require.NoError(t, err) + sender := crypto.PubkeyToAddress(key.PublicKey) + recipient := testAddress(0xb5) + blobHash := common.Hash{0x01} + + state := NewMemoryState() + state.SetBalance(sender, big.NewInt(1_000_000_000_000_000)) + rawTx := signBlobTxWithFees( + t, + key, + chainID, + 0, + recipient, + big.NewInt(1), + nil, + big.NewInt(1), + big.NewInt(3), + big.NewInt(3), + 100_000, + []common.Hash{blobHash}, + ) + ctx := blockContext(chainID) + ctx.BaseFee = big.NewInt(2) + ctx.BlobBaseFee = big.NewInt(3) + + executor := NewExecutor(Config{MinGasPrice: big.NewInt(0)}, WithState(state)) + result, err := executor.ExecuteBlock(context.Background(), BlockRequest{ + Context: ctx, + Txs: [][]byte{rawTx}, + }) + + require.ErrorIs(t, err, errUnsupportedBlobTx) + require.Nil(t, result) + require.Equal(t, big.NewInt(0), state.GetBalance(recipient)) + + tx, sender, err := parseTx(rawTx, ethtypes.LatestSignerForChainID(chainID)) + require.NoError(t, err) + result, err = executor.ExecutePreparedBlock(context.Background(), PreparedBlock{ + Context: ctx, + Txs: []PreparedTx{{ + Tx: tx, + Sender: sender, + }}, + }) + + require.ErrorIs(t, err, errUnsupportedBlobTx) + require.Nil(t, result) + require.Equal(t, big.NewInt(0), state.GetBalance(recipient)) +} + func TestExecutorOCCNonConflictingTransfersMatchSequential(t *testing.T) { chainID := big.NewInt(713715) txCount := 12 @@ -1198,6 +1252,40 @@ func signDynamicFeeTxWithFees(t *testing.T, key *ecdsa.PrivateKey, chainID *big. return raw } +func signBlobTxWithFees( + t *testing.T, + key *ecdsa.PrivateKey, + chainID *big.Int, + nonce uint64, + to common.Address, + value *big.Int, + data []byte, + gasTipCap *big.Int, + gasFeeCap *big.Int, + blobFeeCap *big.Int, + gas uint64, + blobHashes []common.Hash, +) []byte { + t.Helper() + tx := ethtypes.NewTx(ðtypes.BlobTx{ + ChainID: uint256.MustFromBig(chainID), + Nonce: nonce, + GasTipCap: uint256.MustFromBig(gasTipCap), + GasFeeCap: uint256.MustFromBig(gasFeeCap), + Gas: gas, + To: to, + Value: uint256.MustFromBig(value), + Data: data, + BlobFeeCap: uint256.MustFromBig(blobFeeCap), + BlobHashes: append([]common.Hash(nil), blobHashes...), + }) + signed, err := ethtypes.SignTx(tx, ethtypes.LatestSignerForChainID(chainID), key) + require.NoError(t, err) + raw, err := signed.MarshalBinary() + require.NoError(t, err) + return raw +} + func blockContext(chainID *big.Int) BlockContext { return BlockContext{ Number: 1, diff --git a/giga/evmonly/parser.go b/giga/evmonly/parser.go index 99b366fc61..16a6581f5d 100644 --- a/giga/evmonly/parser.go +++ b/giga/evmonly/parser.go @@ -20,6 +20,9 @@ func parseBlockTxs(ctx context.Context, txs [][]byte, signer ethtypes.Signer) ([ if err != nil { return nil, fmt.Errorf("parse tx %d: %w", i, err) } + if err := validateSupportedTx(tx); err != nil { + return nil, fmt.Errorf("parse tx %d: %w", i, err) + } parsed[i] = PreparedTx{Tx: tx, Sender: sender} } return parsed, nil