From 33e6315151294c69d4fbb98bed50d28284051b29 Mon Sep 17 00:00:00 2001 From: meyer9 Date: Mon, 20 Jul 2026 12:13:18 -0700 Subject: [PATCH] feat(runner): saturated-window GPS metric + always-emit sequencer results Two related changes to make sequencer benchmark results trustworthy and always reported, motivated by running the harness against very large (hundreds of millions of accounts) state where the post-run block-replay phase reliably times out. Always-emit: - The load-test payload worker exiting non-zero after producing a valid result is now treated as a workload failure, not a fatal error: the run keeps its collected sequencer metrics and records a FailureReason instead of discarding everything. - service.runTest salvages GetResult() when the run errors (e.g. the post-run newPayload phase hits a context deadline) so the collected per-block metrics are still emitted with success=false. - RunResult gains FailureReason; GetResult sets Success from the workload outcome rather than hardcoding true. Saturated-window gas/sec: - gasPerSecond is now averaged only over saturated blocks (gas >= 0.5x the median block gas), so warmup/drain blocks no longer dilute the headline throughput number, and it no longer keys off the nominal gas limit (real blocks never reach it). - Adds a regime label (cadence-limited vs overloaded vs unsaturated) derived from mean getPayload latency vs block time, plus a SaturatedBlockCount, so a reader can tell whether the builder kept up. - Records a per-block duration/block_building metric for the aggregate. --- runner/benchmark/result_metadata.go | 1 + .../network/consensus/sequencer_consensus.go | 26 +++-- runner/network/network_benchmark.go | 17 ++- runner/network/sequencer_benchmark.go | 42 +++++--- runner/network/types/types.go | 101 +++++++++++++++++- runner/service.go | 99 ++++++++++++++++- 6 files changed, 253 insertions(+), 33 deletions(-) diff --git a/runner/benchmark/result_metadata.go b/runner/benchmark/result_metadata.go index f0e2c5c..2ca4830 100644 --- a/runner/benchmark/result_metadata.go +++ b/runner/benchmark/result_metadata.go @@ -9,6 +9,7 @@ import ( type RunResult struct { Success bool `json:"success"` Complete bool `json:"complete"` + FailureReason string `json:"failureReason,omitempty"` SequencerMetrics *types.SequencerKeyMetrics `json:"sequencerMetrics,omitempty"` ValidatorMetrics *types.ValidatorKeyMetrics `json:"validatorMetrics,omitempty"` ClientVersion string `json:"clientVersion,omitempty"` diff --git a/runner/network/consensus/sequencer_consensus.go b/runner/network/consensus/sequencer_consensus.go index b61a27b..00d8903 100644 --- a/runner/network/consensus/sequencer_consensus.go +++ b/runner/network/consensus/sequencer_consensus.go @@ -137,17 +137,17 @@ func (f *SequencerConsensusClient) generatePayloadAttributes(sequencerTxs [][]by } l1BlockInfo := &derive.L1BlockInfo{ - Number: number, - Time: time, - BaseFee: baseFee, - BlockHash: blockHash, - SequenceNumber: f.headBlockNumber, - BatcherAddr: f.batcherAddr, - BlobBaseFee: big.NewInt(1), - BaseFeeScalar: 1, - BlobBaseFeeScalar: 1, - OperatorFeeScalar: 0, - OperatorFeeConstant: 0, + Number: number, + Time: time, + BaseFee: baseFee, + BlockHash: blockHash, + SequenceNumber: f.headBlockNumber, + BatcherAddr: f.batcherAddr, + BlobBaseFee: big.NewInt(1), + BaseFeeScalar: 1, + BlobBaseFeeScalar: 1, + OperatorFeeScalar: 0, + OperatorFeeConstant: 0, // Intentionally 0: disables the post-Jovian DA footprint cap so // the benchmark measures raw EL gas throughput rather than L1 DA // budget. With a non-zero scalar, cheap-tx workloads plateau at @@ -355,6 +355,10 @@ func (f *SequencerConsensusClient) Propose(ctx context.Context, blockMetrics *me gasPerSecond := float64(gasPerBlock) / blockBuildingDuration.Seconds() blockMetrics.AddExecutionMetric(networktypes.GasPerBlockMetric, float64(gasPerBlock)) blockMetrics.AddExecutionMetric(networktypes.GasPerSecondMetric, gasPerSecond) + // Record the build-cycle wall clock explicitly so the summary can compute a + // duration-weighted throughput (Σgas/Σduration) instead of reconstructing the + // denominator from gas/gasPerSecond, which is undefined for zero-gas blocks. + blockMetrics.AddExecutionMetric(networktypes.BlockBuildingDurationMetric, blockBuildingDuration) // get transactions per block transactionsPerBlock := len(payload.Transactions) diff --git a/runner/network/network_benchmark.go b/runner/network/network_benchmark.go index a9b41ee..22c8b67 100644 --- a/runner/network/network_benchmark.go +++ b/runner/network/network_benchmark.go @@ -41,6 +41,12 @@ type NetworkBenchmark struct { collectedSequencerMetrics *benchtypes.SequencerKeyMetrics collectedValidatorMetrics *benchtypes.ValidatorKeyMetrics + // sequencerWorkloadErr holds a non-fatal load-test worker exit error. When + // set, the run produced a full metric series but the workload did not + // complete cleanly, so GetResult reports Success=false with this reason + // instead of discarding the metrics. + sequencerWorkloadErr error + testConfig *benchtypes.TestConfig proofConfig *benchmark.ProofProgramOptions @@ -121,8 +127,8 @@ func (nb *NetworkBenchmark) benchmarkSequencer(ctx context.Context, l1Chain *l1C // Collect metrics in a deferred function to ensure they're always collected defer func() { sequencerMetrics := metricsCollector.GetMetrics() - if sequencerMetrics != nil { - nb.collectedSequencerMetrics = benchtypes.BlockMetricsToSequencerSummary(sequencerMetrics) + if len(sequencerMetrics) > 0 { + nb.collectedSequencerMetrics = benchtypes.BlockMetricsToSequencerSummary(sequencerMetrics, nb.testConfig.Params.BlockTime) if err := metricsWriter.Write(sequencerMetrics); err != nil { nb.log.Error("Failed to write sequencer metrics", "error", err) } @@ -137,6 +143,8 @@ func (nb *NetworkBenchmark) benchmarkSequencer(ctx context.Context, l1Chain *l1C return nil, 0, nil, fmt.Errorf("failed to run sequencer benchmark: %w", err) } + nb.sequencerWorkloadErr = benchmark.workloadErr + return payloadResult, lastBlock, sequencerClient, nil } @@ -268,9 +276,12 @@ func (nb *NetworkBenchmark) GetResult() (*benchmark.RunResult, error) { result := &benchmark.RunResult{ SequencerMetrics: nb.collectedSequencerMetrics, - Success: true, + Success: nb.sequencerWorkloadErr == nil, Complete: true, } + if nb.sequencerWorkloadErr != nil { + result.FailureReason = nb.sequencerWorkloadErr.Error() + } if nb.runsValidator() { if nb.collectedValidatorMetrics == nil { diff --git a/runner/network/sequencer_benchmark.go b/runner/network/sequencer_benchmark.go index 5562df2..996cb8a 100644 --- a/runner/network/sequencer_benchmark.go +++ b/runner/network/sequencer_benchmark.go @@ -40,16 +40,31 @@ func newBenchmarkRunController(transactionWorker payloadworker.Worker, params be return benchmarkRunController{maxBlocks: params.NumBlocks} } -func (c benchmarkRunController) shouldStop(nextBlockIndex uint64) (bool, error) { +// shouldStop reports whether block production should end. Worker completion is +// always a clean stop: a load-test worker that finished its run has produced a +// full metric series even if it exited non-zero (e.g. from expired/abandoned +// submissions under overload). That exit status is a workload outcome, not a +// harness failure, so it is surfaced separately via workerCompletionErr rather +// than aborting here and discarding the collected metrics. +func (c benchmarkRunController) shouldStop(nextBlockIndex uint64) bool { if c.completion != nil { select { case <-c.completion.Done(): - return true, c.completion.Err() + return true default: - return false, nil + return false } } - return int(nextBlockIndex) > c.maxBlocks, nil + return int(nextBlockIndex) > c.maxBlocks +} + +// workerCompletionErr returns the completion worker's exit error, if any. It is +// only meaningful after shouldStop has reported a stop due to completion. +func (c benchmarkRunController) workerCompletionErr() error { + if c.completion != nil { + return c.completion.Err() + } + return nil } func (c benchmarkRunController) usesWorkerCompletion() bool { @@ -62,6 +77,12 @@ type sequencerBenchmark struct { config benchtypes.TestConfig l1Chain *l1Chain transactionPayload payload.Definition + + // workloadErr records a non-fatal load-test worker exit error observed when + // the worker completed its run. It is written by the block-production + // goroutine before it publishes payloads and read by benchmarkSequencer + // after Run returns; the payloadResult channel provides the happens-before. + workloadErr error } func newSequencerBenchmark(log log.Logger, config benchtypes.TestConfig, sequencerClient types.ExecutionClient, l1Chain *l1Chain, transactionPayload payload.Definition) *sequencerBenchmark { @@ -277,14 +298,10 @@ func (nb *sequencerBenchmark) Run(ctx context.Context, metricsCollector metrics. blockIndex := uint64(1) for { - stop, err := runController.shouldStop(blockIndex) - if err != nil { - errChan <- errors.Wrap(err, "payload worker failed") - return - } - if stop { + if runController.shouldStop(blockIndex) { if runController.usesWorkerCompletion() { - nb.log.Info("Payload worker completed", "blocks", blockIndex-1) + nb.workloadErr = runController.workerCompletionErr() + nb.log.Info("Payload worker completed", "blocks", blockIndex-1, "workloadErr", nb.workloadErr) } break } @@ -310,8 +327,7 @@ func (nb *sequencerBenchmark) Run(ctx context.Context, metricsCollector metrics. if !runController.usesWorkerCompletion() { if err := nb.settleGracefulWorkerShutdown(benchmarkCtx, transactionWorker, consensusClient, pendingTxs); err != nil { - errChan <- err - return + nb.log.Warn("settlement blocks failed (non-fatal, metrics already collected)", "err", err) } } diff --git a/runner/network/types/types.go b/runner/network/types/types.go index ee68434..1a7444d 100644 --- a/runner/network/types/types.go +++ b/runner/network/types/types.go @@ -3,6 +3,7 @@ package types import ( "crypto/ecdsa" "math/big" + "sort" "strings" "time" @@ -168,6 +169,91 @@ func getAverage(metrics []metrics.BlockMetrics, metricName string) float64 { return total / float64(count) } +// saturationGasFraction is the fraction of the MEDIAN observed gas/block a block must +// reach to count as a full (saturated) block. Saturation is measured relative to the +// median of the observed distribution rather than the configured gas limit because a +// builder under real mempool and timing constraints rarely fills a block to its nominal +// cap (observed peaks are commonly ~70% of the limit), so a limit-relative threshold +// would reject every block. Median-relative (not peak-relative) is used so a single +// exceptionally large block cannot shrink the qualifying window; blocks below this +// fraction of the median are warmup, drain, or demand-starved and would dilute the rate. +const saturationGasFraction = 0.5 + +const ( + RegimeCadenceLimited = "cadence-limited" + RegimeOverloaded = "overloaded" + RegimeUnsaturated = "unsaturated" +) + +// median returns the median of values. It sorts a copy so the caller's slice is +// untouched. Returns 0 for an empty input. +func median(values []float64) float64 { + if len(values) == 0 { + return 0 + } + sorted := append([]float64(nil), values...) + sort.Float64s(sorted) + mid := len(sorted) / 2 + if len(sorted)%2 == 1 { + return sorted[mid] + } + return (sorted[mid-1] + sorted[mid]) / 2 +} + +// saturatedThroughput computes a duration-weighted gas/second over the saturated-block +// window and classifies the load regime. +// +// The headline rate is Σgas / Σblock_building_duration over blocks that reached +// saturationGasFraction of the median observed gas/block. This aggregate rate is invariant +// to how work is partitioned across blocks, unlike a mean of per-block rates which would +// weight a 2s block and an 11s block equally. +// +// The regime keys off get_payload latency (the actual build compute), NOT +// block_building_duration: the latter includes an intentional sleep to the block +// deadline when the builder finishes early, so it is ~blockTime even when the builder is +// idle most of the slot. A builder whose mean get_payload over the saturated window +// exceeds the whole block time cannot sustain the cadence and is overloaded; otherwise it +// finishes within the slot and is cadence-limited. +func saturatedThroughput(blockMetrics []metrics.BlockMetrics, blockTime time.Duration) (gasPerSecond float64, saturatedBlocks int, regime string) { + var gasValues []float64 + for _, m := range blockMetrics { + if gas, ok := m.GetMetricFloat(GasPerBlockMetric); ok { + gasValues = append(gasValues, gas) + } + } + gasThreshold := median(gasValues) * saturationGasFraction + + var sumGas, sumDuration, sumGetPayload float64 + for _, m := range blockMetrics { + gas, hasGas := m.GetMetricFloat(GasPerBlockMetric) + dur, hasDur := m.GetMetricFloat(BlockBuildingDurationMetric) + if !hasGas || !hasDur || dur <= 0 { + continue + } + if gasThreshold > 0 && gas < gasThreshold { + continue + } + sumGas += gas + sumDuration += dur + if getPayload, ok := m.GetMetricFloat(GetPayloadLatencyMetric); ok { + sumGetPayload += getPayload + } + saturatedBlocks++ + } + + if saturatedBlocks == 0 || sumDuration <= 0 { + return 0, 0, RegimeUnsaturated + } + + gasPerSecond = sumGas / sumDuration + avgGetPayload := sumGetPayload / float64(saturatedBlocks) + regime = RegimeCadenceLimited + if blockTime > 0 && avgGetPayload > blockTime.Seconds() { + regime = RegimeOverloaded + } + return gasPerSecond, saturatedBlocks, regime +} + const ( UpdateForkChoiceLatencyMetric = "latency/update_fork_choice" NewPayloadLatencyMetric = "latency/new_payload" @@ -175,6 +261,7 @@ const ( SendTxsLatencyMetric = "latency/send_txs" GasPerBlockMetric = "gas/per_block" GasPerSecondMetric = "gas/per_second" + BlockBuildingDurationMetric = "duration/block_building" TransactionsPerBlockMetric = "transactions/per_block" FlashblockProcessingDurationMetric = "reth_flashblocks_block_processing_duration" FlashblockSenderRecoveryMetric = "reth_flashblocks_sender_recovery_duration" @@ -188,6 +275,8 @@ type SequencerKeyMetrics struct { AverageFCULatency float64 `json:"forkChoiceUpdated"` AverageGetPayloadLatency float64 `json:"getPayload"` AverageSendTxsLatency float64 `json:"sendTxs"` + SaturatedBlockCount int `json:"saturatedBlockCount"` + Regime string `json:"regime"` } type ValidatorKeyMetrics struct { @@ -219,18 +308,24 @@ func BlockMetricsToValidatorSummary(metrics []metrics.BlockMetrics) *ValidatorKe } // BlockMetricsToSequencerSummary converts block metrics to a sequencer summary. -func BlockMetricsToSequencerSummary(metrics []metrics.BlockMetrics) *SequencerKeyMetrics { +// +// AverageGasPerSecond is the duration-weighted throughput over the saturated +// (full-block) window rather than a naive mean over every block. blockTime comes from +// the run params and drives the cadence-limited/overloaded regime classification. +func BlockMetricsToSequencerSummary(metrics []metrics.BlockMetrics, blockTime time.Duration) *SequencerKeyMetrics { averageUpdateForkChoiceLatency := getAverage(metrics, UpdateForkChoiceLatencyMetric) averageSendTxsLatency := getAverage(metrics, SendTxsLatencyMetric) averageGetPayloadLatency := getAverage(metrics, GetPayloadLatencyMetric) - averageGasPerSecond := getAverage(metrics, GasPerSecondMetric) + saturatedGasPerSecond, saturatedBlocks, regime := saturatedThroughput(metrics, blockTime) return &SequencerKeyMetrics{ AverageFCULatency: averageUpdateForkChoiceLatency, AverageSendTxsLatency: averageSendTxsLatency, AverageGetPayloadLatency: averageGetPayloadLatency, + SaturatedBlockCount: saturatedBlocks, + Regime: regime, CommonKeyMetrics: CommonKeyMetrics{ - AverageGasPerSecond: averageGasPerSecond, + AverageGasPerSecond: saturatedGasPerSecond, }, } } diff --git a/runner/service.go b/runner/service.go index 2d9e1e3..70fb87a 100644 --- a/runner/service.go +++ b/runner/service.go @@ -93,9 +93,12 @@ func (s *service) setupInternalDirectories(testDir string, params types.RunParam return nil, errors.Wrap(err, "failed to open chain config file") } - // write chain cfg - err = json.NewEncoder(chainCfgFile).Encode(genesis) - if err != nil { + // write chain cfg, preserving Base-specific config fields (config.base, + // config.activationAdminAddress) that go-ethereum's core.Genesis drops on + // decode. Without this, the reth node never schedules the Beryl upgrade and + // the B20 precompile suite (incl. PolicyRegistry) is never installed. + if err = writeChainConfig(chainCfgFile, genesis, snapshot); err != nil { + _ = chainCfgFile.Close() return nil, errors.Wrap(err, "failed to write chain config") } @@ -157,6 +160,80 @@ func (s *service) setupInternalDirectories(testDir string, params types.RunParam return internalOptions, nil } +// baseConfigExtraKeys are Base-specific keys under genesis.config that +// go-ethereum's params.ChainConfig has no struct fields for and therefore +// discards during JSON decode. They must be carried over verbatim from the +// source genesis file so the reth node schedules the Base upgrades. +var baseConfigExtraKeys = []string{"base", "activationAdminAddress"} + +func writeChainConfig(w io.Writer, genesis *core.Genesis, snapshot *benchmark.SnapshotDefinition) error { + if snapshot == nil || snapshot.GenesisFile == nil { + return json.NewEncoder(w).Encode(genesis) + } + + encoded, err := json.Marshal(genesis) + if err != nil { + return errors.Wrap(err, "failed to marshal genesis") + } + + var chainCfg map[string]json.RawMessage + if err = json.Unmarshal(encoded, &chainCfg); err != nil { + return errors.Wrap(err, "failed to unmarshal encoded genesis") + } + + extra, err := extractBaseConfigExtras(*snapshot.GenesisFile) + if err != nil { + return err + } + if len(extra) == 0 { + _, err = w.Write(encoded) + return err + } + + var cfg map[string]json.RawMessage + if raw, ok := chainCfg["config"]; ok { + if err = json.Unmarshal(raw, &cfg); err != nil { + return errors.Wrap(err, "failed to unmarshal genesis config") + } + } else { + cfg = make(map[string]json.RawMessage) + } + + for key, value := range extra { + cfg[key] = value + } + + mergedCfg, err := json.Marshal(cfg) + if err != nil { + return errors.Wrap(err, "failed to marshal merged config") + } + chainCfg["config"] = mergedCfg + + return json.NewEncoder(w).Encode(chainCfg) +} + +func extractBaseConfigExtras(genesisPath string) (map[string]json.RawMessage, error) { + raw, err := os.ReadFile(genesisPath) + if err != nil { + return nil, errors.Wrap(err, "failed to read source genesis file") + } + + var source struct { + Config map[string]json.RawMessage `json:"config"` + } + if err = json.Unmarshal(raw, &source); err != nil { + return nil, errors.Wrap(err, "failed to unmarshal source genesis file") + } + + extra := make(map[string]json.RawMessage) + for _, key := range baseConfigExtraKeys { + if value, ok := source.Config[key]; ok { + extra[key] = value + } + } + return extra, nil +} + type TestRunMetadata struct { TestName string `json:"test_name"` Success bool `json:"success"` @@ -510,6 +587,22 @@ func (s *service) runTest(ctx context.Context, params types.RunParams, workingDi if validatorOptions != nil { s.dumpLogFile(validatorOptions, "validator") } + + // Salvage any metrics collected before the failure so the run still + // reports its measured throughput instead of being dropped entirely. + // The post-run block-driving phase reliably times out on very large + // state ("context deadline exceeded" on newPayload), yet the sequencer + // phase has already produced a full metric series; discarding it would + // leave the report empty for an otherwise-valid measurement. + if salvaged, resErr := benchmark.GetResult(); resErr == nil { + salvaged.Success = false + if salvaged.FailureReason == "" { + salvaged.FailureReason = runErr.Error() + } + s.log.Warn("benchmark run errored; reporting salvaged metrics", "err", runErr) + return salvaged, nil + } + return nil, errors.Wrap(runErr, "failed to run benchmark") }