fix: Handle shutdown RPC errors gracefully (#2329)

<!--
Please read and fill out this form before submitting your PR.

Please make sure you have reviewed our contributors guide before
submitting your
first PR.

NOTE: PR titles should follow semantic commits:
https://www.conventionalcommits.org/en/v1.0.0/
-->

## Overview

Closes #2328 

<!-- 
Please provide an explanation of the PR, including the appropriate
context,
background, goal, and rationale. If there is an issue with this
information,
please provide a tl;dr and link the issue. 

Ex: Closes #<issue number>
-->


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Improved handling of submission cancellations, providing clearer
feedback when operations are canceled due to context termination.
- Added new status and error codes to better represent and communicate
context cancellation events during data availability operations.

- **Bug Fixes**
- Submission processes now exit promptly and log informative messages
when canceled, ensuring cleaner and more predictable behavior.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: Manav-Aggarwal

block/submitter.go

@@ -56,11 +56,11 @@ func (m *Manager) submitHeadersToDA(ctx context.Context) error {
 	gasPrice := m.gasPrice
 	initialGasPrice := gasPrice
 
-daSubmitRetryLoop:
 	for !submittedAllHeaders && attempt < maxSubmitAttempts {
 		select {
 		case <-ctx.Done():
-			break daSubmitRetryLoop
+			m.logger.Info("context done, stopping header submission loop")
+			return nil
 		case <-time.After(backoff):
 		}
 
@@ -78,9 +78,9 @@ daSubmitRetryLoop:
 			}
 		}
 
-		ctx, cancel := context.WithTimeout(ctx, 60*time.Second) // TODO: make this configurable
-		res := types.SubmitWithHelpers(ctx, m.da, m.logger, headersBz, gasPrice, nil)
-		cancel()
+		submitctx, submitCtxCancel := context.WithTimeout(ctx, 60*time.Second) // TODO: make this configurable
+		res := types.SubmitWithHelpers(submitctx, m.da, m.logger, headersBz, gasPrice, nil)
+		submitCtxCancel()
 
 		switch res.Code {
 		case coreda.StatusSuccess:
@@ -115,6 +115,9 @@ daSubmitRetryLoop:
 				gasPrice = gasPrice * m.gasMultiplier
 			}
 			m.logger.Info("retrying DA layer submission with", "backoff", backoff, "gasPrice", gasPrice)
+		case coreda.StatusContextCanceled:
+			m.logger.Info("DA layer submission canceled", "attempt", attempt)
+			return nil
 		default:
 			m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt)
 			backoff = m.exponentialBackoff(backoff)
@@ -175,12 +178,12 @@ func (m *Manager) submitBatchToDA(ctx context.Context, batch coresequencer.Batch
 	initialGasPrice := m.gasPrice
 	gasPrice := initialGasPrice
 
-daSubmitRetryLoop:
-	for !submittedAllTxs && attempt < maxSubmitAttempts {
+	for attempt < maxSubmitAttempts {
 		// Wait for backoff duration or exit if context is done
 		select {
 		case <-ctx.Done():
-			break daSubmitRetryLoop
+			m.logger.Info("context done, stopping batch submission loop")
+			return nil
 		case <-time.After(backoff):
 		}
 
@@ -251,12 +254,13 @@ daSubmitRetryLoop:
 				gasPrice = gasPrice * gasMultiplier
 			}
 			m.logger.Info("retrying DA layer submission with", "backoff", backoff, "gasPrice", gasPrice)
-
+		case coreda.StatusContextCanceled:
+			m.logger.Info("DA layer submission canceled due to context cancellation", "attempt", attempt)
+			return nil
 		case coreda.StatusTooBig:
 			// Blob size adjustment is handled within DA impl or SubmitWithOptions call
 			// fallthrough to default exponential backoff
 			fallthrough
-
 		default:
 			m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt)
 			backoff = m.exponentialBackoff(backoff)

core/da/da.go

@@ -91,6 +91,7 @@ const (
 	StatusContextDeadline
 	StatusError
 	StatusIncorrectAccountSequence
+	StatusContextCanceled
 )
 
 // BaseResult contains basic information returned by DA layer.

core/da/errors.go

@@ -12,4 +12,5 @@ var (
 	ErrTxIncorrectAccountSequence = errors.New("incorrect account sequence")
 	ErrContextDeadline            = errors.New("context deadline")
 	ErrHeightFromFuture           = errors.New("given height is from the future")
+	ErrContextCanceled            = errors.New("context canceled")
 )

da/jsonrpc/client.go

@@ -42,6 +42,10 @@ func (api *API) Get(ctx context.Context, ids []da.ID, _ []byte) ([]da.Blob, erro
 	api.Logger.Debug("Making RPC call", "method", "Get", "num_ids", len(ids), "namespace", string(api.Namespace))
 	res, err := api.Internal.Get(ctx, ids, api.Namespace)
 	if err != nil {
+		if errors.Is(err, context.Canceled) {
+			api.Logger.Debug("RPC call canceled due to context cancellation", "method", "Get")
+			return res, context.Canceled
+		}
 		api.Logger.Error("RPC call failed", "method", "Get", "error", err)
 		// Wrap error for context, potentially using the translated error from the RPC library
 		return nil, fmt.Errorf("failed to get blobs: %w", err)
@@ -64,6 +68,10 @@ func (api *API) GetIDs(ctx context.Context, height uint64, _ []byte) (*da.GetIDs
 			api.Logger.Debug("RPC call indicates height from future", "method", "GetIDs", "height", height)
 			return nil, err // Return the specific ErrHeightFromFuture
 		}
+		if errors.Is(err, context.Canceled) {
+			api.Logger.Debug("RPC call canceled due to context cancellation", "method", "GetIDs")
+			return res, context.Canceled
+		}
 		api.Logger.Error("RPC call failed", "method", "GetIDs", "error", err)
 		return nil, err
 	}
@@ -119,6 +127,10 @@ func (api *API) Submit(ctx context.Context, blobs []da.Blob, gasPrice float64, _
 	api.Logger.Debug("Making RPC call", "method", "Submit", "num_blobs", len(blobs), "gas_price", gasPrice, "namespace", string(api.Namespace))
 	res, err := api.Internal.Submit(ctx, blobs, gasPrice, api.Namespace)
 	if err != nil {
+		if errors.Is(err, context.Canceled) {
+			api.Logger.Debug("RPC call canceled due to context cancellation", "method", "Submit")
+			return res, context.Canceled
+		}
 		api.Logger.Error("RPC call failed", "method", "Submit", "error", err)
 	} else {
 		api.Logger.Debug("RPC call successful", "method", "Submit", "num_ids_returned", len(res))
@@ -168,6 +180,10 @@ func (api *API) SubmitWithOptions(ctx context.Context, inputBlobs []da.Blob, gas
 	api.Logger.Debug("Making RPC call", "method", "SubmitWithOptions", "num_blobs_original", len(inputBlobs), "num_blobs_to_submit", len(blobsToSubmit), "gas_price", gasPrice, "namespace", string(api.Namespace))
 	res, err := api.Internal.SubmitWithOptions(ctx, blobsToSubmit, gasPrice, api.Namespace, options)
 	if err != nil {
+		if errors.Is(err, context.Canceled) {
+			api.Logger.Debug("RPC call canceled due to context cancellation", "method", "SubmitWithOptions")
+			return res, context.Canceled
+		}
 		api.Logger.Error("RPC call failed", "method", "SubmitWithOptions", "error", err)
 	} else {
 		api.Logger.Debug("RPC call successful", "method", "SubmitWithOptions", "num_ids_returned", len(res))

da/jsonrpc/errors.go

@@ -15,5 +15,6 @@ func getKnownErrorsMapping() jsonrpc.Errors {
 	errs.Register(jsonrpc.ErrorCode(coreda.StatusAlreadyInMempool), &coreda.ErrTxAlreadyInMempool)
 	errs.Register(jsonrpc.ErrorCode(coreda.StatusIncorrectAccountSequence), &coreda.ErrTxIncorrectAccountSequence)
 	errs.Register(jsonrpc.ErrorCode(coreda.StatusContextDeadline), &coreda.ErrContextDeadline)
+	errs.Register(jsonrpc.ErrorCode(coreda.StatusContextCanceled), &coreda.ErrContextCanceled)
 	return errs
 }

types/da.go

@@ -29,6 +29,16 @@ func SubmitWithHelpers(
 
 	// Handle errors returned by SubmitWithOptions
 	if err != nil {
+		if errors.Is(err, context.Canceled) {
+			logger.Debug("DA submission canceled via helper due to context cancellation")
+			return coreda.ResultSubmit{
+				BaseResult: coreda.BaseResult{
+					Code:    coreda.StatusContextCanceled,
+					Message: "submission canceled",
+					IDs:     ids,
+				},
+			}
+		}
 		status := coreda.StatusError
 		switch {
 		case errors.Is(err, coreda.ErrTxTimedOut):