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") }