feat: implement multi-account support with round-robin address selection for DA submissions

Author: randygrok

block/internal/submitting/da_submitter.go

@@ -3,6 +3,7 @@ package submitting
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"time"
 
@@ -13,6 +14,7 @@ import (
 	"github.com/evstack/ev-node/block/internal/common"
 	coreda "github.com/evstack/ev-node/core/da"
 	"github.com/evstack/ev-node/pkg/config"
+	pkgda "github.com/evstack/ev-node/pkg/da"
 	"github.com/evstack/ev-node/pkg/genesis"
 	"github.com/evstack/ev-node/pkg/rpc/server"
 	"github.com/evstack/ev-node/pkg/signer"
@@ -125,6 +127,9 @@ type DASubmitter struct {
 	// calculate namespaces bytes once and reuse them
 	namespaceBz     []byte
 	namespaceDataBz []byte
+
+	// address selector for multi-account support
+	addressSelector pkgda.AddressSelector
 }
 
 // NewDASubmitter creates a new DA submitter
@@ -148,6 +153,17 @@ func NewDASubmitter(
 		metrics = common.NopMetrics()
 	}
 
+	// Create address selector based on configuration
+	var addressSelector pkgda.AddressSelector
+	if len(config.DA.SigningAddresses) > 0 {
+		addressSelector = pkgda.NewRoundRobinSelector(config.DA.SigningAddresses)
+		daSubmitterLogger.Info().
+			Int("num_addresses", len(config.DA.SigningAddresses)).
+			Msg("initialized round-robin address selector for multi-account DA submissions")
+	} else {
+		addressSelector = pkgda.NewNoOpSelector()
+	}
+
 	return &DASubmitter{
 		da:              da,
 		config:          config,
@@ -157,6 +173,7 @@ func NewDASubmitter(
 		logger:          daSubmitterLogger,
 		namespaceBz:     coreda.NamespaceFromString(config.DA.GetNamespace()).Bytes(),
 		namespaceDataBz: coreda.NamespaceFromString(config.DA.GetDataNamespace()).Bytes(),
+		addressSelector: addressSelector,
 	}
 }
 
@@ -236,7 +253,6 @@ func (s *DASubmitter) SubmitHeaders(ctx context.Context, cache cache.Manager) er
 		"header",
 		s.namespaceBz,
 		[]byte(s.config.DA.SubmitOptions),
-		cache,
 		func() uint64 { return cache.NumPendingHeaders() },
 	)
 }
@@ -280,7 +296,6 @@ func (s *DASubmitter) SubmitData(ctx context.Context, cache cache.Manager, signe
 		"data",
 		s.namespaceDataBz,
 		[]byte(s.config.DA.SubmitOptions),
-		cache,
 		func() uint64 { return cache.NumPendingData() },
 	)
 }
@@ -341,6 +356,41 @@ func (s *DASubmitter) createSignedData(dataList []*types.SignedData, signer sign
 	return signedDataList, nil
 }
 
+// mergeSubmitOptions merges the base submit options with a signing address.
+// If the base options are valid JSON, the signing address is added to the JSON object.
+// Otherwise, a new JSON object is created with just the signing address.
+// Returns the base options unchanged if no signing address is provided.
+func mergeSubmitOptions(baseOptions []byte, signingAddress string) ([]byte, error) {
+	if signingAddress == "" {
+		return baseOptions, nil
+	}
+
+	// Create options map with signing address
+	optionsMap := make(map[string]interface{})
+
+	// If base options are provided, try to parse them as JSON
+	if len(baseOptions) > 0 {
+		// Try to unmarshal existing options
+		if err := json.Unmarshal(baseOptions, &optionsMap); err != nil {
+			// If it's not valid JSON, start with empty map
+			// Log at debug level since this is expected for non-JSON options
+			optionsMap = make(map[string]interface{})
+		}
+	}
+
+	// Add or override the signing address
+	// Note: Uses "signer_address" to match Celestia's TxConfig JSON schema
+	optionsMap["signer_address"] = signingAddress
+
+	// Marshal back to JSON
+	mergedOptions, err := json.Marshal(optionsMap)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal submit options: %w", err)
+	}
+
+	return mergedOptions, nil
+}
+
 // submitToDA is a generic helper for submitting items to the DA layer with retry, backoff, and gas price logic.
 func submitToDA[T any](
 	s *DASubmitter,
@@ -351,7 +401,6 @@ func submitToDA[T any](
 	itemType string,
 	namespace []byte,
 	options []byte,
-	cache cache.Manager,
 	getTotalPendingFn func() uint64,
 ) error {
 	marshaled, err := marshalItems(ctx, items, marshalFn, itemType)
@@ -393,10 +442,22 @@ func submitToDA[T any](
 			return err
 		}
 
+		// Select signing address and merge with options
+		signingAddress := s.addressSelector.Next()
+		mergedOptions, err := mergeSubmitOptions(options, signingAddress)
+		if err != nil {
+			s.logger.Error().Err(err).Msg("failed to merge submit options with signing address")
+			return fmt.Errorf("failed to merge submit options: %w", err)
+		}
+
+		if signingAddress != "" {
+			s.logger.Debug().Str("signingAddress", signingAddress).Msg("using signing address for DA submission")
+		}
+
 		submitCtx, cancel := context.WithTimeout(ctx, submissionTimeout)
 		defer cancel()
 		// Perform submission
-		res := types.SubmitWithHelpers(submitCtx, s.da, s.logger, marshaled, rs.GasPrice, namespace, options)
+		res := types.SubmitWithHelpers(submitCtx, s.da, s.logger, marshaled, rs.GasPrice, namespace, mergedOptions)
 
 		// Record submission result for observability
 		if daVisualizationServer := server.GetDAVisualizationServer(); daVisualizationServer != nil {

block/internal/submitting/da_submitter_mocks_test.go

@@ -86,7 +86,6 @@ func TestSubmitToDA_MempoolRetry_IncreasesGasAndSucceeds(t *testing.T) {
 		nsBz,
 		opts,
 		nil,
-		nil,
 	)
 	assert.NoError(t, err)
 
@@ -138,7 +137,6 @@ func TestSubmitToDA_UnknownError_RetriesSameGasThenSucceeds(t *testing.T) {
 		nsBz,
 		opts,
 		nil,
-		nil,
 	)
 	assert.NoError(t, err)
 	assert.Equal(t, []float64{5.5, 5.5}, usedGas)
@@ -195,7 +193,6 @@ func TestSubmitToDA_TooBig_HalvesBatch(t *testing.T) {
 		nsBz,
 		opts,
 		nil,
-		nil,
 	)
 	assert.NoError(t, err)
 	assert.Equal(t, []int{4, 2}, batchSizes)
@@ -245,7 +242,6 @@ func TestSubmitToDA_SentinelNoGas_PreservesGasAcrossRetries(t *testing.T) {
 		nsBz,
 		opts,
 		nil,
-		nil,
 	)
 	assert.NoError(t, err)
 	assert.Equal(t, []float64{-1, -1}, usedGas)
@@ -286,7 +282,6 @@ func TestSubmitToDA_PartialSuccess_AdvancesWindow(t *testing.T) {
 		nsBz,
 		opts,
 		nil,
-		nil,
 	)
 	assert.NoError(t, err)
 	assert.Equal(t, 3, totalSubmitted)

block/internal/submitting/da_submitter_options_test.go

@@ -0,0 +1,124 @@
+package submitting
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMergeSubmitOptions_NoSigningAddress(t *testing.T) {
+	baseOptions := []byte(`{"key":"value"}`)
+
+	result, err := mergeSubmitOptions(baseOptions, "")
+	require.NoError(t, err)
+	assert.Equal(t, baseOptions, result, "should return unchanged options when no signing address")
+}
+
+func TestMergeSubmitOptions_EmptyBaseOptions(t *testing.T) {
+	signingAddress := "celestia1abc123"
+
+	result, err := mergeSubmitOptions([]byte{}, signingAddress)
+	require.NoError(t, err)
+
+	var resultMap map[string]interface{}
+	err = json.Unmarshal(result, &resultMap)
+	require.NoError(t, err)
+
+	assert.Equal(t, signingAddress, resultMap["signer_address"])
+}
+
+func TestMergeSubmitOptions_ValidJSON(t *testing.T) {
+	baseOptions := []byte(`{"existing":"option","number":42}`)
+	signingAddress := "celestia1def456"
+
+	result, err := mergeSubmitOptions(baseOptions, signingAddress)
+	require.NoError(t, err)
+
+	var resultMap map[string]interface{}
+	err = json.Unmarshal(result, &resultMap)
+	require.NoError(t, err)
+
+	assert.Equal(t, "option", resultMap["existing"])
+	assert.Equal(t, float64(42), resultMap["number"]) // JSON numbers are float64
+	assert.Equal(t, signingAddress, resultMap["signer_address"])
+}
+
+func TestMergeSubmitOptions_InvalidJSON(t *testing.T) {
+	baseOptions := []byte(`not-json-content`)
+	signingAddress := "celestia1ghi789"
+
+	result, err := mergeSubmitOptions(baseOptions, signingAddress)
+	require.NoError(t, err)
+
+	var resultMap map[string]interface{}
+	err = json.Unmarshal(result, &resultMap)
+	require.NoError(t, err)
+
+	// Should create new JSON object with just the signing address
+	assert.Equal(t, signingAddress, resultMap["signer_address"])
+	assert.Len(t, resultMap, 1, "should only contain signing address when base options are invalid JSON")
+}
+
+func TestMergeSubmitOptions_OverrideExistingAddress(t *testing.T) {
+	baseOptions := []byte(`{"signer_address":"old-address","other":"data"}`)
+	newAddress := "celestia1new456"
+
+	result, err := mergeSubmitOptions(baseOptions, newAddress)
+	require.NoError(t, err)
+
+	var resultMap map[string]interface{}
+	err = json.Unmarshal(result, &resultMap)
+	require.NoError(t, err)
+
+	assert.Equal(t, newAddress, resultMap["signer_address"], "should override existing signing address")
+	assert.Equal(t, "data", resultMap["other"])
+}
+
+func TestMergeSubmitOptions_NilBaseOptions(t *testing.T) {
+	signingAddress := "celestia1jkl012"
+
+	result, err := mergeSubmitOptions(nil, signingAddress)
+	require.NoError(t, err)
+
+	var resultMap map[string]interface{}
+	err = json.Unmarshal(result, &resultMap)
+	require.NoError(t, err)
+
+	assert.Equal(t, signingAddress, resultMap["signer_address"])
+}
+
+func TestMergeSubmitOptions_ComplexJSON(t *testing.T) {
+	baseOptions := []byte(`{
+		"nested": {
+			"key": "value"
+		},
+		"array": [1, 2, 3],
+		"bool": true
+	}`)
+	signingAddress := "celestia1complex"
+
+	result, err := mergeSubmitOptions(baseOptions, signingAddress)
+	require.NoError(t, err)
+
+	var resultMap map[string]interface{}
+	err = json.Unmarshal(result, &resultMap)
+	require.NoError(t, err)
+
+	// Check nested structure is preserved
+	nested, ok := resultMap["nested"].(map[string]interface{})
+	require.True(t, ok)
+	assert.Equal(t, "value", nested["key"])
+
+	// Check array is preserved
+	array, ok := resultMap["array"].([]interface{})
+	require.True(t, ok)
+	assert.Len(t, array, 3)
+
+	// Check bool is preserved
+	assert.Equal(t, true, resultMap["bool"])
+
+	// Check signing address was added
+	assert.Equal(t, signingAddress, resultMap["signer_address"])
+}

core/da/dummy.go

@@ -221,7 +221,7 @@ func (d *DummyDA) SubmitWithOptions(ctx context.Context, blobs []Blob, gasPrice
 
 		d.blobs[idStr] = blob
 		d.commitments[idStr] = commitment
-		d.proofs[idStr] = commitment // Simple proof
+		d.proofs[idStr] = commitment       // Simple proof
 		d.namespaceByID[idStr] = namespace // Store namespace for this blob
 
 		ids = append(ids, id)

docs/learn/config.md

@@ -24,6 +24,7 @@ This document provides a comprehensive reference for all configuration options a
   - [DA Gas Price](#da-gas-price)
   - [DA Gas Multiplier](#da-gas-multiplier)
   - [DA Submit Options](#da-submit-options)
+  - [DA Signing Addresses](#da-signing-addresses)
   - [DA Namespace](#da-namespace)
   - [DA Header Namespace](#da-namespace)
   - [DA Data Namespace](#da-data-namespace)
@@ -377,13 +378,15 @@ _Constant:_ `FlagDAGasMultiplier`
 ### DA Submit Options
 
 **Description:**
-Additional options passed to the DA layer when submitting data. The format and meaning of these options depend on the specific DA implementation being used.
+Additional options passed to the DA layer when submitting data. The format and meaning of these options depend on the specific DA implementation being used. For example, with Celestia, this can include custom gas settings or other submission parameters in JSON format.
+
+**Note:** If you configure multiple signing addresses (see [DA Signing Addresses](#da-signing-addresses)), the selected signing address will be automatically merged into these options as a JSON field `signer_address` (matching Celestia's TxConfig schema). If the base options are already valid JSON, the signing address is added to the existing object; otherwise, a new JSON object is created.
 
 **YAML:**
 
 ```yaml
 da:
-  submit_options: "{"key":"value"}" # Example, format depends on DA layer
+  submit_options: '{"key":"value"}' # Example, format depends on DA layer
 ```
 
 **Command-line Flag:**
@@ -392,6 +395,41 @@ _Example:_ `--rollkit.da.submit_options '{"custom_param":true}'`
 _Default:_ `""` (empty)
 _Constant:_ `FlagDASubmitOptions`
 
+### DA Signing Addresses
+
+**Description:**
+A comma-separated list of signing addresses to use for DA blob submissions. When multiple addresses are provided, they will be used in round-robin fashion to prevent sequence mismatches that can occur with high-throughput Cosmos SDK-based DA layers. This is particularly useful for Celestia when submitting many transactions concurrently.
+
+Each submission will select the next address in the list, and that address will be automatically added to the `submit_options` as `signer_address`. This ensures that the DA layer (e.g., celestia-node) uses the specified account for signing that particular blob submission.
+
+**Setup Requirements:**
+
+- All addresses must be loaded into the DA node's keyring and have sufficient funds for transaction fees
+- For Celestia, see the guide on setting up multiple accounts in the DA node documentation
+
+**YAML:**
+
+```yaml
+da:
+  signing_addresses:
+    - "celestia1abc123..."
+    - "celestia1def456..."
+    - "celestia1ghi789..."
+```
+
+**Command-line Flag:**
+`--rollkit.da.signing_addresses <string>`
+_Example:_ `--rollkit.da.signing_addresses celestia1abc...,celestia1def...,celestia1ghi...`
+_Default:_ `[]` (empty, uses default DA node behavior)
+_Constant:_ `FlagDASigningAddresses`
+
+**Behavior:**
+
+- If no signing addresses are configured, submissions use the DA layer's default signing behavior
+- If one address is configured, all submissions use that address
+- If multiple addresses are configured, they are used in round-robin order to distribute the load and prevent nonce/sequence conflicts
+- The address selection is thread-safe for concurrent submissions
+
 ### DA Namespace
 
 **Description:**