WIP feat(op-batcher): failover to ethda when altda is down

op-alt-da/daclient.go

@@ -16,6 +16,11 @@ var ErrNotFound = errors.New("not found")
 // ErrInvalidInput is returned when the input is not valid for posting to the DA storage.
 var ErrInvalidInput = errors.New("invalid input")
 
+// ErrAltDADown is returned when the alt DA returns a 503 status code.
+// It is used to signify that the alt DA is down and the client should failover to the eth DA.
+// See https://github.com/ethereum-optimism/specs/issues/434
+var ErrAltDADown = errors.New("alt DA is down: failover to eth DA")
+
 // DAClient is an HTTP client to communicate with a DA storage service.
 // It creates commitments and retrieves input data + verifies if needed.
 type DAClient struct {
@@ -131,6 +136,9 @@ func (c *DAClient) setInput(ctx context.Context, img []byte) (CommitmentData, er
 		return nil, err
 	}
 	defer resp.Body.Close()
+	if resp.StatusCode == http.StatusServiceUnavailable {
+		return nil, ErrAltDADown
+	}
 	if resp.StatusCode != http.StatusOK {
 		return nil, fmt.Errorf("failed to store data: %v", resp.StatusCode)
 	}

op-batcher/batcher/channel.go

@@ -46,7 +46,7 @@ func newChannel(log log.Logger, metr metrics.Metricer, cfg ChannelConfig, rollup
 
 // TxFailed records a transaction as failed. It will attempt to resubmit the data
 // in the failed transaction.
-func (c *channel) TxFailed(id string) {
+func (c *channel) TxFailed(id string, failover bool) {
 	if data, ok := c.pendingTransactions[id]; ok {
 		c.log.Trace("marked transaction as failed", "id", id)
 		// Note: when the batcher is changed to send multiple frames per tx,
@@ -57,7 +57,7 @@ func (c *channel) TxFailed(id string) {
 	} else {
 		c.log.Warn("unknown transaction marked as failed", "id", id)
 	}
-
+	c.cfg.
 	c.metr.RecordBatchTxFailed()
 }
 

op-batcher/batcher/channel_manager.go

@@ -89,13 +89,13 @@ func (s *channelManager) Clear(l1OriginLastClosedChannel eth.BlockID) {
 
 // TxFailed records a transaction as failed. It will attempt to resubmit the data
 // in the failed transaction.
-func (s *channelManager) TxFailed(_id txID) {
+func (s *channelManager) TxFailed(_id txID, failover bool) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	id := _id.String()
 	if channel, ok := s.txChannels[id]; ok {
 		delete(s.txChannels, id)
-		channel.TxFailed(id)
+		channel.TxFailed(id, failover)
 		if s.closed && channel.NoneSubmitted() {
 			s.log.Info("Channel has no submitted transactions, clearing for shutdown", "chID", channel.ID())
 			s.removePendingChannel(channel)
@@ -510,7 +510,7 @@ func (s *channelManager) Requeue(newCfg ChannelConfig) {
 		s.metr.RecordL2BlockInPendingQueue(b)
 	}
 
-	// Channels which where already being submitted are put back
+	// Channels which were already being submitted are put back
 	s.channelQueue = newChannelQueue
 	s.currentChannel = nil
 	// Setting the defaultCfg will cause new channels

op-batcher/batcher/driver.go

@@ -784,12 +784,13 @@ func (l *BatchSubmitter) recordFailedDARequest(id txID, err error) {
 	if err != nil {
 		l.Log.Warn("DA request failed", logFields(id, err)...)
 	}
-	l.state.TxFailed(id)
+	failover := errors.Is(err, altda.ErrAltDADown)
+	l.state.TxFailed(id, failover)
 }
 
 func (l *BatchSubmitter) recordFailedTx(id txID, err error) {
 	l.Log.Warn("Transaction failed to send", logFields(id, err)...)
-	l.state.TxFailed(id)
+	l.state.TxFailed(id, false)
 }
 
 func (l *BatchSubmitter) recordConfirmedTx(id txID, receipt *types.Receipt) {