support p2p-only followers without DA

tac0turtle · tac0turtle · commit b72b7bd5d954 · 2026-07-15T18:03:03.000+02:00
diff --git a/apps/evm/cmd/init.go b/apps/evm/cmd/init.go
@@ -34,6 +34,9 @@ func InitCmd() *cobra.Command {
 			// we use load in order to parse all the flags
 			cfg, _ := rollconf.Load(cmd)
 			cfg.Node.Aggregator = aggregator
+			if cfg.Node.Aggregator && cfg.DA.Address == "" {
+				cfg.DA.Address = cfg.EffectiveDAAddress()
+			}
 			if err := cfg.Validate(); err != nil {
 				return fmt.Errorf("error validating config: %w", err)
 			}
diff --git a/apps/evm/cmd/run.go b/apps/evm/cmd/run.go
@@ -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.EffectiveDAAddress(), 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)
diff --git a/apps/testapp/cmd/init.go b/apps/testapp/cmd/init.go
@@ -35,6 +35,9 @@ func InitCmd() *cobra.Command {
 			// we use load in order to parse all the flags
 			cfg, _ := rollconf.Load(cmd)
 			cfg.Node.Aggregator = aggregator
+			if cfg.Node.Aggregator && cfg.DA.Address == "" {
+				cfg.DA.Address = cfg.EffectiveDAAddress()
+			}
 			cfg.Node.BlockTime = rollconf.DurationWrapper{Duration: 100 * time.Millisecond}
 			if err := cfg.Validate(); err != nil {
 				return fmt.Errorf("error validating config: %w", err)
diff --git a/apps/testapp/cmd/run.go b/apps/testapp/cmd/run.go
@@ -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.EffectiveDAAddress(), nodeConfig.DA.AuthToken, "")
 	if err != nil {
 		return nil, fmt.Errorf("failed to create blob client: %w", err)
 	}
diff --git a/block/components.go b/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,
diff --git a/block/components_test.go b/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)
diff --git a/block/internal/syncing/syncer.go b/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}:
diff --git a/docs/learn/config.md b/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 retain the legacy `http://localhost:7980` fallback when this value is unset.
 
 **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
diff --git a/docs/learn/specs/da.md b/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; aggregators fall back to `http://localhost:7980`)
 * `--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
diff --git a/docs/reference/specs/da.md b/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; aggregators fall back to `http://localhost:7980`)
 * `--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
diff --git a/pkg/cmd/run_node.go b/pkg/cmd/run_node.go
@@ -150,12 +150,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.EffectiveDAAddress(), 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 {
diff --git a/pkg/config/config.go b/pkg/config/config.go
@@ -270,6 +270,25 @@ 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. Aggregators
+// retain the legacy local DA fallback for backwards compatibility.
+func (c Config) DAEnabled() bool {
+	return !c.Node.Light && (c.Node.Aggregator || c.DA.Address != "")
+}
+
+// EffectiveDAAddress returns the configured DA address, or the legacy local
+// endpoint for aggregators that do not explicitly configure one.
+func (c Config) EffectiveDAAddress() string {
+	if c.DA.Address != "" {
+		return c.DA.Address
+	}
+	if c.Node.Aggregator {
+		return DefaultAggregatorDAAddress
+	}
+	return ""
+}
+
 // GetNamespace returns the namespace for header submissions.
 func (d *DAConfig) GetNamespace() string {
 	return d.Namespace
diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go
@@ -24,7 +24,8 @@ func TestDefaultConfig(t *testing.T) {
 	assert.Equal(t, false, def.Node.Aggregator)
 	assert.Equal(t, false, def.Node.Promotable)
 	assert.Equal(t, false, def.Node.Light)
-	assert.Equal(t, DefaultConfig().DA.Address, def.DA.Address)
+	assert.Empty(t, def.DA.Address)
+	assert.False(t, def.DAEnabled())
 	assert.Equal(t, "", def.DA.AuthToken)
 	assert.Equal(t, "", def.DA.SubmitOptions)
 	assert.NotEmpty(t, def.DA.Namespace)
@@ -42,6 +43,58 @@ func TestDefaultConfig(t *testing.T) {
 	assert.NoError(t, def.Validate())
 }
 
+func TestDAEnabled(t *testing.T) {
+	tests := []struct {
+		name            string
+		configure       func(*Config)
+		expectedEnabled bool
+		expectedAddress string
+	}{
+		{
+			name:            "follower without DA address",
+			expectedEnabled: false,
+			expectedAddress: "",
+		},
+		{
+			name: "follower with DA address",
+			configure: func(cfg *Config) {
+				cfg.DA.Address = "http://da.example:7980"
+			},
+			expectedEnabled: true,
+			expectedAddress: "http://da.example:7980",
+		},
+		{
+			name: "aggregator without DA address uses legacy default",
+			configure: func(cfg *Config) {
+				cfg.Node.Aggregator = true
+			},
+			expectedEnabled: true,
+			expectedAddress: DefaultAggregatorDAAddress,
+		},
+		{
+			name: "light node does not initialize DA",
+			configure: func(cfg *Config) {
+				cfg.Node.Light = true
+				cfg.DA.Address = "http://da.example:7980"
+			},
+			expectedEnabled: false,
+			expectedAddress: "http://da.example:7980",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := DefaultConfig()
+			if tt.configure != nil {
+				tt.configure(&cfg)
+			}
+
+			assert.Equal(t, tt.expectedEnabled, cfg.DAEnabled())
+			assert.Equal(t, tt.expectedAddress, cfg.EffectiveDAAddress())
+		})
+	}
+}
+
 func TestAddFlags(t *testing.T) {
 	// AddFlags and the assertions below each call DefaultConfig(), whose DA.Namespace
 	// is randString-seeded by time.Now().Unix(); run in a synctest bubble so both calls observe the same fake clock.
diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go
