Meyer9/wip 1

.gitignore

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

genesis.json

@@ -1 +1,57 @@
-{"config":{"chainId":13371337,"homesteadBlock":0,"eip150Block":0,"eip155Block":0,"eip158Block":0,"byzantiumBlock":0,"constantinopleBlock":0,"petersburgBlock":0,"istanbulBlock":0,"muirGlacierBlock":0,"berlinBlock":0,"londonBlock":0,"arrowGlacierBlock":0,"grayGlacierBlock":0,"mergeNetsplitBlock":0,"shanghaiTime":0,"cancunTime":0,"pragueTime":0,"bedrockBlock":0,"regolithTime":0,"canyonTime":0,"ecotoneTime":0,"fjordTime":0,"graniteTime":0,"holoceneTime":0,"isthmusTime":0,"terminalTotalDifficulty":1,"depositContractAddress":"0x0000000000000000000000000000000000000000","optimism":{"eip1559Elasticity":1,"eip1559Denominator":50,"eip1559DenominatorCanyon":50}},"nonce":"0x0","timestamp":"0x682255c5","extraData":"0x000000003200000001","gasLimit":"0xffffffffffffffff","difficulty":"0x1","mixHash":"0x0000000000000000000000000000000000000000000000000000000000000000","coinbase":"0x4200000000000000000000000000000000000011","alloc":{"0000f90827f1c53a10cb7a02335b175320002935":{"code":"0x3373fffffffffffffffffffffffffffffffffffffffe14604657602036036042575f35600143038111604257611fff81430311604257611fff9006545f5260205ff35b5f5ffd5b5f35611fff60014303065500","balance":"0x0"}},"number":"0x0","gasUsed":"0x0","parentHash":"0x0000000000000000000000000000000000000000000000000000000000000000","baseFeePerGas":null,"excessBlobGas":null,"blobGasUsed":null}
+{
+    "config": {
+        "chainId": 13371337,
+        "homesteadBlock": 0,
+        "eip150Block": 0,
+        "eip155Block": 0,
+        "eip158Block": 0,
+        "byzantiumBlock": 0,
+        "constantinopleBlock": 0,
+        "petersburgBlock": 0,
+        "istanbulBlock": 0,
+        "muirGlacierBlock": 0,
+        "berlinBlock": 0,
+        "londonBlock": 0,
+        "arrowGlacierBlock": 0,
+        "grayGlacierBlock": 0,
+        "mergeNetsplitBlock": 0,
+        "shanghaiTime": 0,
+        "cancunTime": 0,
+        "pragueTime": 0,
+        "bedrockBlock": 0,
+        "regolithTime": 0,
+        "canyonTime": 0,
+        "ecotoneTime": 0,
+        "fjordTime": 0,
+        "graniteTime": 0,
+        "holoceneTime": 0,
+        "isthmusTime": 0,
+        "jovianTime": 0,
+        "terminalTotalDifficulty": 1,
+        "depositContractAddress": "0x0000000000000000000000000000000000000000",
+        "optimism": {
+            "eip1559Elasticity": 1,
+            "eip1559Denominator": 50,
+            "eip1559DenominatorCanyon": 50
+        }
+    },
+    "nonce": "0x0",
+    "timestamp": "0x682255c5",
+    "extraData": "0x0100000032000000010000000000000000",
+    "gasLimit": "0xFFF",
+    "difficulty": "0x1",
+    "mixHash": "0x0000000000000000000000000000000000000000000000000000000000000000",
+    "coinbase": "0x4200000000000000000000000000000000000011",
+    "alloc": {
+        "0000f90827f1c53a10cb7a02335b175320002935": {
+            "code": "0x3373fffffffffffffffffffffffffffffffffffffffe14604657602036036042575f35600143038111604257611fff81430311604257611fff9006545f5260205ff35b5f5ffd5b5f35611fff60014303065500",
+            "balance": "0x0"
+        }
+    },
+    "number": "0x0",
+    "gasUsed": "0x1",
+    "parentHash": "0x0000000000000000000000000000000000000000000000000000000000000000",
+    "baseFeePerGas": "0x1",
+    "excessBlobGas": null,
+    "blobGasUsed": null
+}

runner/benchmark/definition.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"os/exec"
 	"path"
+	"path/filepath"
 	"strings"
 
 	"github.com/base/base-bench/runner/payload"
@@ -63,7 +64,9 @@ func (s SnapshotDefinition) CreateSnapshot(nodeType string, outputDir string) er
 		return fmt.Errorf("failed to get absolute path of outputDir: %w", err)
 	}
 
-	outputDir = path.Join(currentDir, outputDir)
+	if !filepath.IsAbs(outputDir) {
+		outputDir = path.Join(currentDir, outputDir)
+	}
 
 	var cmdBin string
 	var args []string
@@ -96,6 +99,18 @@ type DatadirConfig struct {
 	Validator *string `yaml:"validator"`
 }
 
+// ReplayConfig specifies configuration for replaying transactions from an
+// external node instead of generating synthetic transactions.
+type ReplayConfig struct {
+	// SourceRPCURL is the RPC endpoint of the node to fetch transactions from
+	SourceRPCURL string `yaml:"source_rpc_url"`
+
+	// StartBlock is the first block to replay transactions from.
+	// If not specified (0), it will be automatically detected from the
+	// snapshot's head block + 1.
+	StartBlock uint64 `yaml:"start_block,omitempty"`
+}
+
 // TestDefinition is the user-facing YAML configuration for specifying a
 // matrix of benchmark runs.
 type TestDefinition struct {
@@ -105,6 +120,7 @@ type TestDefinition struct {
 	Tags         *map[string]string   `yaml:"tags"`
 	Variables    []Param              `yaml:"variables"`
 	ProofProgram *ProofProgramOptions `yaml:"proof_program"`
+	Replay       *ReplayConfig        `yaml:"replay"`
 }
 
 func (bc *TestDefinition) Check() error {

runner/benchmark/matrix.go

@@ -17,6 +17,7 @@ type TestPlan struct {
 	Snapshot     *SnapshotDefinition
 	ProofProgram *ProofProgramOptions
 	Thresholds   *ThresholdConfig
+	Replay       *ReplayConfig
 }
 
 func NewTestPlanFromConfig(c TestDefinition, testFileName string, config *BenchmarkConfig) (*TestPlan, error) {
@@ -42,6 +43,7 @@ func NewTestPlanFromConfig(c TestDefinition, testFileName string, config *Benchm
 		Snapshot:     c.Snapshot,
 		ProofProgram: proofProgram,
 		Thresholds:   c.Metrics,
+		Replay:       c.Replay,
 	}, nil
 }
 

runner/benchmark/portmanager/ports.go

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

runner/clients/rbuilder/client.go

@@ -3,6 +3,7 @@ package rbuilder
 import (
 	"context"
 	"errors"
+	"fmt"
 
 	"github.com/ethereum-optimism/optimism/op-service/client"
 	"github.com/ethereum/go-ethereum/log"
@@ -27,7 +28,8 @@ type RbuilderClient struct {
 
 	// stdout io.WriteCloser
 	// stderr io.WriteCloser
-	// ports    portmanager.PortManager
+	ports         portmanager.PortManager
+	websocketPort uint64
 
 	elClient types.ExecutionClient
 
@@ -43,13 +45,18 @@ func NewRbuilderClient(logger log.Logger, options *config.InternalClientOptions,
 		logger:   logger,
 		options:  options,
 		elClient: rethClient,
+		ports:    ports,
 	}
 }
 
 // Run runs the reth client with the given runtime config.
 func (r *RbuilderClient) Run(ctx context.Context, cfg *types.RuntimeConfig) error {
+	r.websocketPort = r.ports.AcquirePort("rbuilder", portmanager.FlashblocksWebsocketPortPurpose)
+
 	cfg2 := *cfg
 	cfg2.Args = append(cfg2.Args, "--flashblocks.enabled")
+	cfg2.Args = append(cfg2.Args, "--flashblocks.port", fmt.Sprintf("%d", r.websocketPort))
+	cfg2.Args = append(cfg2.Args, "--flashblocks.fixed")
 	err := r.elClient.Run(ctx, &cfg2)
 	if err != nil {
 		return err
@@ -68,6 +75,7 @@ func (r *RbuilderClient) MetricsCollector() metrics.Collector {
 
 // Stop stops the reth client.
 func (r *RbuilderClient) Stop() {
+	r.ports.ReleasePort(r.websocketPort)
 	r.elClient.Stop()
 }
 

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,7 +92,9 @@ 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, "-vvv")
+	args = append(args, "--disable-discovery")
+	args = append(args, "--port", fmt.Sprintf("%d", r.p2pPort))
+	args = append(args, "-vvvv")
 
 	// increase mempool size
 	args = append(args, "--txpool.pending-max-count", "100000000")
@@ -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/configutil/rollup_config.go

@@ -50,7 +50,7 @@ func GetRollupConfig(genesis *core.Genesis, chain fakel1.L1Chain, batcherAddr co
 				}),
 			},
 		},
-		BlockTime:               1,
+		BlockTime:               2,
 		MaxSequencerDrift:       20,
 		SeqWindowSize:           24,
 		L1ChainID:               big.NewInt(1),
@@ -63,6 +63,7 @@ func GetRollupConfig(genesis *core.Genesis, chain fakel1.L1Chain, batcherAddr co
 		GraniteTime:             genesis.Config.GraniteTime,
 		HoloceneTime:            genesis.Config.HoloceneTime,
 		IsthmusTime:             genesis.Config.IsthmusTime,
+		JovianTime:              genesis.Config.JovianTime,
 		InteropTime:             genesis.Config.InteropTime,
 		BatchInboxAddress:       common.Address{1},
 		DepositContractAddress:  common.Address{1},

runner/network/consensus/client.go

@@ -21,6 +21,10 @@ type ConsensusClientOptions struct {
 	GasLimit uint64
 	// GasLimitSetup is the gas limit for the setup payload
 	GasLimitSetup uint64
+	// AllowTxFailures allows transactions to fail without stopping the benchmark.
+	// When true, failed transactions are logged as warnings instead of errors.
+	// Useful for replay mode where some transactions may fail due to state differences.
+	AllowTxFailures bool
 }
 
 // BaseConsensusClient contains common functionality shared between different consensus client implementations.

runner/network/consensus/sequencer_consensus.go

@@ -26,6 +26,7 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
 )
 
 // SequencerConsensusClient is a fake consensus client that generates blocks on a timer.
@@ -117,11 +118,11 @@ func (f *SequencerConsensusClient) generatePayloadAttributes(sequencerTxs [][]by
 	var b8 eth.Bytes8
 	copy(b8[:], eip1559.EncodeHolocene1559Params(50, 1))
 
-	timestamp := f.lastTimestamp + 1
+	timestamp := time.Now().Add(f.options.BlockTime).Unix()
 
 	number := uint64(0)
 	time := uint64(0)
-	baseFee := big.NewInt(1)
+	baseFee := big.NewInt(10)
 	blockHash := common.Hash{}
 	if f.l1Chain != nil {
 		block, err := f.l1Chain.GetBlockByNumber(1)
@@ -133,6 +134,7 @@ func (f *SequencerConsensusClient) generatePayloadAttributes(sequencerTxs [][]by
 		baseFee = block.BaseFee()
 		blockHash = block.Hash()
 	}
+	f.log.Info("Base fee", "baseFee", baseFee)
 
 	l1BlockInfo := &derive.L1BlockInfo{
 		Number:              number,
@@ -184,6 +186,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 +197,7 @@ func (f *SequencerConsensusClient) generatePayloadAttributes(sequencerTxs [][]by
 		ParentBeaconBlockRoot: &root,
 		NoTxPool:              false,
 		EIP1559Params:         &b8,
+		MinBaseFee:            &minBaseFee,
 	}
 
 	return payloadAttrs, &root, nil
@@ -208,29 +212,57 @@ func (f *SequencerConsensusClient) Propose(ctx context.Context, blockMetrics *me
 	sendCallsPerBatch := 100
 	batches := (len(sendTxs) + sendCallsPerBatch - 1) / sendCallsPerBatch
 
-	for i := 0; i < batches; i++ {
-		batch := sendTxs[i*sendCallsPerBatch : min((i+1)*sendCallsPerBatch, len(sendTxs))]
-		results := make([]interface{}, len(batch))
-
-		batchCall := make([]rpc.BatchElem, len(batch))
-		for j, tx := range batch {
-			batchCall[j] = rpc.BatchElem{
-				Method: "eth_sendRawTransaction",
-				Args:   []interface{}{hexutil.Encode(tx)},
-				Result: &results[j],
-			}
+	// Process batches in parallel, 4 at a time
+	parallelBatches := 4
+	failedTxCount := 0
+	for i := 0; i < batches; i += parallelBatches {
+		g, gCtx := errgroup.WithContext(ctx)
+
+		for j := 0; j < parallelBatches && i+j < batches; j++ {
+			batchIdx := i + j
+			g.Go(func() error {
+				batch := sendTxs[batchIdx*sendCallsPerBatch : min((batchIdx+1)*sendCallsPerBatch, len(sendTxs))]
+				results := make([]interface{}, len(batch))
+
+				batchCall := make([]rpc.BatchElem, len(batch))
+				for k, tx := range batch {
+					batchCall[k] = rpc.BatchElem{
+						Method: "eth_sendRawTransaction",
+						Args:   []interface{}{hexutil.Encode(tx)},
+						Result: &results[k],
+					}
+				}
+
+				err := f.client.Client().BatchCallContext(gCtx, batchCall)
+				if err != nil {
+					if f.options.AllowTxFailures {
+						f.log.Warn("Failed to send transaction batch", "error", err)
+						return nil
+					}
+					return errors.Wrap(err, "failed to send transactions")
+				}
+
+				for _, tx := range batchCall {
+					if tx.Error != nil {
+						if f.options.AllowTxFailures {
+							f.log.Warn("Transaction failed", "error", tx.Error)
+							failedTxCount++
+						} else {
+							return errors.Wrapf(tx.Error, "failed to send transaction %#v", tx.Args[0])
+						}
+					}
+				}
+				return nil
+			})
 		}
 
-		err := f.client.Client().BatchCallContext(ctx, batchCall)
-		if err != nil {
-			return nil, errors.Wrap(err, "failed to send transactions")
+		if err := g.Wait(); err != nil {
+			return nil, err
 		}
+	}
 
-		for _, tx := range batchCall {
-			if tx.Error != nil {
-				return nil, errors.Wrapf(tx.Error, "failed to send transaction %#v", tx.Args[0])
-			}
-		}
+	if failedTxCount > 0 {
+		f.log.Warn("Some transactions failed during replay", "failed_count", failedTxCount, "total_count", len(sendTxs))
 	}
 
 	duration := time.Since(startTime)

runner/network/consensus/validator_consensus.go

@@ -33,7 +33,7 @@ func (f *SyncingConsensusClient) propose(ctx context.Context, payload *engine.Ex
 
 	root := crypto.Keccak256Hash([]byte("fake-beacon-block-root"), big.NewInt(1).Bytes())
 
-	f.log.Info("Validate payload", "payload_index", payload.Number)
+	f.log.Info("Validate payload", "payload_index", payload.Number, "num_txs", len(payload.Transactions))
 	startTime := time.Now()
 	err := f.newPayload(ctx, payload, root)
 	if err != nil {
@@ -64,18 +64,18 @@ func (f *SyncingConsensusClient) propose(ctx context.Context, payload *engine.Ex
 }
 
 // Start starts the fake consensus client.
-func (f *SyncingConsensusClient) Start(ctx context.Context, payloads []engine.ExecutableData, metricsCollector metrics.Collector, firstTestBlock uint64) error {
+func (f *SyncingConsensusClient) Start(ctx context.Context, payloads []engine.ExecutableData, metricsCollector metrics.Collector, lastSetupBlock uint64) error {
 	f.log.Info("Starting sync benchmark", "num_payloads", len(payloads))
 	m := metrics.NewBlockMetrics()
 	for i := 0; i < len(payloads); i++ {
-		m.SetBlockNumber(uint64(max(0, int(payloads[i].Number)-int(firstTestBlock))))
+		m.SetBlockNumber(uint64(max(0, int(payloads[i].Number)-int(lastSetupBlock))))
 		f.log.Info("Proposing payload", "payload_index", i)
 		err := f.propose(ctx, &payloads[i], m)
 		if err != nil {
 			return err
 		}
 
-		if payloads[i].Number >= firstTestBlock {
+		if payloads[i].Number > lastSetupBlock {
 			err = metricsCollector.Collect(ctx, m)
 			if err != nil {
 				f.log.Error("Failed to collect metrics", "error", err)