op-batcher: bound every Espresso network call and drop the streamer on stop

op-batcher/batcher/driver.go

@@ -319,9 +319,7 @@ func (l *BatchSubmitter) StopBatchSubmitting(ctx context.Context) error {
 	l.wg.Wait()
 	l.cancelKillCtx()
 
-	if l.espressoStreamer != nil {
-		l.espressoStreamer.Stop()
-	}
+	l.teardownEspressoStreamer()
 
 	l.Log.Info("Batch Submitter stopped")
 	return nil

op-batcher/batcher/espresso.go

@@ -120,7 +120,7 @@ type espressoTransactionSubmitter struct {
 	verifyReceiptJobQueue      chan espressoVerifyReceiptJob
 	verifyReceiptRespQueue     chan espressoVerifyReceiptJobResponse
 	verifyReceiptWorkerQueue   chan chan espressoVerifyReceiptJobAttempt
-	espresso                   espressoClient.EspressoClient
+	espresso                   EspressoSubmitClient
 	latestBlockHeight          atomic.Uint64 // shared HotShot block height, updated by trackBlockHeight
 	verifyReceiptMaxBlocks     uint64
 	verifyReceiptSafetyTimeout time.Duration
@@ -134,7 +134,7 @@ type espressoTransactionSubmitter struct {
 // creating the EspressoTransactionSubmitter.
 type EspressoTransactionSubmitterConfig struct {
 	Ctx                                context.Context
-	EspressoClient                     espressoClient.EspressoClient
+	EspressoClient                     EspressoSubmitClient
 	Wg                                 *sync.WaitGroup
 	SubmitJobQueueCapacity             int
 	SubmitResponseQueueCapacity        int
@@ -160,7 +160,7 @@ func WithContext(ctx context.Context) EspressoTransactionSubmitterOption {
 
 // WithEspressoClient is an option that can be used to set the Espresso client
 // for the EspressoTransactionSubmitterConfig.
-func WithEspressoClient(client espressoClient.EspressoClient) EspressoTransactionSubmitterOption {
+func WithEspressoClient(client EspressoSubmitClient) EspressoTransactionSubmitterOption {
 	return func(config *EspressoTransactionSubmitterConfig) {
 		config.EspressoClient = client
 	}
@@ -631,7 +631,7 @@ func (s *espressoTransactionSubmitter) scheduleVerifyReceiptsJobs() {
 func espressoSubmitTransactionWorker(
 	ctx context.Context,
 	wg *sync.WaitGroup,
-	cli espressoClient.EspressoClient,
+	cli EspressoSubmitClient,
 	workerQueue chan<- chan espressoTransactionJobAttempt,
 ) {
 	ctx, cancel := context.WithCancel(ctx)
@@ -688,7 +688,7 @@ func espressoSubmitTransactionWorker(
 func espressoVerifyTransactionWorker(
 	ctx context.Context,
 	wg *sync.WaitGroup,
-	cli espressoClient.EspressoClient,
+	cli EspressoSubmitClient,
 	workerQueue chan<- chan espressoVerifyReceiptJobAttempt,
 	latestHeight *atomic.Uint64,
 	retryDelay time.Duration,
@@ -915,7 +915,15 @@ func (l *BatchSubmitter) espressoBatchLoadingLoop(ctx context.Context, wg *sync.
 					break
 				}
 
-				batch := l.espressoStreamer.Peek(ctx)
+				// Peek retries undecided batches' validity checks, which read L1
+				// (a contract call and a header fetch, both uncached on a miss)
+				// with no deadline of their own. Unbounded, a hung L1 RPC would
+				// silently stall frame publication for good, so it gets the same
+				// per-call bound as every other raw RPC; on expiry the batch
+				// stays undecided and is retried next tick.
+				peekCtx, peekCancel := l.networkTimeoutCtx(ctx)
+				batch := l.espressoStreamer.Peek(peekCtx)
+				peekCancel()
 				if batch == nil {
 					break
 				}

op-batcher/batcher/espresso_client.go

@@ -0,0 +1,51 @@
+package batcher
+
+import (
+	"context"
+	"time"
+
+	espressoCommon "github.com/EspressoSystems/espresso-network/sdks/go/types"
+)
+
+// EspressoSubmitClient is the slice of the Espresso SDK client consumed by the
+// transaction submitter and its workers. Narrowed from the SDK's full
+// EspressoClient so the per-call deadline wrapper below only has to cover
+// methods that are actually called (the full interface includes streaming
+// endpoints, which a per-call deadline would break).
+type EspressoSubmitClient interface {
+	SubmitTransaction(ctx context.Context, tx espressoCommon.Transaction) (*espressoCommon.TaggedBase64, error)
+	FetchTransactionByHash(ctx context.Context, hash *espressoCommon.TaggedBase64) (espressoCommon.TransactionQueryData, error)
+	FetchLatestBlockHeight(ctx context.Context) (uint64, error)
+}
+
+// boundedEspressoClient enforces a per-call deadline on every Espresso SDK
+// call. The SDK issues plain HTTP requests with no client-side timeout, and the
+// submit/verify workers otherwise call it with their long-lived loop contexts —
+// a black-holed connection would hold a worker forever, and with every worker
+// wedged, submission stays stopped even after the endpoint recovers.
+type boundedEspressoClient struct {
+	inner   EspressoSubmitClient
+	timeout time.Duration
+}
+
+func newBoundedEspressoClient(inner EspressoSubmitClient, timeout time.Duration) *boundedEspressoClient {
+	return &boundedEspressoClient{inner: inner, timeout: timeout}
+}
+
+func (c *boundedEspressoClient) SubmitTransaction(ctx context.Context, tx espressoCommon.Transaction) (*espressoCommon.TaggedBase64, error) {
+	ctx, cancel := context.WithTimeout(ctx, c.timeout)
+	defer cancel()
+	return c.inner.SubmitTransaction(ctx, tx)
+}
+
+func (c *boundedEspressoClient) FetchTransactionByHash(ctx context.Context, hash *espressoCommon.TaggedBase64) (espressoCommon.TransactionQueryData, error) {
+	ctx, cancel := context.WithTimeout(ctx, c.timeout)
+	defer cancel()
+	return c.inner.FetchTransactionByHash(ctx, hash)
+}
+
+func (c *boundedEspressoClient) FetchLatestBlockHeight(ctx context.Context) (uint64, error) {
+	ctx, cancel := context.WithTimeout(ctx, c.timeout)
+	defer cancel()
+	return c.inner.FetchLatestBlockHeight(ctx)
+}

op-batcher/batcher/espresso_client_test.go

@@ -0,0 +1,45 @@
+package batcher
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	espressoCommon "github.com/EspressoSystems/espresso-network/sdks/go/types"
+	"github.com/stretchr/testify/require"
+)
+
+// blockingEspressoClient hangs every call until its context is cancelled,
+// modeling the SDK's timeout-less HTTP client on a black-holed connection.
+type blockingEspressoClient struct{}
+
+func (blockingEspressoClient) SubmitTransaction(ctx context.Context, tx espressoCommon.Transaction) (*espressoCommon.TaggedBase64, error) {
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+func (blockingEspressoClient) FetchTransactionByHash(ctx context.Context, hash *espressoCommon.TaggedBase64) (espressoCommon.TransactionQueryData, error) {
+	<-ctx.Done()
+	return espressoCommon.TransactionQueryData{}, ctx.Err()
+}
+
+func (blockingEspressoClient) FetchLatestBlockHeight(ctx context.Context) (uint64, error) {
+	<-ctx.Done()
+	return 0, ctx.Err()
+}
+
+// TestBoundedEspressoClientAppliesDeadline pins that every wrapped method
+// carries a per-call deadline even when the caller's context has none (as the
+// workers' long-lived loop contexts do not).
+func TestBoundedEspressoClientAppliesDeadline(t *testing.T) {
+	c := newBoundedEspressoClient(blockingEspressoClient{}, 10*time.Millisecond)
+
+	_, err := c.SubmitTransaction(context.Background(), espressoCommon.Transaction{})
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+
+	_, err = c.FetchTransactionByHash(context.Background(), nil)
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+
+	_, err = c.FetchLatestBlockHeight(context.Background())
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+}

op-batcher/batcher/espresso_driver.go

@@ -234,12 +234,25 @@ func (l *BatchSubmitter) waitForLocalSafeHead(ctx context.Context) (eth.L2BlockR
 func (l *BatchSubmitter) rollbackFailedStart() {
 	l.cancelShutdownCtx()
 	l.cancelKillCtx()
-	if l.espressoStreamer != nil {
-		l.espressoStreamer.Stop()
-	}
+	l.teardownEspressoStreamer()
 	l.running = false
 }
 
+// teardownEspressoStreamer stops and drops the streamer, if any. Every teardown
+// path (StopBatchSubmitting, rollbackFailedStart) must come through here: a dead
+// run's streamer must never survive into the next start, or that start's
+// clearState would run the espressoReanchorTarget gate against it — a retry
+// loop with no deadline, running under the start mutex that the stop able to
+// cancel it would itself need — to re-anchor an object the start is about to
+// replace anyway (setupEspressoStreamer anchors the fresh streamer itself).
+func (l *BatchSubmitter) teardownEspressoStreamer() {
+	if l.espressoStreamer == nil {
+		return
+	}
+	l.espressoStreamer.Stop()
+	l.espressoStreamer = nil
+}
+
 // startEspressoLoops registers the batcher with the BatchAuthenticator
 // contract, resolves the TEE verifier address, spawns the Espresso transaction
 // submitter, and starts the four Espresso-specific batcher goroutines (in
@@ -269,7 +282,7 @@ func (l *BatchSubmitter) startEspressoLoops(receiptsCh chan txmgr.TxReceipt[txRe
 	l.espressoSubmitter = NewEspressoTransactionSubmitter(
 		WithContext(l.shutdownCtx),
 		WithWaitGroup(l.wg),
-		WithEspressoClient(l.Espresso.Client),
+		WithEspressoClient(newBoundedEspressoClient(l.Espresso.Client, l.Config.NetworkTimeout)),
 		WithVerifyReceiptMaxBlocks(l.Config.Espresso.VerifyReceiptMaxBlocks),
 		WithVerifyReceiptSafetyTimeout(l.Config.Espresso.VerifyReceiptSafetyTimeout),
 		WithVerifyReceiptRetryDelay(l.Config.Espresso.VerifyReceiptRetryDelay),
@@ -329,7 +342,8 @@ func (l *BatchSubmitter) shouldSkipPublishForActiveSeq(ctx context.Context) bool
 // LocalSafeL2: the caller must retry the whole clear rather than perform it
 // partially. Reports a nil target (and ok=true) when there is nothing to
 // re-anchor: --espresso.enabled unset, or the startup path, where clearState
-// runs before the streamer is constructed.
+// runs before the streamer is constructed. Every start takes that path — see
+// teardownEspressoStreamer for why no earlier run's streamer can be left over.
 func (l *BatchSubmitter) espressoReanchorTarget(ctx context.Context) (target *eth.L2BlockRef, ok bool) {
 	if !l.Config.Espresso.Enabled || l.espressoStreamer == nil {
 		return nil, true

op-batcher/batcher/espresso_driver_test.go

@@ -0,0 +1,47 @@
+package batcher
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	espressoStreamers "github.com/EspressoSystems/espresso-streamers/op"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/stretchr/testify/require"
+
+	"github.com/ethereum-optimism/optimism/op-service/testlog"
+)
+
+// startedSubmitter returns a BatchSubmitter in the state StartBatchSubmitting
+// leaves it in right after taking ownership (running, fresh contexts, empty
+// waitgroup) with a streamer attached. A zero-value Streamer is safe here:
+// Stop() on it is a no-op (nil cancel), and these tests only exercise
+// teardown, never the streamer itself.
+func startedSubmitter(t *testing.T) *BatchSubmitter {
+	l := &BatchSubmitter{}
+	l.Log = testlog.Logger(t, log.LevelDebug)
+	l.running = true
+	l.shutdownCtx, l.cancelShutdownCtx = context.WithCancel(context.Background())
+	l.killCtx, l.cancelKillCtx = context.WithCancel(context.Background())
+	l.wg = &sync.WaitGroup{}
+	l.espressoStreamer = &espressoStreamers.Streamer{}
+	return l
+}
+
+// TestStopBatchSubmittingDropsStreamer pins that a stopped run's streamer does
+// not survive into the next start; see teardownEspressoStreamer for why.
+func TestStopBatchSubmittingDropsStreamer(t *testing.T) {
+	l := startedSubmitter(t)
+	require.NoError(t, l.StopBatchSubmitting(context.Background()))
+	require.Nil(t, l.espressoStreamer)
+	require.False(t, l.running)
+}
+
+// TestRollbackFailedStartDropsStreamer pins the same invariant for a start
+// that constructed a streamer and then failed.
+func TestRollbackFailedStartDropsStreamer(t *testing.T) {
+	l := startedSubmitter(t)
+	l.rollbackFailedStart()
+	require.Nil(t, l.espressoStreamer)
+	require.False(t, l.running)
+}