feat: allow specifying test data dirs

.gitignore

@@ -19,3 +19,4 @@ coverage/
 node_modules
 yarn-debug.log*
 yarn-error.log*
+.env*.local

report/.yarnrc.yml

@@ -1,3 +1,5 @@
+enableGlobalCache: false
+
 enableTelemetry: false
 
 nodeLinker: node-modules

runner/benchmark/definition.go

@@ -91,9 +91,15 @@ type BenchmarkConfig struct {
 	TransactionPayloads []payload.Definition `yaml:"payloads"`
 }
 
+type DatadirConfig struct {
+	Sequencer *string `yaml:"sequencer"`
+	Validator *string `yaml:"validator"`
+}
+
 // TestDefinition is the user-facing YAML configuration for specifying a
 // matrix of benchmark runs.
 type TestDefinition struct {
+	Datadir      *DatadirConfig       `yaml:"datadirs"`
 	Snapshot     *SnapshotDefinition  `yaml:"snapshot"`
 	Metrics      *ThresholdConfig     `yaml:"metrics"`
 	Tags         *map[string]string   `yaml:"tags"`

runner/benchmark/matrix.go

@@ -13,6 +13,7 @@ type ThresholdConfig struct {
 // TestPlan represents a list of test runs to be executed.
 type TestPlan struct {
 	Runs         []TestRun
+	Datadir      *DatadirConfig
 	Snapshot     *SnapshotDefinition
 	ProofProgram *ProofProgramOptions
 	Thresholds   *ThresholdConfig
@@ -37,6 +38,7 @@ func NewTestPlanFromConfig(c TestDefinition, testFileName string, config *Benchm
 
 	return &TestPlan{
 		Runs:         testRuns,
+		Datadir:      c.Datadir,
 		Snapshot:     c.Snapshot,
 		ProofProgram: proofProgram,
 		Thresholds:   c.Metrics,

runner/benchmark/portmanager/ports.go

@@ -13,6 +13,7 @@ const (
 	AuthELPortPurpose
 	ELMetricsPortPurpose
 	BuilderMetricsPortPurpose
+	P2PPortPurpose
 )
 
 type PortManager interface {

runner/benchmark/snapshots.go

@@ -12,7 +12,7 @@ type SnapshotManager interface {
 	// EnsureSnapshot ensures that a snapshot exists for the given node type and
 	// role. If it does not exist, it will create it using the given snapshot
 	// definition. It returns the path to the snapshot.
-	EnsureSnapshot(definition SnapshotDefinition, nodeType string, role string) (string, error)
+	EnsureSnapshot(datadirsConfig *DatadirConfig, definition SnapshotDefinition, nodeType string, role string) (string, error)
 }
 
 type snapshotStoragePath struct {
@@ -58,7 +58,7 @@ func NewSnapshotManager(snapshotsDir string) SnapshotManager {
 	}
 }
 
-func (b *benchmarkDatadirState) EnsureSnapshot(definition SnapshotDefinition, nodeType string, role string) (string, error) {
+func (b *benchmarkDatadirState) EnsureSnapshot(datadirsConfig *DatadirConfig, definition SnapshotDefinition, nodeType string, role string) (string, error) {
 	snapshotDatadir := snapshotStoragePath{
 		nodeType: nodeType,
 		role:     role,
@@ -69,10 +69,15 @@ func (b *benchmarkDatadirState) EnsureSnapshot(definition SnapshotDefinition, no
 		return datadir, nil
 	}
 
-	hashCommand := sha256.New().Sum([]byte(definition.Command))
-
-	snapshotPath := filepath.Join(b.snapshotsDir, fmt.Sprintf("%s_%s_%x", nodeType, role, hashCommand[:12]))
-
+	var snapshotPath string
+	if datadirsConfig != nil && role == "validator" && datadirsConfig.Validator != nil {
+		snapshotPath = *datadirsConfig.Validator
+	} else if datadirsConfig != nil && role == "sequencer" && datadirsConfig.Sequencer != nil {
+		snapshotPath = *datadirsConfig.Sequencer
+	} else {
+		hashCommand := sha256.New().Sum([]byte(definition.Command))
+		snapshotPath = filepath.Join(b.snapshotsDir, fmt.Sprintf("%s_%s_%x", nodeType, role, hashCommand[:12]))
+	}
 	// Create a new datadir for this snapshot.
 	err := definition.CreateSnapshot(nodeType, snapshotPath)
 	if err != nil {

runner/clients/reth/client.go

@@ -38,6 +38,7 @@ type RethClient struct {
 	ports       portmanager.PortManager
 	metricsPort uint64
 	rpcPort     uint64
+	p2pPort     uint64
 	authRPCPort uint64
 
 	stdout io.WriteCloser
@@ -79,6 +80,7 @@ func (r *RethClient) Run(ctx context.Context, cfg *types.RuntimeConfig) error {
 	args = append(args, "--datadir", r.options.DataDirPath)
 
 	r.rpcPort = r.ports.AcquirePort("reth", portmanager.ELPortPurpose)
+	r.p2pPort = r.ports.AcquirePort("reth", portmanager.P2PPortPurpose)
 	r.authRPCPort = r.ports.AcquirePort("reth", portmanager.AuthELPortPurpose)
 	r.metricsPort = r.ports.AcquirePort("reth", portmanager.ELMetricsPortPurpose)
 
@@ -90,6 +92,8 @@ func (r *RethClient) Run(ctx context.Context, cfg *types.RuntimeConfig) error {
 	args = append(args, "--authrpc.jwtsecret", r.options.JWTSecretPath)
 	args = append(args, "--metrics", fmt.Sprintf("%d", r.metricsPort))
 	args = append(args, "--engine.state-provider-metrics")
+	args = append(args, "--disable-discovery")
+	args = append(args, "--port", fmt.Sprintf("%d", r.p2pPort))
 	args = append(args, "-vvv")
 
 	// increase mempool size
@@ -200,6 +204,7 @@ func (r *RethClient) Stop() {
 	r.ports.ReleasePort(r.rpcPort)
 	r.ports.ReleasePort(r.authRPCPort)
 	r.ports.ReleasePort(r.metricsPort)
+	r.ports.ReleasePort(r.p2pPort)
 
 	r.stdout = nil
 	r.stderr = nil

runner/network/consensus/sequencer_consensus.go

@@ -184,6 +184,7 @@ func (f *SequencerConsensusClient) generatePayloadAttributes(sequencerTxs [][]by
 
 	root := crypto.Keccak256Hash([]byte("fake-beacon-block-root"), big.NewInt(int64(1)).Bytes())
 
+	minBaseFee := uint64(1)
 	payloadAttrs := &eth.PayloadAttributes{
 		Timestamp:             eth.Uint64Quantity(timestamp),
 		PrevRandao:            eth.Bytes32{},
@@ -194,6 +195,7 @@ func (f *SequencerConsensusClient) generatePayloadAttributes(sequencerTxs [][]by
 		ParentBeaconBlockRoot: &root,
 		NoTxPool:              false,
 		EIP1559Params:         &b8,
+		MinBaseFee:            &minBaseFee,
 	}
 
 	return payloadAttrs, &root, nil

runner/network/network_benchmark.go

@@ -3,6 +3,7 @@ package network
 import (
 	"context"
 	"fmt"
+	"math/big"
 	"os"
 	"path"
 
@@ -12,12 +13,15 @@ import (
 	"github.com/base/base-bench/runner/clients/types"
 	"github.com/base/base-bench/runner/config"
 	"github.com/base/base-bench/runner/payload"
+	ethTypes "github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/crypto"
 
 	"github.com/base/base-bench/runner/logger"
 	"github.com/base/base-bench/runner/metrics"
 
 	benchtypes "github.com/base/base-bench/runner/network/types"
 	"github.com/ethereum/go-ethereum/beacon/engine"
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/pkg/errors"
 )
@@ -69,36 +73,25 @@ func (nb *NetworkBenchmark) Run(ctx context.Context) error {
 	}
 
 	// Benchmark the sequencer first to build payloads
-	payloads, firstTestBlock, err := nb.benchmarkSequencer(ctx, l1Chain)
+	payloads, lastSetupBlock, sequencerClient, err := nb.benchmarkSequencer(ctx, l1Chain)
 	if err != nil {
 		return fmt.Errorf("failed to run sequencer benchmark: %w", err)
 	}
 
 	// Benchmark the validator to sync the payloads
-	if err := nb.benchmarkValidator(ctx, payloads, firstTestBlock, l1Chain); err != nil {
+	if err := nb.benchmarkValidator(ctx, payloads, lastSetupBlock, l1Chain, sequencerClient); err != nil {
 		return fmt.Errorf("failed to run validator benchmark: %w", err)
 	}
 
 	return nil
 }
 
-func (nb *NetworkBenchmark) benchmarkSequencer(ctx context.Context, l1Chain *l1Chain) ([]engine.ExecutableData, uint64, error) {
+func (nb *NetworkBenchmark) benchmarkSequencer(ctx context.Context, l1Chain *l1Chain) ([]engine.ExecutableData, uint64, types.ExecutionClient, error) {
 	sequencerClient, err := setupNode(ctx, nb.log, nb.testConfig.Params, nb.sequencerOptions, nb.ports)
 	if err != nil {
-		return nil, 0, fmt.Errorf("failed to setup sequencer node: %w", err)
+		return nil, 0, nil, fmt.Errorf("failed to setup sequencer node: %w", err)
 	}
 
-	// Ensure client is stopped even if benchmark fails
-	defer func() {
-		currentHeader, err := sequencerClient.Client().HeaderByNumber(ctx, nil)
-		if err != nil {
-			nb.log.Error("Failed to get current block number", "error", err)
-		} else {
-			nb.log.Info("Sequencer node stopped at block", "number", currentHeader.Number.Uint64(), "hash", currentHeader.Hash().Hex())
-		}
-		sequencerClient.Stop()
-	}()
-
 	// Create metrics collector and writer
 	metricsCollector := sequencerClient.MetricsCollector()
 	metricsWriter := metrics.NewFileMetricsWriter(nb.sequencerOptions.MetricsPath)
@@ -115,12 +108,21 @@ func (nb *NetworkBenchmark) benchmarkSequencer(ctx context.Context, l1Chain *l1C
 	}()
 
 	benchmark := newSequencerBenchmark(nb.log, *nb.testConfig, sequencerClient, l1Chain, nb.transactionPayload)
-	return benchmark.Run(ctx, metricsCollector)
+
+	executionData, lastBlock, err := benchmark.Run(ctx, metricsCollector)
+
+	if err != nil {
+		sequencerClient.Stop()
+		return nil, 0, nil, err
+	}
+
+	return executionData, lastBlock, sequencerClient, err
 }
 
-func (nb *NetworkBenchmark) benchmarkValidator(ctx context.Context, payloads []engine.ExecutableData, firstTestBlock uint64, l1Chain *l1Chain) error {
+func (nb *NetworkBenchmark) benchmarkValidator(ctx context.Context, payloads []engine.ExecutableData, lastSetupBlock uint64, l1Chain *l1Chain, sequencerClient types.ExecutionClient) error {
 	validatorClient, err := setupNode(ctx, nb.log, nb.testConfig.Params, nb.validatorOptions, nb.ports)
 	if err != nil {
+		sequencerClient.Stop()
 		return fmt.Errorf("failed to setup validator node: %w", err)
 	}
 
@@ -134,6 +136,52 @@ func (nb *NetworkBenchmark) benchmarkValidator(ctx context.Context, payloads []e
 		validatorClient.Stop()
 	}()
 
+	// check if validator is behind first test block
+	validatorHeader, err := validatorClient.Client().HeaderByNumber(ctx, nil)
+	if err != nil {
+		sequencerClient.Stop()
+		return fmt.Errorf("failed to get validator header: %w", err)
+	}
+
+	nb.log.Info("Validator header", "number", validatorHeader.Number.Uint64(), "lastSetupBlock", lastSetupBlock)
+
+	if validatorHeader.Number.Cmp(big.NewInt(int64(lastSetupBlock))) < 0 {
+		nb.log.Info("Validator is behind first test block, catching up", "validator_block", validatorHeader.Number.Uint64(), "first_test_block", lastSetupBlock)
+		// fetch all blocks the validator node is missing
+		for i := validatorHeader.Number.Uint64() + 1; i <= lastSetupBlock; i++ {
+			block, err := sequencerClient.Client().BlockByNumber(ctx, big.NewInt(int64(i)))
+			if err != nil {
+				sequencerClient.Stop()
+				return fmt.Errorf("failed to get block %d: %w", i, err)
+			}
+
+			log.Info("Sending newpayload to validator node to catch up", "block", block.NumberU64())
+
+			// send newpayload to validator node
+			payload := engine.BlockToExecutableData(block, big.NewInt(0), []*ethTypes.BlobTxSidecar{}, [][]byte{}).ExecutionPayload
+			root := crypto.Keccak256Hash([]byte("fake-beacon-block-root"), big.NewInt(int64(1)).Bytes())
+
+			err = validatorClient.AuthClient().CallContext(ctx, nil, "engine_newPayloadV4", payload, []common.Hash{}, root, []common.Hash{})
+			if err != nil {
+				validatorClient.Stop()
+				return fmt.Errorf("failed to send newpayload to validator node: %w", err)
+			}
+
+			forkchoiceUpdate := engine.ForkchoiceStateV1{
+				HeadBlockHash:      payload.BlockHash,
+				SafeBlockHash:      payload.BlockHash,
+				FinalizedBlockHash: payload.BlockHash,
+			}
+
+			err = validatorClient.AuthClient().CallContext(ctx, nil, "engine_forkchoiceUpdatedV3", forkchoiceUpdate, nil)
+			if err != nil {
+				validatorClient.Stop()
+				return fmt.Errorf("failed to send forkchoice update to validator node: %w", err)
+			}
+		}
+	}
+	sequencerClient.Stop()
+
 	// Create metrics collector and writer
 	metricsCollector := validatorClient.MetricsCollector()
 	metricsWriter := metrics.NewFileMetricsWriter(nb.validatorOptions.MetricsPath)
@@ -150,7 +198,7 @@ func (nb *NetworkBenchmark) benchmarkValidator(ctx context.Context, payloads []e
 	}()
 
 	benchmark := newValidatorBenchmark(nb.log, *nb.testConfig, validatorClient, l1Chain, nb.proofConfig)
-	return benchmark.Run(ctx, payloads, firstTestBlock, metricsCollector)
+	return benchmark.Run(ctx, payloads, lastSetupBlock, metricsCollector)
 }
 
 func (nb *NetworkBenchmark) GetResult() (*benchmark.RunResult, error) {

runner/network/sequencer_benchmark.go

@@ -168,8 +168,6 @@ func (nb *sequencerBenchmark) Run(ctx context.Context, metricsCollector metrics.
 		close(setupComplete)
 	}()
 
-	var lastSetupBlock uint64
-
 	headBlockHeader, err := sequencerClient.Client().HeaderByNumber(ctx, nil)
 	if err != nil {
 		nb.log.Warn("failed to get head block header", "err", err)
@@ -195,13 +193,12 @@ func (nb *sequencerBenchmark) Run(ctx context.Context, metricsCollector metrics.
 	setupLoop:
 		for {
 			_blockMetrics := metrics.NewBlockMetrics()
-			payload, err := consensusClient.Propose(benchmarkCtx, _blockMetrics, true)
+			_, err := consensusClient.Propose(benchmarkCtx, _blockMetrics, true)
 			if err != nil {
 				errChan <- err
 				return
 			}
 
-			payloads = append(payloads, *payload)
 			select {
 			case <-setupComplete:
 				break setupLoop
@@ -212,8 +209,6 @@ func (nb *sequencerBenchmark) Run(ctx context.Context, metricsCollector metrics.
 
 		}
 
-		lastSetupBlock = payloads[len(payloads)-1].Number
-		nb.log.Info("Last setup block", "block", lastSetupBlock)
 		blockMetrics := metrics.NewBlockMetrics()
 
 		// run for a few blocks
@@ -258,6 +253,6 @@ func (nb *sequencerBenchmark) Run(ctx context.Context, metricsCollector metrics.
 	case err := <-errChan:
 		return nil, 0, err
 	case payloads := <-payloadResult:
-		return payloads, lastSetupBlock + 1, nil
+		return payloads, payloads[0].Number - 1, nil
 	}
 }

runner/payload/factory.go

@@ -2,7 +2,7 @@ package payload
 
 import (
 	"context"
-	"errors"
+	"fmt"
 
 	clienttypes "github.com/base/base-bench/runner/clients/types"
 	benchtypes "github.com/base/base-bench/runner/network/types"
@@ -41,7 +41,7 @@ func NewPayloadWorker(ctx context.Context, log log.Logger, testConfig *benchtype
 		worker, err = simulator.NewSimulatorPayloadWorker(
 			ctx, log, sequencerClient.ClientURL(), params, privateKey, amount, &genesis, definition.Params)
 	default:
-		return nil, errors.New("invalid payload type")
+		return nil, fmt.Errorf("invalid payload type: %s", definition.Type)
 	}
 
 	return worker, err