op-node/rollup/derive: Implement pipeline stage multiplexing (#12506)

* op-node/rollup/derive: Implement pipeline stage multiplexing

* fix BatchStage empty batch generation

* fix fork configuration in LargeL1Gaps test

op-chain-ops/genesis/config.go

@@ -369,6 +369,81 @@ func offsetToUpgradeTime(offset *hexutil.Uint64, genesisTime uint64) *uint64 {
 	return &v
 }
 
+func (d *UpgradeScheduleDeployConfig) ForkTimeOffset(fork rollup.ForkName) *uint64 {
+	switch fork {
+	case rollup.Regolith:
+		return (*uint64)(d.L2GenesisRegolithTimeOffset)
+	case rollup.Canyon:
+		return (*uint64)(d.L2GenesisCanyonTimeOffset)
+	case rollup.Delta:
+		return (*uint64)(d.L2GenesisDeltaTimeOffset)
+	case rollup.Ecotone:
+		return (*uint64)(d.L2GenesisEcotoneTimeOffset)
+	case rollup.Fjord:
+		return (*uint64)(d.L2GenesisFjordTimeOffset)
+	case rollup.Granite:
+		return (*uint64)(d.L2GenesisGraniteTimeOffset)
+	case rollup.Holocene:
+		return (*uint64)(d.L2GenesisHoloceneTimeOffset)
+	case rollup.Interop:
+		return (*uint64)(d.L2GenesisInteropTimeOffset)
+	default:
+		panic(fmt.Sprintf("unknown fork: %s", fork))
+	}
+}
+
+func (d *UpgradeScheduleDeployConfig) SetForkTimeOffset(fork rollup.ForkName, offset *uint64) {
+	switch fork {
+	case rollup.Regolith:
+		d.L2GenesisRegolithTimeOffset = (*hexutil.Uint64)(offset)
+	case rollup.Canyon:
+		d.L2GenesisCanyonTimeOffset = (*hexutil.Uint64)(offset)
+	case rollup.Delta:
+		d.L2GenesisDeltaTimeOffset = (*hexutil.Uint64)(offset)
+	case rollup.Ecotone:
+		d.L2GenesisEcotoneTimeOffset = (*hexutil.Uint64)(offset)
+	case rollup.Fjord:
+		d.L2GenesisFjordTimeOffset = (*hexutil.Uint64)(offset)
+	case rollup.Granite:
+		d.L2GenesisGraniteTimeOffset = (*hexutil.Uint64)(offset)
+	case rollup.Holocene:
+		d.L2GenesisHoloceneTimeOffset = (*hexutil.Uint64)(offset)
+	case rollup.Interop:
+		d.L2GenesisInteropTimeOffset = (*hexutil.Uint64)(offset)
+	default:
+		panic(fmt.Sprintf("unknown fork: %s", fork))
+	}
+}
+
+var scheduleableForks = rollup.ForksFrom(rollup.Regolith)
+
+// ActivateForkAtOffset activates the given fork at the given offset. Previous forks are activated
+// at genesis and later forks are deactivated.
+// If multiple forks should be activated at a later time than genesis, first call
+// ActivateForkAtOffset with the earliest fork and then SetForkTimeOffset to individually set later
+// forks.
+func (d *UpgradeScheduleDeployConfig) ActivateForkAtOffset(fork rollup.ForkName, offset uint64) {
+	if !rollup.IsValidFork(fork) || fork == rollup.Bedrock {
+		panic(fmt.Sprintf("invalid fork: %s", fork))
+	}
+	ts := new(uint64)
+	for i, f := range scheduleableForks {
+		if f == fork {
+			d.SetForkTimeOffset(fork, &offset)
+			ts = nil
+		} else {
+			d.SetForkTimeOffset(scheduleableForks[i], ts)
+		}
+	}
+}
+
+// ActivateForkAtGenesis activates the given fork, and all previous forks, at genesis.
+// Later forks are deactivated.
+// See also [ActivateForkAtOffset].
+func (d *UpgradeScheduleDeployConfig) ActivateForkAtGenesis(fork rollup.ForkName) {
+	d.ActivateForkAtOffset(fork, 0)
+}
+
 func (d *UpgradeScheduleDeployConfig) RegolithTime(genesisTime uint64) *uint64 {
 	return offsetToUpgradeTime(d.L2GenesisRegolithTimeOffset, genesisTime)
 }
@@ -402,7 +477,6 @@ func (d *UpgradeScheduleDeployConfig) InteropTime(genesisTime uint64) *uint64 {
 }
 
 func (d *UpgradeScheduleDeployConfig) AllocMode(genesisTime uint64) L2AllocsMode {
-
 	forks := d.forks()
 	for i := len(forks) - 1; i >= 0; i-- {
 		if forkTime := offsetToUpgradeTime(forks[i].L2GenesisTimeOffset, genesisTime); forkTime != nil && *forkTime == 0 {

op-chain-ops/genesis/config_test.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-service/testlog"
 )
 
@@ -45,7 +46,10 @@ func TestRegolithTimeZero(t *testing.T) {
 	config := &DeployConfig{
 		L2InitializationConfig: L2InitializationConfig{
 			UpgradeScheduleDeployConfig: UpgradeScheduleDeployConfig{
-				L2GenesisRegolithTimeOffset: &regolithOffset}}}
+				L2GenesisRegolithTimeOffset: &regolithOffset,
+			},
+		},
+	}
 	require.Equal(t, uint64(0), *config.RegolithTime(1234))
 }
 
@@ -54,7 +58,10 @@ func TestRegolithTimeAsOffset(t *testing.T) {
 	config := &DeployConfig{
 		L2InitializationConfig: L2InitializationConfig{
 			UpgradeScheduleDeployConfig: UpgradeScheduleDeployConfig{
-				L2GenesisRegolithTimeOffset: &regolithOffset}}}
+				L2GenesisRegolithTimeOffset: &regolithOffset,
+			},
+		},
+	}
 	require.Equal(t, uint64(1500+5000), *config.RegolithTime(5000))
 }
 
@@ -63,7 +70,10 @@ func TestCanyonTimeZero(t *testing.T) {
 	config := &DeployConfig{
 		L2InitializationConfig: L2InitializationConfig{
 			UpgradeScheduleDeployConfig: UpgradeScheduleDeployConfig{
-				L2GenesisCanyonTimeOffset: &canyonOffset}}}
+				L2GenesisCanyonTimeOffset: &canyonOffset,
+			},
+		},
+	}
 	require.Equal(t, uint64(0), *config.CanyonTime(1234))
 }
 
@@ -72,7 +82,10 @@ func TestCanyonTimeOffset(t *testing.T) {
 	config := &DeployConfig{
 		L2InitializationConfig: L2InitializationConfig{
 			UpgradeScheduleDeployConfig: UpgradeScheduleDeployConfig{
-				L2GenesisCanyonTimeOffset: &canyonOffset}}}
+				L2GenesisCanyonTimeOffset: &canyonOffset,
+			},
+		},
+	}
 	require.Equal(t, uint64(1234+1500), *config.CanyonTime(1234))
 }
 
@@ -124,3 +137,41 @@ func TestL1Deployments(t *testing.T) {
 	// One that doesn't exist returns empty string
 	require.Equal(t, "", deployments.GetName(common.Address{19: 0xff}))
 }
+
+// This test guarantees that getters and setters for all forks are present.
+func TestUpgradeScheduleDeployConfig_ForkGettersAndSetters(t *testing.T) {
+	var d UpgradeScheduleDeployConfig
+	for i, fork := range rollup.ForksFrom(rollup.Regolith) {
+		require.Nil(t, d.ForkTimeOffset(fork))
+		offset := uint64(i * 42)
+		d.SetForkTimeOffset(fork, &offset)
+		require.Equal(t, offset, *d.ForkTimeOffset(fork))
+	}
+}
+
+func TestUpgradeScheduleDeployConfig_ActivateForkAtOffset(t *testing.T) {
+	var d UpgradeScheduleDeployConfig
+	ts := uint64(42)
+	t.Run("invalid", func(t *testing.T) {
+		require.Panics(t, func() { d.ActivateForkAtOffset(rollup.Bedrock, ts) })
+	})
+
+	t.Run("regolith", func(t *testing.T) {
+		d.ActivateForkAtOffset(rollup.Regolith, ts)
+		require.EqualValues(t, &ts, d.L2GenesisRegolithTimeOffset)
+		for _, fork := range scheduleableForks[1:] {
+			require.Nil(t, d.ForkTimeOffset(fork))
+		}
+	})
+
+	t.Run("ecotone", func(t *testing.T) {
+		d.ActivateForkAtOffset(rollup.Ecotone, ts)
+		require.EqualValues(t, &ts, d.L2GenesisEcotoneTimeOffset)
+		for _, fork := range scheduleableForks[:3] {
+			require.Zero(t, *d.ForkTimeOffset(fork))
+		}
+		for _, fork := range scheduleableForks[4:] {
+			require.Nil(t, d.ForkTimeOffset(fork))
+		}
+	})
+}

op-e2e/actions/derivation/blocktime_test.go

@@ -7,6 +7,7 @@ import (
 	actionsHelpers "github.com/ethereum-optimism/optimism/op-e2e/actions/helpers"
 	upgradesHelpers "github.com/ethereum-optimism/optimism/op-e2e/actions/upgrades/helpers"
 	"github.com/ethereum-optimism/optimism/op-e2e/e2eutils"
+	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-service/testlog"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -163,11 +164,13 @@ func LargeL1Gaps(gt *testing.T, deltaTimeOffset *hexutil.Uint64) {
 	dp.DeployConfig.L2BlockTime = 2
 	dp.DeployConfig.SequencerWindowSize = 4
 	dp.DeployConfig.MaxSequencerDrift = 32
-	dp.DeployConfig.L2GenesisEcotoneTimeOffset = nil
-	dp.DeployConfig.L2GenesisFjordTimeOffset = nil
+	if deltaTimeOffset != nil {
+		dp.DeployConfig.ActivateForkAtOffset(rollup.Delta, uint64(*deltaTimeOffset))
+	} else {
+		dp.DeployConfig.ActivateForkAtGenesis(rollup.Canyon)
+	}
 	// TODO(client-pod#831): The Ecotone (and Fjord) activation blocks don't include user txs,
 	// so disabling these forks for now.
-	upgradesHelpers.ApplyDeltaTimeOffset(dp, deltaTimeOffset)
 	sd := e2eutils.Setup(t, dp, actionsHelpers.DefaultAlloc)
 	log := testlog.Logger(t, log.LevelDebug)
 

op-e2e/actions/helpers/env.go

@@ -0,0 +1,83 @@
+package helpers
+
+import (
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+
+	"github.com/ethereum-optimism/optimism/op-chain-ops/genesis"
+	"github.com/ethereum-optimism/optimism/op-e2e/e2eutils"
+	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
+	"github.com/ethereum-optimism/optimism/op-service/testlog"
+)
+
+type Env struct {
+	Log  log.Logger
+	Logs *testlog.CapturingHandler
+
+	SetupData *e2eutils.SetupData
+
+	Miner       *L1Miner
+	Seq         *L2Sequencer
+	SeqEngine   *L2Engine
+	Verifier    *L2Verifier
+	VerifEngine *L2Engine
+	Batcher     *L2Batcher
+}
+
+type EnvOpt struct {
+	DeployConfigMod func(*genesis.DeployConfig)
+}
+
+func WithActiveFork(fork rollup.ForkName, offset uint64) EnvOpt {
+	return EnvOpt{
+		DeployConfigMod: func(d *genesis.DeployConfig) {
+			d.ActivateForkAtOffset(fork, offset)
+		},
+	}
+}
+
+func WithActiveGenesisFork(fork rollup.ForkName) EnvOpt {
+	return WithActiveFork(fork, 0)
+}
+
+// DefaultFork specifies the default fork to use when setting up the action test environment.
+// Currently manually set to Holocene.
+// Replace with `var DefaultFork = func() rollup.ForkName { return rollup.AllForks[len(rollup.AllForks)-1] }()` after Interop launch.
+const DefaultFork = rollup.Holocene
+
+// SetupEnv sets up a default action test environment. If no fork is specified, the default fork as
+// specified by the package variable [defaultFork] is used.
+func SetupEnv(t StatefulTesting, opts ...EnvOpt) (env Env) {
+	dp := e2eutils.MakeDeployParams(t, DefaultRollupTestParams())
+
+	log, logs := testlog.CaptureLogger(t, log.LevelDebug)
+	env.Log, env.Logs = log, logs
+
+	dp.DeployConfig.ActivateForkAtGenesis(DefaultFork)
+	for _, opt := range opts {
+		if dcMod := opt.DeployConfigMod; dcMod != nil {
+			dcMod(dp.DeployConfig)
+		}
+	}
+
+	sd := e2eutils.Setup(t, dp, DefaultAlloc)
+	env.SetupData = sd
+	env.Miner, env.SeqEngine, env.Seq = SetupSequencerTest(t, sd, log)
+	env.Miner.ActL1SetFeeRecipient(common.Address{'A'})
+	env.VerifEngine, env.Verifier = SetupVerifier(t, sd, log, env.Miner.L1Client(t, sd.RollupCfg), env.Miner.BlobStore(), &sync.Config{})
+	rollupSeqCl := env.Seq.RollupClient()
+	env.Batcher = NewL2Batcher(log, sd.RollupCfg, DefaultBatcherCfg(dp),
+		rollupSeqCl, env.Miner.EthClient(), env.SeqEngine.EthClient(), env.SeqEngine.EngineClient(t, sd.RollupCfg))
+
+	return
+}
+
+func (env Env) ActBatchSubmitAllAndMine(t Testing) (l1InclusionBlock *types.Block) {
+	env.Batcher.ActSubmitAll(t)
+	batchTx := env.Batcher.LastSubmitted
+	env.Miner.ActL1StartBlock(12)(t)
+	env.Miner.ActL1IncludeTxByHash(batchTx.Hash())(t)
+	return env.Miner.ActL1EndBlock(t)
+}

op-e2e/actions/helpers/l1_miner.go

@@ -203,10 +203,10 @@ func (s *L1Miner) ActL1SetFeeRecipient(coinbase common.Address) {
 }
 
 // ActL1EndBlock finishes the new L1 block, and applies it to the chain as unsafe block
-func (s *L1Miner) ActL1EndBlock(t Testing) {
+func (s *L1Miner) ActL1EndBlock(t Testing) *types.Block {
 	if !s.l1Building {
 		t.InvalidAction("cannot end L1 block when not building block")
-		return
+		return nil
 	}
 
 	s.l1Building = false
@@ -253,11 +253,12 @@ func (s *L1Miner) ActL1EndBlock(t Testing) {
 	if err != nil {
 		t.Fatalf("failed to insert block into l1 chain")
 	}
+	return block
 }
 
-func (s *L1Miner) ActEmptyBlock(t Testing) {
+func (s *L1Miner) ActEmptyBlock(t Testing) *types.Block {
 	s.ActL1StartBlock(12)(t)
-	s.ActL1EndBlock(t)
+	return s.ActL1EndBlock(t)
 }
 
 func (s *L1Miner) Close() error {