op-batcher: exit process on criticial throttling RPC error

op-batcher/batcher/batch_submitter.go

@@ -29,6 +29,6 @@ func Main(version string) cliapp.LifecycleAction {
 		opservice.ValidateEnvVars(flags.EnvVarPrefix, flags.Flags, l)
 
 		l.Info("Initializing Batch Submitter")
-		return BatcherServiceFromCLIConfig(cliCtx.Context, version, cfg, l)
+		return BatcherServiceFromCLIConfig(cliCtx.Context, version, cfg, l, closeApp)
 	}
 }

op-batcher/batcher/driver.go

@@ -99,6 +99,7 @@ type DriverSetup struct {
 // batches to L1 for availability.
 type BatchSubmitter struct {
 	DriverSetup
+	closeApp context.CancelCauseFunc
 
 	wg                               *sync.WaitGroup
 	shutdownCtx, killCtx             context.Context
@@ -121,13 +122,14 @@ type BatchSubmitter struct {
 }
 
 // NewBatchSubmitter initializes the BatchSubmitter driver from a preconfigured DriverSetup
-func NewBatchSubmitter(setup DriverSetup) *BatchSubmitter {
+func NewBatchSubmitter(setup DriverSetup, closeApp context.CancelCauseFunc) *BatchSubmitter {
 	state := NewChannelManager(setup.Log, setup.Metr, setup.ChannelConfig, setup.RollupConfig)
 	if setup.ChannelOutFactory != nil {
 		state.SetChannelOutFactory(setup.ChannelOutFactory)
 	}
 
 	batcher := &BatchSubmitter{
+		closeApp:    closeApp,
 		DriverSetup: setup,
 		channelMgr:  state,
 	}
@@ -623,20 +625,12 @@ func (l *BatchSubmitter) singleEndpointThrottler(wg *sync.WaitGroup, throttleSig
 			return
 		}
 
-		var rpcErr rpc.Error
-		if errors.As(err, &rpcErr) && eth.ErrorCode(rpcErr.ErrorCode()).IsGenericRPCError() {
-			l.Log.Error("SetMaxDASize RPC method unavailable on endpoint, shutting down. Either enable it or disable throttling.",
-				"endpoint", endpoint, "err", err)
-
-			// We have a strict requirement that all endpoints must have the SetMaxDASize endpoint, and shut down if this RPC method is not available
-			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-			defer cancel()
-			// Call StopBatchSubmitting in another goroutine to avoid deadlock.
-			go func() {
-				_ = l.StopBatchSubmitting(ctx)
-			}()
-			return
+		if isCriticalThrottlingRPCError(err) {
+			// We have a strict requirement that all endpoints must have the SetMaxDASize endpoint,
+			// and shut down if this RPC method is not available or returns another application-level error.
+			l.shutdownOnCriticalError(fmt.Errorf("SetMaxDASize RPC method unavailable at %s,  either enable it or disable throttling: %w", endpoint, err))
 		} else if err != nil {
+			// Transport-level errors are retried.
 			l.Log.Warn("SetMaxDASize RPC failed for endpoint, retrying.", "endpoint", endpoint, "err", err)
 			retryTimer.Reset(retryInterval)
 			return
@@ -671,6 +665,17 @@ func (l *BatchSubmitter) singleEndpointThrottler(wg *sync.WaitGroup, throttleSig
 	}
 }
 
+func isCriticalThrottlingRPCError(err error) bool {
+	var rpcErr rpc.Error
+	return errors.As(err, &rpcErr) && eth.ErrorCode(rpcErr.ErrorCode()).IsGenericRPCError()
+}
+
+func (l *BatchSubmitter) shutdownOnCriticalError(err error) {
+	l.Log.Error("Shutting down batcher on critical error", "err", err)
+	// Call closeApp to trigger process to exit (gracefully)
+	l.closeApp(err)
+}
+
 // throttlingLoop acts as a distributor that spawns individual throttling loops for each endpoint
 // and fans out the unsafe bytes updates to each endpoint
 func (l *BatchSubmitter) throttlingLoop(wg *sync.WaitGroup, unsafeBytesUpdated chan int64) {

op-batcher/batcher/driver_test.go

@@ -66,7 +66,7 @@ func setup(t *testing.T) (*BatchSubmitter, *mockL2EndpointProvider) {
 		},
 		ChannelConfig:    defaultTestChannelConfig(),
 		EndpointProvider: ep,
-	}), ep
+	}, nil), ep
 }
 
 func TestBatchSubmitter_SafeL1Origin(t *testing.T) {

op-batcher/batcher/service.go

@@ -84,15 +84,15 @@ type DriverSetupOption func(setup *DriverSetup)
 // BatcherServiceFromCLIConfig creates a new BatcherService from a CLIConfig.
 // The service components are fully started, except for the driver,
 // which will not be submitting batches (if it was configured to) until the Start part of the lifecycle.
-func BatcherServiceFromCLIConfig(ctx context.Context, version string, cfg *CLIConfig, log log.Logger, opts ...DriverSetupOption) (*BatcherService, error) {
+func BatcherServiceFromCLIConfig(ctx context.Context, version string, cfg *CLIConfig, log log.Logger, closeApp context.CancelCauseFunc, opts ...DriverSetupOption) (*BatcherService, error) {
 	var bs BatcherService
-	if err := bs.initFromCLIConfig(ctx, version, cfg, log, opts...); err != nil {
+	if err := bs.initFromCLIConfig(ctx, version, cfg, log, closeApp, opts...); err != nil {
 		return nil, errors.Join(err, bs.Stop(ctx)) // try to clean up our failed initialization attempt
 	}
 	return &bs, nil
 }
 
-func (bs *BatcherService) initFromCLIConfig(ctx context.Context, version string, cfg *CLIConfig, log log.Logger, opts ...DriverSetupOption) error {
+func (bs *BatcherService) initFromCLIConfig(ctx context.Context, version string, cfg *CLIConfig, log log.Logger, closeApp context.CancelCauseFunc, opts ...DriverSetupOption) error {
 	bs.Version = version
 	bs.Log = log
 	bs.NotSubmittingOnStart = cfg.Stopped
@@ -182,7 +182,7 @@ func (bs *BatcherService) initFromCLIConfig(ctx context.Context, version string,
 	if err := bs.initPProf(cfg); err != nil {
 		return fmt.Errorf("failed to init profiling: %w", err)
 	}
-	bs.initDriver(opts...)
+	bs.initDriver(closeApp, opts...)
 	if err := bs.initRPCServer(cfg); err != nil {
 		return fmt.Errorf("failed to start RPC server: %w", err)
 	}
@@ -381,7 +381,7 @@ func (bs *BatcherService) initMetricsServer(cfg *CLIConfig) error {
 	return nil
 }
 
-func (bs *BatcherService) initDriver(opts ...DriverSetupOption) {
+func (bs *BatcherService) initDriver(closeApp context.CancelCauseFunc, opts ...DriverSetupOption) {
 	ds := DriverSetup{
 		Log:              bs.Log,
 		Metr:             bs.Metrics,
@@ -396,7 +396,7 @@ func (bs *BatcherService) initDriver(opts ...DriverSetupOption) {
 	for _, opt := range opts {
 		opt(&ds)
 	}
-	bs.driver = NewBatchSubmitter(ds)
+	bs.driver = NewBatchSubmitter(ds, closeApp)
 }
 
 func (bs *BatcherService) initRPCServer(cfg *CLIConfig) error {

op-devstack/sysgo/l2_batcher.go

@@ -117,7 +117,7 @@ func WithBatcher(batcherID stack.L2BatcherID, l1ELID stack.L1ELNodeID, l2CLID st
 
 		batcher, err := bss.BatcherServiceFromCLIConfig(
 			p.Ctx(), "0.0.1", batcherCLIConfig,
-			logger)
+			logger, nil)
 		require.NoError(err)
 		require.NoError(batcher.Start(p.Ctx()))
 		p.Cleanup(func() {

op-e2e/interop/supersystem_l2.go

@@ -301,7 +301,7 @@ func (s *interopE2ESystem) newBatcherForL2(
 	}
 	batcher, err := bss.BatcherServiceFromCLIConfig(
 		context.Background(), "0.0.1", batcherCLIConfig,
-		logger.New("service", "batcher"))
+		logger.New("service", "batcher"), nil)
 	require.NoError(s.t, err)
 	require.NoError(s.t, batcher.Start(context.Background()))
 	s.t.Cleanup(func() {

op-e2e/system/conductor/sequencer_failover_setup.go

@@ -308,7 +308,7 @@ func setupBatcher(t *testing.T, sys *e2esys.System, conductors map[string]*condu
 		CompressionAlgo:              derive.Zlib,
 	}
 
-	batcher, err := bss.BatcherServiceFromCLIConfig(context.Background(), "0.0.1", batcherCLIConfig, sys.Cfg.Loggers["batcher"])
+	batcher, err := bss.BatcherServiceFromCLIConfig(context.Background(), "0.0.1", batcherCLIConfig, sys.Cfg.Loggers["batcher"], nil)
 	require.NoError(t, err)
 	err = batcher.Start(context.Background())
 	require.NoError(t, err)

op-e2e/system/e2esys/setup.go

@@ -1025,7 +1025,7 @@ func (cfg SystemConfig) Start(t *testing.T, startOpts ...StartOption) (*System,
 	}
 
 	// Batch Submitter
-	batcher, err := bss.BatcherServiceFromCLIConfig(context.Background(), "0.0.1", batcherCLIConfig, sys.Cfg.Loggers["batcher"])
+	batcher, err := bss.BatcherServiceFromCLIConfig(context.Background(), "0.0.1", batcherCLIConfig, sys.Cfg.Loggers["batcher"], nil)
 	if err != nil {
 		return nil, fmt.Errorf("failed to setup batch submitter: %w", err)
 	}