feat: support p2p-only followers without DA

CHANGELOG.md

@@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Allow full followers without a DA endpoint to sync through P2P only, and reject aggregator startup without an explicit DA endpoint.
 - Reject unsupported `net-info --output` formats instead of falling back to text output [#3360](https://github.com/evstack/ev-node/pull/3360).
 
 ## v1.1.4

apps/evm/cmd/run.go

@@ -40,7 +40,7 @@ var RunCmd = &cobra.Command{
 	Aliases: []string{"node", "run"},
 	Short:   "Run the evolve node with EVM execution client",
 	RunE: func(cmd *cobra.Command, args []string) error {
-		nodeConfig, err := rollcmd.ParseConfig(cmd)
+		nodeConfig, err := rollcmd.ParseStartConfig(cmd)
 		if err != nil {
 			return err
 		}
@@ -59,13 +59,17 @@ var RunCmd = &cobra.Command{
 			return err
 		}
 
-		blobClient, err := blobrpc.NewWSClient(cmd.Context(), logger, nodeConfig.DA.Address, nodeConfig.DA.AuthToken, "")
-		if err != nil {
-			return fmt.Errorf("failed to create blob client: %w", err)
+		var daClient block.FullDAClient
+		if nodeConfig.DAEnabled() {
+			blobClient, err := blobrpc.NewWSClient(cmd.Context(), logger, nodeConfig.GetDAAddress(), nodeConfig.DA.AuthToken, "")
+			if err != nil {
+				return fmt.Errorf("failed to create blob client: %w", err)
+			}
+			defer blobClient.Close()
+			daClient = block.NewDAClient(blobClient, nodeConfig, logger)
+		} else {
+			logger.Info().Msg("DA address is not configured; syncing through P2P only")
 		}
-		defer blobClient.Close()
-
-		daClient := block.NewDAClient(blobClient, nodeConfig, logger)
 
 		headerNamespace := da.NamespaceFromString(nodeConfig.DA.GetNamespace())
 		dataNamespace := da.NamespaceFromString(nodeConfig.DA.GetDataNamespace())
@@ -100,6 +104,10 @@ var RunCmd = &cobra.Command{
 		}
 
 		if forceInclusionAddr != "" {
+			if daClient == nil {
+				return fmt.Errorf("force inclusion server requires a DA address")
+			}
+
 			ethURL, err := cmd.Flags().GetString(evm.FlagEvmEthURL)
 			if err != nil {
 				return fmt.Errorf("failed to get '%s' flag: %w", evm.FlagEvmEthURL, err)

apps/testapp/cmd/init_test.go

@@ -47,7 +47,7 @@ func TestInitCommand(t *testing.T) {
 	require.NoError(t, os.WriteFile(passphraseFile, []byte("test"), 0600))
 
 	// Set home flag to the test directory
-	cmd.SetArgs([]string{"init", "--home", dir, "--evnode.node.aggregator", "--evnode.signer.passphrase_file", passphraseFile})
+	cmd.SetArgs([]string{"init", "--home", dir, "--evnode.node.aggregator", "--evnode.da.address", "http://localhost:7980", "--evnode.signer.passphrase_file", passphraseFile})
 
 	// Execute the command
 	err = cmd.Execute()

apps/testapp/cmd/run.go

@@ -33,7 +33,7 @@ var RunCmd = &cobra.Command{
 	Aliases: []string{"node", "run"},
 	Short:   "Run the testapp node",
 	RunE: func(command *cobra.Command, args []string) error {
-		nodeConfig, err := cmd.ParseConfig(command)
+		nodeConfig, err := cmd.ParseStartConfig(command)
 		if err != nil {
 			return err
 		}
@@ -113,6 +113,10 @@ func createSequencer(
 	genesis genesis.Genesis,
 	executor execution.Executor,
 ) (coresequencer.Sequencer, error) {
+	if !nodeConfig.Node.Aggregator {
+		return nil, nil
+	}
+
 	if enabled, _ := cmd.Flags().GetBool(flagSoloSequencer); enabled {
 		if nodeConfig.Node.BasedSequencer {
 			return nil, fmt.Errorf("solo sequencer cannot be used with based")
@@ -121,7 +125,7 @@ func createSequencer(
 		return solo.NewSoloSequencer(logger, []byte(genesis.ChainID), executor), nil
 	}
 
-	blobClient, err := blobrpc.NewWSClient(ctx, logger, nodeConfig.DA.Address, nodeConfig.DA.AuthToken, "")
+	blobClient, err := blobrpc.NewWSClient(ctx, logger, nodeConfig.GetDAAddress(), nodeConfig.DA.AuthToken, "")
 	if err != nil {
 		return nil, fmt.Errorf("failed to create blob client: %w", err)
 	}

block/components.go

@@ -184,24 +184,27 @@ func NewSyncComponents(
 	}
 	pruner := pruner.New(logger, store, execPruner, config.Pruning, config.Node.BlockTime.Duration, config.DA.Address)
 
-	// Create submitter for sync nodes (no signer, only DA inclusion processing)
-	var daSubmitter submitting.DASubmitterAPI = submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger, headerDAHintAppender, dataDAHintAppender)
-	if config.Instrumentation.IsTracingEnabled() {
-		daSubmitter = submitting.WithTracingDASubmitter(daSubmitter)
+	var submitter *submitting.Submitter
+	if daClient != nil {
+		// Create submitter for DA-enabled sync nodes (no signer, only DA inclusion processing).
+		var daSubmitter submitting.DASubmitterAPI = submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger, headerDAHintAppender, dataDAHintAppender)
+		if config.Instrumentation.IsTracingEnabled() {
+			daSubmitter = submitting.WithTracingDASubmitter(daSubmitter)
+		}
+		submitter = submitting.NewSubmitter(
+			store,
+			exec,
+			cacheManager,
+			metrics,
+			config,
+			genesis,
+			daSubmitter,
+			nil, // No sequencer for sync nodes
+			nil, // No signer for sync nodes
+			logger,
+			errorCh,
+		)
 	}
-	submitter := submitting.NewSubmitter(
-		store,
-		exec,
-		cacheManager,
-		metrics,
-		config,
-		genesis,
-		daSubmitter,
-		nil, // No sequencer for sync nodes
-		nil, // No signer for sync nodes
-		logger,
-		errorCh,
-	)
 
 	return &Components{
 		Syncer:    syncer,

block/components_test.go

@@ -132,6 +132,39 @@ func TestNewSyncComponents_Creation(t *testing.T) {
 	assert.Nil(t, components.Executor) // Sync nodes don't have executors
 }
 
+func TestNewSyncComponents_WithoutDA(t *testing.T) {
+	ds := sync.MutexWrap(datastore.NewMapDatastore())
+	memStore := store.New(ds)
+
+	cfg := config.DefaultConfig()
+	gen := genesis.Genesis{
+		ChainID:         "test-chain",
+		InitialHeight:   1,
+		StartTime:       time.Now(),
+		ProposerAddress: []byte("test-proposer"),
+	}
+
+	components, err := NewSyncComponents(
+		cfg,
+		gen,
+		memStore,
+		testmocks.NewMockExecutor(t),
+		nil,
+		extmocks.NewMockStore[*types.P2PSignedHeader](t),
+		extmocks.NewMockStore[*types.P2PData](t),
+		noopDAHintAppender{},
+		noopDAHintAppender{},
+		zerolog.Nop(),
+		NopMetrics(),
+		DefaultBlockOptions(),
+		nil,
+	)
+
+	require.NoError(t, err)
+	assert.NotNil(t, components.Syncer)
+	assert.Nil(t, components.Submitter)
+}
+
 func TestNewAggregatorComponents_Creation(t *testing.T) {
 	ds := sync.MutexWrap(datastore.NewMapDatastore())
 	memStore := store.New(ds)

block/internal/syncing/syncer.go

@@ -181,16 +181,18 @@ func (s *Syncer) Start(ctx context.Context) (err error) {
 		return fmt.Errorf("failed to initialize syncer state: %w", err)
 	}
 
-	// Initialize handlers
-	daRetriever := NewDARetriever(s.daClient, s.cache, s.genesis, s.logger)
-	daRetriever.setExpectedProposerProvider(s.expectedProposerForHeight)
-	s.daRetriever = daRetriever
-	if s.config.Instrumentation.IsTracingEnabled() {
-		s.daRetriever = WithTracingDARetriever(s.daRetriever)
-	}
+	if s.daClient != nil {
+		// Initialize DA-specific handlers only when DA is configured.
+		daRetriever := NewDARetriever(s.daClient, s.cache, s.genesis, s.logger)
+		daRetriever.setExpectedProposerProvider(s.expectedProposerForHeight)
+		s.daRetriever = daRetriever
+		if s.config.Instrumentation.IsTracingEnabled() {
+			s.daRetriever = WithTracingDARetriever(s.daRetriever)
+		}
 
-	s.fiRetriever = da.NewForcedInclusionRetriever(s.daClient, s.logger, s.config.DA.BlockTime.Duration, s.config.Instrumentation.IsTracingEnabled(), s.genesis.DAStartHeight, s.genesis.DAEpochForcedInclusion)
-	s.fiRetriever.Start(ctx)
+		s.fiRetriever = da.NewForcedInclusionRetriever(s.daClient, s.logger, s.config.DA.BlockTime.Duration, s.config.Instrumentation.IsTracingEnabled(), s.genesis.DAStartHeight, s.genesis.DAEpochForcedInclusion)
+		s.fiRetriever.Start(ctx)
+	}
 	s.p2pHandler = NewP2PHandler(s.headerStore, s.dataStore, s.cache, s.genesis, s.logger)
 
 	currentHeight, initErr := s.store.Height(ctx)
@@ -213,19 +215,21 @@ func (s *Syncer) Start(ctx context.Context) (err error) {
 	// Start main processing loop
 	s.wg.Go(func() { s.processLoop(ctx) })
 
-	// Start the DA follower (subscribe + catchup) and other workers
-	s.daFollower = NewDAFollower(DAFollowerConfig{
-		Client:        s.daClient,
-		Retriever:     s.daRetriever,
-		Logger:        s.logger,
-		EventSink:     s,
-		Namespace:     s.daClient.GetHeaderNamespace(),
-		DataNamespace: s.daClient.GetDataNamespace(),
-		StartDAHeight: s.daRetrieverHeight.Load(),
-		DABlockTime:   s.config.DA.BlockTime.Duration,
-	})
-	if err = s.daFollower.Start(ctx); err != nil {
-		return fmt.Errorf("failed to start DA follower: %w", err)
+	if s.daClient != nil {
+		// Start the DA follower (subscribe + catchup) when DA is configured.
+		s.daFollower = NewDAFollower(DAFollowerConfig{
+			Client:        s.daClient,
+			Retriever:     s.daRetriever,
+			Logger:        s.logger,
+			EventSink:     s,
+			Namespace:     s.daClient.GetHeaderNamespace(),
+			DataNamespace: s.daClient.GetDataNamespace(),
+			StartDAHeight: s.daRetrieverHeight.Load(),
+			DABlockTime:   s.config.DA.BlockTime.Duration,
+		})
+		if err = s.daFollower.Start(ctx); err != nil {
+			return fmt.Errorf("failed to start DA follower: %w", err)
+		}
 	}
 
 	s.startSyncWorkers(ctx)
@@ -602,7 +606,7 @@ func (s *Syncer) processHeightEvent(ctx context.Context, event *common.DAHeightE
 
 	// If this is a P2P event with a DA height hint, trigger targeted DA retrieval
 	// This allows us to fetch the block directly from the specified DA height instead of sequential scanning
-	if event.Source == common.SourceP2P {
+	if event.Source == common.SourceP2P && s.daFollower != nil {
 		var daHeightHints []uint64
 		switch {
 		case event.DaHeightHints == [2]uint64{0, 0}:

docs/learn/config.md

@@ -472,7 +472,7 @@ da:
 ### DA Service Address
 
 **Description:**
-The network address (host:port) of the Data Availability layer service. Evolve connects to this endpoint to submit and retrieve block data.
+The network address (host:port) of the Data Availability layer service. A full follower with this value unset syncs exclusively through P2P and does not create a DA client. Aggregators must configure this value before startup.
 
 **YAML:**
 
@@ -484,7 +484,7 @@ da:
 **Command-line Flag:**
 `--evnode.da.address <string>`
 _Example:_ `--evnode.da.address 192.168.1.100:26659`
-_Default:_ `"http://localhost:7980"`
+_Default:_ `""` (empty; P2P-only follower)
 _Constant:_ `FlagDAAddress`
 
 ### DA Authentication Token

docs/learn/specs/da.md

@@ -6,7 +6,7 @@ Evolve provides a generic [data availability interface][da-interface] for modula
 
 `Client` can connect via JSON-RPC transports using Evolve's [jsonrpc][jsonrpc] implementations. The connection can be configured using the following cli flags:
 
-* `--rollkit.da.address`: url address of the DA service (default: "grpc://localhost:26650")
+* `--rollkit.da.address`: URL address of the DA service (default: empty for P2P-only followers; required for aggregators)
 * `--rollkit.da.auth_token`: authentication token of the DA service
 * `--rollkit.da.namespace`: namespace to use when submitting blobs to the DA service (deprecated)
 * `--rollkit.da.header_namespace`: namespace to use when submitting headers to the DA service

docs/reference/specs/da.md

@@ -6,7 +6,7 @@ Evolve provides a generic [data availability interface][da-interface] for modula
 
 `Client` can connect via JSON-RPC transports using Evolve's [jsonrpc][jsonrpc] implementations. The connection can be configured using the following cli flags:
 
-* `--rollkit.da.address`: url address of the DA service (default: "grpc://localhost:26650")
+* `--rollkit.da.address`: URL address of the DA service (default: empty for P2P-only followers; required for aggregators)
 * `--rollkit.da.auth_token`: authentication token of the DA service
 * `--rollkit.da.namespace`: namespace to use when submitting blobs to the DA service (deprecated)
 * `--rollkit.da.header_namespace`: namespace to use when submitting headers to the DA service

pkg/cmd/run_node.go

@@ -42,6 +42,19 @@ func ParseConfig(cmd *cobra.Command) (rollconf.Config, error) {
 	return nodeConfig, nil
 }
 
+// ParseStartConfig loads and validates configuration required to start a node.
+func ParseStartConfig(cmd *cobra.Command) (rollconf.Config, error) {
+	nodeConfig, err := ParseConfig(cmd)
+	if err != nil {
+		return rollconf.Config{}, err
+	}
+	if nodeConfig.Node.Aggregator && nodeConfig.GetDAAddress() == "" {
+		return rollconf.Config{}, fmt.Errorf("DA address is required when aggregator mode is enabled")
+	}
+
+	return nodeConfig, nil
+}
+
 // SetupLogger configures and returns a logger based on the provided configuration.
 // It applies the following settings from the config:
 //   - Log format (text or JSON)
@@ -150,12 +163,17 @@ func StartNode(
 		}
 	}
 
-	blobClient, err := blobrpc.NewWSClient(ctx, logger, nodeConfig.DA.Address, nodeConfig.DA.AuthToken, "")
-	if err != nil {
-		return fmt.Errorf("failed to create blob client: %w", err)
+	var daClient block.FullDAClient
+	if nodeConfig.DAEnabled() {
+		blobClient, err := blobrpc.NewWSClient(ctx, logger, nodeConfig.GetDAAddress(), nodeConfig.DA.AuthToken, "")
+		if err != nil {
+			return fmt.Errorf("failed to create blob client: %w", err)
+		}
+		defer blobClient.Close()
+		daClient = block.NewDAClient(blobClient, nodeConfig, logger)
+	} else {
+		logger.Info().Msg("DA address is not configured; syncing through P2P only")
 	}
-	defer blobClient.Close()
-	daClient := block.NewDAClient(blobClient, nodeConfig, logger)
 
 	// sanity check for based sequencer
 	if nodeConfig.Node.BasedSequencer && genesis.DAStartHeight == 0 {

pkg/cmd/run_node_test.go

@@ -147,6 +147,7 @@ func TestAggregatorFlagInvariants(t *testing.T) {
 	validValues := []bool{false, true, true}
 
 	for i, flags := range flagVariants {
+		flags = append(flags, "--rollkit.da.address", "http://da.example:7980")
 		args := append([]string{"start"}, flags...)
 
 		executor, sequencer, keyProvider, nodeKey, ds, stopDAHeightTicker := createTestComponents(context.Background(), t)
@@ -173,6 +174,23 @@ func TestAggregatorFlagInvariants(t *testing.T) {
 	}
 }
 
+func TestParseStartConfig_AggregatorRequiresDAAddress(t *testing.T) {
+	executor, sequencer, keyProvider, nodeKey, ds, stopDAHeightTicker := createTestComponents(context.Background(), t)
+	defer stopDAHeightTicker()
+
+	nodeConfig := rollconf.DefaultConfig()
+	nodeConfig.RootDir = t.TempDir()
+
+	cmd := newRunNodeCmd(t.Context(), executor, sequencer, keyProvider, nodeKey, ds, nodeConfig)
+	require.NoError(t, cmd.ParseFlags([]string{"start", "--rollkit.node.aggregator"}))
+
+	_, err := ParseConfig(cmd)
+	require.NoError(t, err)
+
+	_, err = ParseStartConfig(cmd)
+	require.EqualError(t, err, "DA address is required when aggregator mode is enabled")
+}
+
 func TestPromotableFlagInvariants(t *testing.T) {
 	flagVariants := [][]string{{
 		"--rollkit.node.promotable=false",
@@ -235,7 +253,7 @@ func TestDefaultAggregatorValue(t *testing.T) {
 			// Create a new command without specifying any flags
 			var args []string
 			if tc.expected {
-				args = []string{"start", "--rollkit.node.aggregator"}
+				args = []string{"start", "--rollkit.node.aggregator", "--rollkit.da.address", "http://da.example:7980"}
 			} else {
 				args = []string{"start", "--rollkit.node.aggregator=false"}
 			}

pkg/config/config.go

@@ -270,6 +270,17 @@ type DAConfig struct {
 	BatchMinItems      uint64          `mapstructure:"batch_min_items" yaml:"batch_min_items" comment:"Minimum number of items (headers or data) to accumulate before considering submission. Helps avoid submitting single items when more are expected soon. Default: 1."`
 }
 
+// DAEnabled reports whether this node should initialize a DA client. Full
+// followers without a DA address sync exclusively through P2P.
+func (c Config) DAEnabled() bool {
+	return !c.Node.Light && c.GetDAAddress() != ""
+}
+
+// GetDAAddress returns the configured DA address without surrounding whitespace.
+func (c Config) GetDAAddress() string {
+	return strings.TrimSpace(c.DA.Address)
+}
+
 // GetNamespace returns the namespace for header submissions.
 func (d *DAConfig) GetNamespace() string {
 	return d.Namespace