feat(runner): saturated-window GPS metric + always-emit sequencer results

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"`

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)

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 {

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

runner/network/types/types.go

@@ -3,6 +3,7 @@ package types
 import (
 	"crypto/ecdsa"
 	"math/big"
+	"sort"
 	"strings"
 	"time"
 
@@ -168,13 +169,99 @@ 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"
 	GetPayloadLatencyMetric            = "latency/get_payload"
 	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,
 		},
 	}
 }