Add origin height to Espresso streamer

.github/workflows/docker-images.yml

@@ -2,17 +2,9 @@ name: Build and Push Docker Images
 
 on:
   push:
-    branches: [main, celo*]
-    paths:
-      - "espresso/docker/**"
-      - "espresso/docker-compose.yml"
-      - "config/**"
+    branches:
+      - "celo-integration*"
   pull_request:
-    branches: [main, celo*, integration]
-    paths:
-      - "espresso/docker/**"
-      - "espresso/docker-compose.yml"
-      - "config/**"
   workflow_dispatch:
 
 env:
@@ -45,7 +37,7 @@ jobs:
       - name: Install Foundry
         uses: foundry-rs/foundry-toolchain@v1
         with:
-          version: nightly-654c8f01721e43dbc8a53c7a3b022548cb82b2f9  # same as for the nix environment
+          version: nightly-654c8f01721e43dbc8a53c7a3b022548cb82b2f9 # same as for the nix environment
 
       - name: Install dasel
         run: |
@@ -465,4 +457,3 @@ jobs:
             TARGET_BASE_IMAGE=alpine:3.22
             TARGETOS=linux
             TARGETARCH=amd64
-

.github/workflows/espresso-devnet-tests.yaml

@@ -1,11 +1,8 @@
 name: Run Espresso Devnet tests
 on:
   pull_request:
-    branches:
-      - "celo-integration*"
   push:
     branches:
-      - "master"
       - "celo-integration*"
   workflow_dispatch:
 
@@ -57,7 +54,6 @@ jobs:
       - name: Run Challenge Game test
         run: go test -timeout 30m -p 1 -count 1 -run 'TestChallengeGame' -v  ./espresso/devnet-tests/...
 
-
       - name: Run Withdraw test
         run: go test -timeout 30m -p 1 -count 1 -run 'TestWithdrawal' -v  ./espresso/devnet-tests/...
 

.github/workflows/espresso-enclave.yaml

@@ -2,8 +2,6 @@ name: Run enclave tests on EC2 instance
 
 on:
   pull_request:
-    branches:
-      - "celo-integration*"
   push:
     branches:
       - "celo-integration*"
@@ -19,7 +17,6 @@ jobs:
     timeout-minutes: 40
 
     steps:
-
       - name: Checkout repository
         uses: actions/checkout@v4
 

.github/workflows/espresso-integration.yaml

@@ -1,11 +1,8 @@
 name: Run Espresso integration tests
 on:
   pull_request:
-    branches:
-      - "celo-integration*"
   push:
     branches:
-      - "master"
       - "celo-integration*"
   workflow_dispatch:
 

espresso/cli.go

@@ -8,7 +8,12 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/urfave/cli/v2"
+
+	espressoClient "github.com/EspressoSystems/espresso-network/sdks/go/client"
+	espressoLightClient "github.com/EspressoSystems/espresso-network/sdks/go/light-client"
 )
 
 // espressoFlags returns the flag names for espresso
@@ -28,6 +33,9 @@ var (
 	LightClientAddrFlagName          = espressoFlags("light-client-addr")
 	L1UrlFlagName                    = espressoFlags("l1-url")
 	TestingBatcherPrivateKeyFlagName = espressoFlags("testing-batcher-private-key")
+	OriginHeight                     = espressoFlags("origin-height")
+	NamespaceFlagName                = espressoFlags("namespace")
+	RollupL1UrlFlagName              = espressoFlags("rollup-l1-url")
 )
 
 func CLIFlags(envPrefix string, category string) []cli.Flag {
@@ -77,6 +85,24 @@ func CLIFlags(envPrefix string, category string) []cli.Flag {
 			EnvVars:  espressoEnvs(envPrefix, "TESTING_BATCHER_PRIVATE_KEY"),
 			Category: category,
 		},
+		&cli.Uint64Flag{
+			Name:     OriginHeight,
+			Usage:    "Espresso transactions below this height will not be considered",
+			EnvVars:  espressoEnvs(envPrefix, "ORIGIN_HEIGHT"),
+			Category: category,
+		},
+		&cli.Uint64Flag{
+			Name:     NamespaceFlagName,
+			Usage:    "Namespace of Espresso transactions",
+			EnvVars:  espressoEnvs(envPrefix, "NAMESPACE"),
+			Category: category,
+		},
+		&cli.StringFlag{
+			Name:     RollupL1UrlFlagName,
+			Usage:    "RPC URL of L1 backing the Rollup we're streaming for",
+			EnvVars:  espressoEnvs(envPrefix, "ROLLUP_L1_URL"),
+			Category: category,
+		},
 	}
 }
 
@@ -87,7 +113,10 @@ type CLIConfig struct {
 	QueryServiceURLs         []string
 	LightClientAddr          common.Address
 	L1URL                    string
+	RollupL1URL              string
 	TestingBatcherPrivateKey *ecdsa.PrivateKey
+	Namespace                uint64
+	OriginHeight             uint64
 }
 
 func (c CLIConfig) Check() error {
@@ -102,6 +131,12 @@ func (c CLIConfig) Check() error {
 		if c.L1URL == "" {
 			return fmt.Errorf("L1 URL is required when Espresso is enabled")
 		}
+		if c.RollupL1URL == "" {
+			return fmt.Errorf("rollup L1 URL is required when Espresso is enabled")
+		}
+		if c.Namespace == 0 {
+			return fmt.Errorf("namespace is required when Espresso is enabled")
+		}
 	}
 	return nil
 }
@@ -112,6 +147,9 @@ func ReadCLIConfig(c *cli.Context) CLIConfig {
 		PollInterval: c.Duration(PollIntervalFlagName),
 		UseFetchAPI:  c.Bool(UseFetchApiFlagName),
 		L1URL:        c.String(L1UrlFlagName),
+		RollupL1URL:  c.String(RollupL1UrlFlagName),
+		Namespace:    c.Uint64(NamespaceFlagName),
+		OriginHeight: c.Uint64(OriginHeight),
 	}
 
 	config.QueryServiceURLs = c.StringSlice(QueryServiceUrlsFlagName)
@@ -128,3 +166,48 @@ func ReadCLIConfig(c *cli.Context) CLIConfig {
 
 	return config
 }
+
+func BatchStreamerFromCLIConfig[B Batch](
+	cfg CLIConfig,
+	log log.Logger,
+	unmarshalBatch func([]byte) (*B, error),
+) (*BatchStreamer[B], error) {
+	if !cfg.Enabled {
+		return nil, fmt.Errorf("Espresso is not enabled")
+	}
+
+	l1Client, err := ethclient.Dial(cfg.L1URL)
+	if err != nil {
+		return nil, fmt.Errorf("failed to dial L1 RPC at %s: %w", cfg.L1URL, err)
+	}
+
+	RollupL1Client, err := ethclient.Dial(cfg.RollupL1URL)
+	if err != nil {
+		return nil, fmt.Errorf("failed to dial Rollup L1 RPC at %s: %w", cfg.RollupL1URL, err)
+	}
+
+	espressoClient, err := espressoClient.NewMultipleNodesClient(cfg.QueryServiceURLs)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create Espresso client: %w", err)
+	}
+
+	espressoLightClient, err := espressoLightClient.NewLightclientCaller(cfg.LightClientAddr, l1Client)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create Espresso light client")
+	}
+
+	streamer := NewEspressoStreamer(
+		cfg.Namespace,
+		NewAdaptL1BlockRefClient(l1Client),
+		NewAdaptL1BlockRefClient(RollupL1Client),
+		espressoClient,
+		espressoLightClient,
+		log,
+		unmarshalBatch,
+		cfg.PollInterval,
+		cfg.OriginHeight,
+	)
+	streamer.UseFetchApi = cfg.UseFetchAPI
+
+	return streamer, nil
+}

espresso/environment/2_espresso_liveness_test.go

@@ -13,7 +13,6 @@ import (
 	espressoLightClient "github.com/EspressoSystems/espresso-network/sdks/go/light-client"
 	"github.com/ethereum-optimism/optimism/espresso"
 	env "github.com/ethereum-optimism/optimism/espresso/environment"
-	"github.com/ethereum-optimism/optimism/op-batcher/batcher"
 	"github.com/ethereum-optimism/optimism/op-e2e/e2eutils/wait"
 	"github.com/ethereum-optimism/optimism/op-e2e/system/e2esys"
 	"github.com/ethereum-optimism/optimism/op-e2e/system/helpers"
@@ -262,14 +261,16 @@ func TestE2eDevnetWithEspressoDegradedLivenessViaCaffNode(t *testing.T) {
 		require.NoError(t, err, "light client creation failed")
 		streamer := espresso.NewEspressoStreamer(
 			system.RollupConfig.L2ChainID.Uint64(),
-			batcher.NewAdaptL1BlockRefClient(l1Client),
+			espresso.NewAdaptL1BlockRefClient(l1Client),
+			espresso.NewAdaptL1BlockRefClient(l1Client),
 			espressoClient.NewClient(server.URL),
 			lightClient,
 			l,
 			func(b []byte) (*derive.EspressoBatch, error) {
 				return derive.UnmarshalEspressoTransaction(b, system.RollupConfig.Genesis.SystemConfig.BatcherAddr)
 			},
 			100*time.Millisecond,
+			0,
 		)
 
 		l1Client, _ := client.NewRPC(streamBlocksCtx, l, system.NodeEndpoint(e2esys.RoleL1).RPC())

espresso/environment/enclave_helpers.go

@@ -96,6 +96,7 @@ func LaunchBatcherInEnclave() E2eDevnetLauncherOption {
 						appendArg(&args, flags.L1EthRpcFlag.Name, l1Rpc)
 						appendArg(&args, txmgr.L1RPCFlagName, l1Rpc)
 						appendArg(&args, espresso.L1UrlFlagName, l1Rpc)
+						appendArg(&args, espresso.RollupL1UrlFlagName, l1Rpc)
 						l2EthRpc := sys.EthInstances[e2esys.RoleSeq].UserRPC().(endpoint.HttpRPC).HttpRPC()
 						appendArg(&args, flags.L2EthRpcFlag.Name, l2EthRpc)
 						rollupRpc := sys.RollupNodes[e2esys.RoleSeq].UserRPC().(endpoint.HttpRPC).HttpRPC()

espresso/environment/espresso_caff_node.go

@@ -118,6 +118,7 @@ func LaunchCaffNode(t *testing.T, system *e2esys.System, espressoDevNode Espress
 		// To create a valid multiple nodes client, we need to provide at least 2 URLs.
 		QueryServiceURLs: []string{u.String(), u.String()},
 		L1URL:            system.L1.UserRPC().RPC(),
+		RollupL1URL:      system.L1.UserRPC().RPC(),
 		LightClientAddr:  common.HexToAddress(ESPRESSO_LIGHT_CLIENT_ADDRESS),
 	}
 

espresso/ethclient.go

@@ -0,0 +1,31 @@
+package espresso
+
+import (
+	"context"
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/ethclient"
+)
+
+// AdaptL1BlockRefClient is a wrapper around eth.L1BlockRef that implements the espresso.L1Client interface
+type AdaptL1BlockRefClient struct {
+	L1Client *ethclient.Client
+}
+
+// NewAdaptL1BlockRefClient creates a new L1BlockRefClient
+func NewAdaptL1BlockRefClient(L1Client *ethclient.Client) *AdaptL1BlockRefClient {
+	return &AdaptL1BlockRefClient{
+		L1Client: L1Client,
+	}
+}
+
+// HeaderHashByNumber implements the espresso.L1Client interface
+func (c *AdaptL1BlockRefClient) HeaderHashByNumber(ctx context.Context, number *big.Int) (common.Hash, error) {
+	expectedL1BlockRef, err := c.L1Client.HeaderByNumber(ctx, number)
+	if err != nil {
+		return common.Hash{}, err
+	}
+
+	return expectedL1BlockRef.Hash(), nil
+}

espresso/streamer.go

@@ -74,6 +74,7 @@ type BatchStreamer[B Batch] struct {
 	Namespace uint64
 
 	L1Client                      L1Client
+	RollupL1Client                L1Client
 	EspressoClient                EspressoClient
 	EspressoLightClient           LightClientCallerInterface
 	Log                           log.Logger
@@ -87,6 +88,8 @@ type BatchStreamer[B Batch] struct {
 	fallbackBatchPos uint64
 	// HotShot position that we can fallback to, guaranteeing not to skip any unsafe batches
 	fallbackHotShotPos uint64
+	// HotShot position we start reading from, exclusive
+	originHotShotPos uint64
 	// Latest finalized block on the L1.
 	FinalizedL1 eth.L1BlockRef
 
@@ -110,14 +113,17 @@ var _ EspressoStreamer[Batch] = (*BatchStreamer[Batch])(nil)
 func NewEspressoStreamer[B Batch](
 	namespace uint64,
 	l1Client L1Client,
+	rollupL1Client L1Client,
 	espressoClient EspressoClient,
 	lightClient LightClientCallerInterface,
 	log log.Logger,
 	unmarshalBatch func([]byte) (*B, error),
 	pollingHotShotPollingInterval time.Duration,
+	originHotShotPos uint64,
 ) *BatchStreamer[B] {
 	return &BatchStreamer[B]{
 		L1Client:                      l1Client,
+		RollupL1Client:                rollupL1Client,
 		EspressoClient:                espressoClient,
 		EspressoLightClient:           lightClient,
 		Log:                           log,
@@ -127,6 +133,9 @@ func NewEspressoStreamer[B Batch](
 		PollingHotShotPollingInterval: pollingHotShotPollingInterval,
 		RemainingBatches:              make(map[common.Hash]B),
 		unmarshalBatch:                unmarshalBatch,
+		originHotShotPos:              originHotShotPos,
+		fallbackHotShotPos:            originHotShotPos,
+		hotShotPos:                    originHotShotPos,
 	}
 }
 
@@ -147,15 +156,18 @@ func (s *BatchStreamer[B]) RefreshSafeL1Origin(safeL1Origin eth.BlockID) error {
 		s.Reset()
 	}
 
-	return err
+	if err != nil {
+		return fmt.Errorf("failed to confirm espresso block height: %w", err)
+	}
+	return nil
 }
 
 // Update streamer state based on L1 and L2 sync status
 func (s *BatchStreamer[B]) Refresh(ctx context.Context, finalizedL1 eth.L1BlockRef, safeBatchNumber uint64, safeL1Origin eth.BlockID) error {
 	s.FinalizedL1 = finalizedL1
 
 	if err := s.RefreshSafeL1Origin(safeL1Origin); err != nil {
-		return err
+		return fmt.Errorf("failed to refresh safe L1 origin: %w", err)
 	}
 
 	// NOTE: be sure to update s.finalizedL1 before checking this condition and returning
@@ -187,7 +199,7 @@ func (s *BatchStreamer[B]) CheckBatch(ctx context.Context, batch B) (BatchValidi
 		return BatchUndecided, 0
 	}
 
-	l1headerHash, err := s.L1Client.HeaderHashByNumber(ctx, new(big.Int).SetUint64(origin.Number))
+	l1headerHash, err := s.RollupL1Client.HeaderHashByNumber(ctx, new(big.Int).SetUint64(origin.Number))
 	if err != nil {
 		// Signal to resync to be able to fetch the L1 header.
 		s.Log.Warn("Failed to fetch the L1 header, pending resync", "error", err)
@@ -263,7 +275,7 @@ func (s *BatchStreamer[B]) Update(ctx context.Context) error {
 	// the current block height available to process.
 	currentBlockHeight, err := s.EspressoClient.FetchLatestBlockHeight(ctx)
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to fetch latest block height: %w", err)
 	}
 
 	// Streaming API implementation
@@ -544,10 +556,10 @@ func (s *BatchStreamer[B]) confirmEspressoBlockHeight(safeL1Origin eth.BlockID)
 	hotshotState, err := s.EspressoLightClient.
 		FinalizedState(&bind.CallOpts{BlockNumber: new(big.Int).SetUint64(safeL1Origin.Number)})
 	if errors.Is(err, bind.ErrNoCode) {
-		s.fallbackHotShotPos = 0
+		s.fallbackHotShotPos = s.originHotShotPos
 		return false, nil
 	} else if err != nil {
-		return false, err
+		return false, fmt.Errorf("failed to get finalized state from light client: %w", err)
 	}
 
 	shouldReset = hotshotState.BlockHeight < s.fallbackHotShotPos