op-node: gracefully retry da connection err

op-node/rollup/derive/calldata_source.go

@@ -75,9 +75,20 @@ func NewDataSource(ctx context.Context, log log.Logger, cfg *rollup.Config, daCf
 			batcherAddr: batcherAddr,
 		}
 	} else {
+		data, err := DataFromEVMTransactions(cfg, daCfg, batcherAddr, txs, log.New("origin", block))
+		if err != nil {
+			return &DataSource{
+				open:        false,
+				id:          block,
+				cfg:         cfg,
+				fetcher:     fetcher,
+				log:         log,
+				batcherAddr: batcherAddr,
+			}
+		}
 		return &DataSource{
 			open: true,
-			data: DataFromEVMTransactions(cfg, daCfg, batcherAddr, txs, log.New("origin", block)),
+			data: data,
 		}
 	}
 }
@@ -89,7 +100,10 @@ func (ds *DataSource) Next(ctx context.Context) (eth.Data, error) {
 	if !ds.open {
 		if _, txs, err := ds.fetcher.InfoAndTxsByHash(ctx, ds.id.Hash); err == nil {
 			ds.open = true
-			ds.data = DataFromEVMTransactions(ds.cfg, ds.daCfg, ds.batcherAddr, txs, log.New("origin", ds.id))
+			ds.data, err = DataFromEVMTransactions(ds.cfg, ds.daCfg, ds.batcherAddr, txs, log.New("origin", ds.id))
+			if err != nil {
+				return nil, NewTemporaryError(fmt.Errorf("failed to open calldata source: %w", err))
+			}
 		} else if errors.Is(err, ethereum.NotFound) {
 			return nil, NewResetError(fmt.Errorf("failed to open calldata source: %w", err))
 		} else {
@@ -108,7 +122,7 @@ func (ds *DataSource) Next(ctx context.Context) (eth.Data, error) {
 // DataFromEVMTransactions filters all of the transactions and returns the calldata from transactions
 // that are sent to the batch inbox address from the batch sender address.
 // This will return an empty array if no valid transactions are found.
-func DataFromEVMTransactions(config *rollup.Config, daCfg *rollup.DAConfig, batcherAddr common.Address, txs types.Transactions, log log.Logger) []eth.Data {
+func DataFromEVMTransactions(config *rollup.Config, daCfg *rollup.DAConfig, batcherAddr common.Address, txs types.Transactions, log log.Logger) ([]eth.Data, error) {
 	var out []eth.Data
 	l1Signer := config.L1Signer()
 	for j, tx := range txs {
@@ -124,19 +138,24 @@ func DataFromEVMTransactions(config *rollup.Config, daCfg *rollup.DAConfig, batc
 				continue // not an authorized batch submitter, ignore
 			}
 
-			height, index, err := decodeETHData(tx.Data())
-			if err != nil {
-				log.Warn("unable to decode data pointer", "index", j, "err", err)
-				continue
-			}
-			data, err := daCfg.Client.NamespacedData(context.Background(), daCfg.NamespaceId, uint64(height))
-			if err != nil {
-				log.Warn("unable to retrieve data from da", "err", err)
+			if daCfg != nil {
+				height, index, err := decodeETHData(tx.Data())
+				if err != nil {
+					log.Warn("unable to decode data pointer", "index", j, "err", err)
+					continue
+				}
+				data, err := daCfg.Client.NamespacedData(context.Background(), daCfg.NamespaceId, uint64(height))
+				if err != nil {
+					log.Warn("unable to retrieve data from da", "err", err)
+					return nil, err
+				}
+				out = append(out, data[index])
+			} else {
+				out = append(out, tx.Data())
 			}
-			out = append(out, data[index])
 		}
 	}
-	return out
+	return out, nil
 }
 
 // decodeETHData will decode the data retrieved from the EVM, this data

op-node/rollup/derive/calldata_source_test.go

@@ -121,8 +121,9 @@ func TestDataFromEVMTransactions(t *testing.T) {
 			}
 		}
 
-		out := DataFromEVMTransactions(cfg, batcherAddr, txs, testlog.Logger(t, log.LvlCrit))
+		out, err := DataFromEVMTransactions(cfg, nil, batcherAddr, txs, testlog.Logger(t, log.LvlCrit))
 		require.ElementsMatch(t, expectedData, out)
+		require.NoError(t, err)
 	}
 
 }