Recovery from fallback batcher

espresso/cli.go

@@ -33,7 +33,8 @@ var (
 	LightClientAddrFlagName          = espressoFlags("light-client-addr")
 	L1UrlFlagName                    = espressoFlags("l1-url")
 	TestingBatcherPrivateKeyFlagName = espressoFlags("testing-batcher-private-key")
-	OriginHeight                     = espressoFlags("origin-height")
+	CaffeinationHeightEspresso       = espressoFlags("origin-height-espresso")
+	CaffeinationHeightL2             = espressoFlags("origin-height-l2")
 	NamespaceFlagName                = espressoFlags("namespace")
 	RollupL1UrlFlagName              = espressoFlags("rollup-l1-url")
 	AttestationServiceFlagName       = espressoFlags("espresso-attestation-service")
@@ -87,9 +88,16 @@ func CLIFlags(envPrefix string, category string) []cli.Flag {
 			Category: category,
 		},
 		&cli.Uint64Flag{
-			Name:     OriginHeight,
+			Name:     CaffeinationHeightEspresso,
 			Usage:    "Espresso transactions below this height will not be considered",
-			EnvVars:  espressoEnvs(envPrefix, "ORIGIN_HEIGHT"),
+			EnvVars:  espressoEnvs(envPrefix, "CAFFEINATION_HEIGHT_ESPRESSO"),
+			Category: category,
+		},
+		&cli.Uint64Flag{
+			Name:     CaffeinationHeightL2,
+			Usage:    "L2 height at which derivation pipeline of Caff node switches to Espresso",
+			Value:    0,
+			EnvVars:  espressoEnvs(envPrefix, "CAFFEINATION_HEIGHT_L2"),
 			Category: category,
 		},
 		&cli.Uint64Flag{
@@ -123,7 +131,8 @@ type CLIConfig struct {
 	RollupL1URL                string
 	TestingBatcherPrivateKey   *ecdsa.PrivateKey
 	Namespace                  uint64
-	OriginHeight               uint64
+	CaffeinationHeightEspresso uint64
+	CaffeinationHeightL2       uint64
 	EspressoAttestationService string
 }
 
@@ -160,7 +169,8 @@ func ReadCLIConfig(c *cli.Context) CLIConfig {
 		L1URL:                      c.String(L1UrlFlagName),
 		RollupL1URL:                c.String(RollupL1UrlFlagName),
 		Namespace:                  c.Uint64(NamespaceFlagName),
-		OriginHeight:               c.Uint64(OriginHeight),
+		CaffeinationHeightEspresso: c.Uint64(CaffeinationHeightEspresso),
+		CaffeinationHeightL2:       c.Uint64(CaffeinationHeightL2),
 		EspressoAttestationService: c.String(AttestationServiceFlagName),
 	}
 
@@ -215,7 +225,8 @@ func BatchStreamerFromCLIConfig[B Batch](
 		log,
 		unmarshalBatch,
 		cfg.PollInterval,
-		cfg.OriginHeight,
+		cfg.CaffeinationHeightEspresso,
+		cfg.CaffeinationHeightL2,
 	)
 	streamer.UseFetchApi = cfg.UseFetchAPI
 

espresso/environment/14_batcher_fallback_test.go

@@ -2,12 +2,17 @@ package environment_test
 
 import (
 	"context"
+	"math/big"
 	"testing"
+	"time"
 
+	espressoClient "github.com/EspressoSystems/espresso-network/sdks/go/client"
 	env "github.com/ethereum-optimism/optimism/espresso/environment"
+	"github.com/ethereum-optimism/optimism/op-batcher/batcher"
 	"github.com/ethereum-optimism/optimism/op-batcher/bindings"
 	"github.com/ethereum-optimism/optimism/op-e2e/e2eutils/wait"
 	"github.com/ethereum-optimism/optimism/op-e2e/system/e2esys"
+	"github.com/ethereum-optimism/optimism/op-node/config"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/stretchr/testify/require"
 )
@@ -18,10 +23,21 @@ func TestBatcherSwitching(t *testing.T) {
 
 	launcher := new(env.EspressoDevNodeLauncherDocker)
 
-	system, espressoDevNode, err := launcher.StartE2eDevnet(ctx, t)
+	// We will need this config to start a new instance of "TEE" batcher
+	// with parameters tweaked.
+	batcherConfig := &batcher.CLIConfig{}
+	system, espressoDevNode, err := launcher.StartE2eDevnet(ctx, t, env.WithSequencerUseFinalized(true), env.GetBatcherConfig(batcherConfig))
 	require.NoError(t, err)
 
 	l1Client := system.NodeClient(e2esys.RoleL1)
+	verifClient := system.NodeClient(e2esys.RoleVerif)
+	espClient := espressoClient.NewClient(espressoDevNode.EspressoUrls()[0])
+
+	deployerTransactor, err := bind.NewKeyedTransactorWithChainID(system.Config().Secrets.Deployer, system.Cfg.L1ChainIDBig())
+	require.NoError(t, err)
+
+	batchAuthenticator, err := bindings.NewBatchAuthenticator(system.RollupConfig.BatchAuthenticatorAddress, l1Client)
+	require.NoError(t, err)
 
 	defer env.Stop(t, system)
 	defer env.Stop(t, espressoDevNode)
@@ -37,21 +53,83 @@ func TestBatcherSwitching(t *testing.T) {
 	require.NoError(t, err)
 
 	// Switch active batcher
-	options, err := bind.NewKeyedTransactorWithChainID(system.Config().Secrets.Deployer, system.Cfg.L1ChainIDBig())
+	tx, err := batchAuthenticator.SwitchBatcher(deployerTransactor)
+	require.NoError(t, err)
+	_, err = wait.ForReceiptOK(ctx, l1Client, tx.Hash())
 	require.NoError(t, err)
 
-	batchAuthenticator, err := bindings.NewBatchAuthenticator(system.RollupConfig.BatchAuthenticatorAddress, l1Client)
+	// Start the fallback batcher
+	err = system.FallbackBatchSubmitter.TestDriver().StartBatchSubmitting()
 	require.NoError(t, err)
 
-	tx, err := batchAuthenticator.SwitchBatcher(options)
+	// Everything should still work
+	env.RunSimpleL2Burn(ctx, t, system)
+
+	// Stop the fallback batcher
+	err = system.FallbackBatchSubmitter.TestDriver().StopBatchSubmitting(ctx)
 	require.NoError(t, err)
-	_, err = wait.ForReceiptOK(ctx, l1Client, tx.Hash())
+
+	// Switch batcher back to the "TEE" batcher
+	tx, err = batchAuthenticator.SwitchBatcher(deployerTransactor)
+	require.NoError(t, err)
+	switchReceipt, err := wait.ForReceiptOK(ctx, l1Client, tx.Hash())
 	require.NoError(t, err)
 
-	// Start the fallback batcher
-	err = system.FallbackBatchSubmitter.TestDriver().StartBatchSubmitting()
+	// Give things time to settle
+	var l2Height uint64
+
+	ticker := time.NewTicker(100 * time.Millisecond)
+	timeoutCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
+	defer cancel()
+
+Loop:
+	for {
+		select {
+		case <-timeoutCtx.Done():
+			panic("Timeout waiting for verifier derivation pipeline to advance past the fallback batcher switchoff point")
+		case <-ticker.C:
+			status, err := system.RollupClient(e2esys.RoleVerif).SyncStatus(ctx)
+			require.NoError(t, err)
+			if status.CurrentL1.Number > switchReceipt.BlockNumber.Uint64() {
+				l2Height = status.LocalSafeL2.Number
+				break Loop
+			}
+		}
+	}
+
+	espHeight, err := espClient.FetchLatestBlockHeight(ctx)
+	require.NoError(t, err)
+
+	// Start a new "TEE" batcher
+	batcherConfig.Espresso.CaffeinationHeightEspresso = espHeight
+	batcherConfig.Espresso.CaffeinationHeightL2 = l2Height
+	newBatcher, err := batcher.BatcherServiceFromCLIConfig(ctx, "0.0.1", batcherConfig, system.BatchSubmitter.Log)
+	require.NoError(t, err)
+	err = newBatcher.Start(ctx)
 	require.NoError(t, err)
 
 	// Everything should still work
 	env.RunSimpleL2Burn(ctx, t, system)
+
+	caffNode, err := env.LaunchCaffNode(t, system, espressoDevNode, func(c *config.Config) {
+		c.Rollup.CaffNodeConfig.CaffeinationHeightEspresso = espHeight
+		c.Rollup.CaffNodeConfig.CaffeinationHeightL2 = l2Height
+	})
+	require.NoError(t, err)
+	defer env.Stop(t, caffNode)
+
+	caffClient := system.NodeClient(env.RoleCaffNode)
+
+	verifHeight, err := verifClient.BlockNumber(ctx)
+	require.NoError(t, err)
+	verifBlock, err := verifClient.BlockByNumber(ctx, new(big.Int).SetUint64(verifHeight))
+	require.NoError(t, err)
+
+	err = wait.ForBlock(ctx, caffClient, verifHeight)
+	require.NoError(t, err)
+
+	caffBlock, err := caffClient.BlockByNumber(ctx, new(big.Int).SetUint64(verifHeight))
+	require.NoError(t, err)
+
+	require.Equal(t, verifBlock.Hash(), caffBlock.Hash())
 }

espresso/environment/2_espresso_liveness_test.go

@@ -271,6 +271,7 @@ func TestE2eDevnetWithEspressoDegradedLivenessViaCaffNode(t *testing.T) {
 			},
 			100*time.Millisecond,
 			0,
+			1,
 		)
 
 		l1Client, _ := client.NewRPC(streamBlocksCtx, l, system.NodeEndpoint(e2esys.RoleL1).RPC())

espresso/environment/espresso_caff_node.go

@@ -15,6 +15,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-e2e/e2eutils/opnode"
 	"github.com/ethereum-optimism/optimism/op-e2e/system/e2esys"
 	"github.com/ethereum-optimism/optimism/op-node/chaincfg"
+	"github.com/ethereum-optimism/optimism/op-node/config"
 	"github.com/ethereum-optimism/optimism/op-service/testlog"
 	"github.com/ethereum/go-ethereum/common"
 )
@@ -73,14 +74,16 @@ type CaffNodeInstance struct {
 	Geth   *geth.GethInstance
 }
 
+type ConfigOption func(*config.Config)
+
 // Close closes the caff node geth instance and op node instance.
 func (c *CaffNodeInstance) Close(ctx context.Context) error {
 	return errors.Join(c.OpNode.Stop(ctx), c.Geth.Close())
 }
 
 // LaunchCaffNode launches a caff node in the given system. It will
 // configure the caff node to connect to the given espresso dev node.
-func LaunchCaffNode(t *testing.T, system *e2esys.System, espressoDevNode EspressoDevNode) (*CaffNodeInstance, error) {
+func LaunchCaffNode(t *testing.T, system *e2esys.System, espressoDevNode EspressoDevNode, opts ...ConfigOption) (*CaffNodeInstance, error) {
 	sequencerHostAndPort := espressoDevNode.SequencerPort()
 	_, sequencerPort, err := net.SplitHostPort(sequencerHostAndPort)
 	if have, want := err, error(nil); have != want {
@@ -122,6 +125,10 @@ func LaunchCaffNode(t *testing.T, system *e2esys.System, espressoDevNode Espress
 		LightClientAddr:  common.HexToAddress(ESPRESSO_LIGHT_CLIENT_ADDRESS),
 	}
 
+	for _, opt := range opts {
+		opt(&caffNodeConfig)
+	}
+
 	// Configure
 	e2esys.ConfigureL1(&caffNodeConfig, system.EthInstances[e2esys.RoleL1], system.L1BeaconEndpoint())
 	e2esys.ConfigureL2(&caffNodeConfig, caffNodeGeth, system.Cfg.JWTSecret)

espresso/environment/optitmism_espresso_test_helpers.go

@@ -568,6 +568,26 @@ func SetBatcherKey(privateKey ecdsa.PrivateKey) E2eDevnetLauncherOption {
 	}
 }
 
+// *c will be set to batcher config. Any devnet launcher options that modify the batcher config
+// should be called before this one.
+func GetBatcherConfig(c *batcher.CLIConfig) E2eDevnetLauncherOption {
+	return func(ct *E2eDevnetLauncherContext) E2eSystemOption {
+		return E2eSystemOption{
+			StartOptions: []e2esys.StartOption{
+				{
+					Role: "get-batcher-config",
+					BatcherMod: func(cfg *batcher.CLIConfig, sys *e2esys.System) {
+						cfg.TargetNumFrames = 10
+						cfg.MaxL1TxSize = 250
+						cfg.MaxChannelDuration = 1000
+						*c = *cfg
+					},
+				},
+			},
+		}
+	}
+}
+
 // SetEspressoUrls allows to set the list of urls for the Espresso client in such a way that N of them are "good" and M of them are "bad".
 // Good urls are the urls defined by this test framework repeated M times. The bad url is provided to the function
 // This function is introduced for testing purposes. It allows to check the enforcement of the majority rule (Test 12)

espresso/streamer.go

@@ -81,10 +81,10 @@ type BatchStreamer[B Batch] struct {
 
 	// Batch number we're to give out next
 	BatchPos uint64
-	// HotShot block that was visited last
-	hotShotPos uint64
 	// Position of the last safe batch, we can use it as the position to fallback when resetting
 	fallbackBatchPos uint64
+	// HotShot block that was visited last
+	hotShotPos uint64
 	// HotShot position that we can fallback to, guaranteeing not to skip any unsafe batches
 	fallbackHotShotPos uint64
 	// HotShot position we start reading from, exclusive
@@ -119,15 +119,18 @@ func NewEspressoStreamer[B Batch](
 	unmarshalBatch func([]byte) (*B, error),
 	pollingHotShotPollingInterval time.Duration,
 	originHotShotPos uint64,
+	originBatchPos uint64,
 ) *BatchStreamer[B] {
 	return &BatchStreamer[B]{
-		L1Client:                      l1Client,
-		RollupL1Client:                rollupL1Client,
-		EspressoClient:                espressoClient,
-		EspressoLightClient:           lightClient,
-		Log:                           log,
-		Namespace:                     namespace,
-		BatchPos:                      1,
+		L1Client:            l1Client,
+		RollupL1Client:      rollupL1Client,
+		EspressoClient:      espressoClient,
+		EspressoLightClient: lightClient,
+		Log:                 log,
+		Namespace:           namespace,
+		// Internally, BatchPos is the position of the batch we are to give out next, hence the +1
+		BatchPos:                      originBatchPos + 1,
+		fallbackBatchPos:              originBatchPos + 1,
 		BatchBuffer:                   NewBatchBuffer[B](),
 		PollingHotShotPollingInterval: pollingHotShotPollingInterval,
 		RemainingBatches:              make(map[common.Hash]B),

espresso/streamer_test.go

@@ -39,6 +39,7 @@ func TestNewEspressoStreamer(t *testing.T) {
 		nil, nil, nil, derive.CreateEspressoBatchUnmarshaler(common.Address{}),
 		50*time.Millisecond,
 		0,
+		1,
 	)
 }
 
@@ -362,6 +363,7 @@ func setupStreamerTesting(namespace uint64, batcherAddress common.Address) (*Moc
 		derive.CreateEspressoBatchUnmarshaler(batcherAddress),
 		50*time.Millisecond,
 		0,
+		1,
 	)
 
 	return state, streamer

op-node/config/config.go

@@ -87,8 +87,6 @@ type Config struct {
 
 	// Experimental. Enables new opstack RPC namespace. Used by op-test-sequencer.
 	ExperimentalOPStackAPI bool
-	// Caff Node config
-	CaffNodeConfig CaffNodeConfig
 }
 
 // CaffNodeConfig is the config for the Caff Node

op-node/rollup/derive/attributes_queue.go

@@ -60,8 +60,9 @@ type AttributesQueue struct {
 	concluding  bool
 	lastAttribs *AttributesWithParent
 
-	isCaffNode       bool
-	espressoStreamer *espresso.BatchStreamer[EspressoBatch]
+	isCaffNode           bool
+	caffeinationHeightL2 uint64
+	espressoStreamer     *espresso.BatchStreamer[EspressoBatch]
 }
 
 type SingularBatchProvider interface {
@@ -96,12 +97,13 @@ func initEspressoStreamer(log log.Logger, cfg *rollup.Config) *espresso.BatchStr
 
 func NewAttributesQueue(log log.Logger, cfg *rollup.Config, builder AttributesBuilder, prev SingularBatchProvider) *AttributesQueue {
 	return &AttributesQueue{
-		log:              log,
-		config:           cfg,
-		builder:          builder,
-		prev:             prev,
-		isCaffNode:       cfg.CaffNodeConfig.Enabled,
-		espressoStreamer: initEspressoStreamer(log, cfg),
+		log:                  log,
+		config:               cfg,
+		builder:              builder,
+		prev:                 prev,
+		isCaffNode:           cfg.CaffNodeConfig.Enabled,
+		caffeinationHeightL2: cfg.CaffNodeConfig.CaffeinationHeightL2,
+		espressoStreamer:     initEspressoStreamer(log, cfg),
 	}
 }
 
@@ -175,7 +177,7 @@ func (aq *AttributesQueue) NextAttributes(ctx context.Context, parent eth.L2Bloc
 		var batch *SingularBatch
 		var concluding bool
 		var err error
-		if aq.isCaffNode {
+		if aq.isCaffNode && parent.Number >= aq.caffeinationHeightL2 {
 			if aq.espressoStreamer == nil {
 				aq.log.Error("Espresso streamer not initialized as expected when isCaffNode is ON")
 				return nil, ErrCritical