better split + fix da inclusion for p2p

Author: julienrbrt

block/internal/syncing/syncer.go

@@ -243,11 +243,11 @@ func (s *Syncer) syncLoop() {
 		return
 	}
 
-	lastHeaderHeight := initialHeight
-	lastDataHeight := initialHeight
+	lastHeaderHeight := &initialHeight
+	lastDataHeight := &initialHeight
 
 	// Backoff control when DA replies with errors
-	var nextDARequestAt time.Time
+	nextDARequestAt := &time.Time{}
 
 	blockTicker := time.NewTicker(s.config.Node.BlockTime.Duration)
 	defer blockTicker.Stop()
@@ -261,44 +261,27 @@ func (s *Syncer) syncLoop() {
 		// Process pending events from cache on every iteration
 		s.processPendingEvents()
 
-		// Try fetching from both DA and P2P in configurable order
-		if s.tryFetchStrategies(&nextDARequestAt, &lastHeaderHeight, &lastDataHeight, blockTicker.C) {
-			continue // events were processed, restart loop immediately
+		// Fetch events from DA layer first and optimistically p2p when necessary.
+		// This is the default behavior.
+		if !s.config.Sync.PreferP2P {
+			if s.tryFetchFromDA(nextDARequestAt) {
+				continue
+			}
+			if s.tryFetchFromP2P(lastHeaderHeight, lastDataHeight, blockTicker.C) {
+				continue
+			}
+		} else {
+			if s.tryFetchFromP2P(lastHeaderHeight, lastDataHeight, blockTicker.C) {
+				continue
+			}
+			if s.tryFetchFromDA(nextDARequestAt) {
+				continue
+			}
 		}
 
 		// Prevent busy-waiting when no events are available
-		waitTime := min(10*time.Millisecond, s.config.Node.BlockTime.Duration)
-		time.Sleep(waitTime)
-	}
-}
-
-// tryFetchStrategies attempts to fetch from both DA and P2P based on configuration priority.
-// The order of fetching depends on the PreferP2P configuration setting:
-// - PreferP2P=false (default): DA first, then P2P (original and recommended behavior)
-// - PreferP2P=true: P2P first, then DA
-//
-// Both strategies are always attempted on each call, but the order affects which
-// events are processed first and may impact overall sync performance.
-func (s *Syncer) tryFetchStrategies(nextDARequestAt *time.Time, lastHeaderHeight, lastDataHeight *uint64, blockTicker <-chan time.Time) bool {
-	eventsProcessed := false
-
-	if s.config.PreferP2P {
-		if s.tryFetchFromP2P(lastHeaderHeight, lastDataHeight, blockTicker) {
-			eventsProcessed = true
-		}
-		if s.tryFetchFromDA(nextDARequestAt) {
-			eventsProcessed = true
-		}
-	} else {
-		if s.tryFetchFromDA(nextDARequestAt) {
-			eventsProcessed = true
-		}
-		if s.tryFetchFromP2P(lastHeaderHeight, lastDataHeight, blockTicker) {
-			eventsProcessed = true
-		}
+		time.Sleep(min(10*time.Millisecond, s.config.Node.BlockTime.Duration))
 	}
-
-	return eventsProcessed
 }
 
 // tryFetchFromDA attempts to fetch events from the DA layer.
@@ -474,15 +457,17 @@ func (s *Syncer) trySyncNextBlock(event *common.DAHeightEvent) error {
 		return errors.Join(errInvalidBlock, fmt.Errorf("failed to validate block: %w", err))
 	}
 
-	// Mark as DA included
-	headerHash := header.Hash().String()
-	s.cache.SetHeaderDAIncluded(headerHash, event.HeaderDaIncludedHeight)
+	// Mark as DA included (only if set, p2p sync does not set it)
+	if event.HeaderDaIncludedHeight > 0 {
+		headerHash := header.Hash().String()
+		s.cache.SetHeaderDAIncluded(headerHash, event.HeaderDaIncludedHeight)
 
-	s.logger.Info().
-		Str("header_hash", headerHash).
-		Uint64("da_height", event.HeaderDaIncludedHeight).
-		Uint64("height", header.Height()).
-		Msg("header marked as DA included")
+		s.logger.Info().
+			Str("header_hash", headerHash).
+			Uint64("da_height", event.HeaderDaIncludedHeight).
+			Uint64("height", header.Height()).
+			Msg("header marked as DA included")
+	}
 
 	// Apply block
 	newState, err := s.applyBlock(header.Header, data, currentState)

pkg/config/config.go

@@ -28,6 +28,8 @@ const (
 	FlagRootDir = "home"
 	// FlagDBPath is a flag for specifying the database path
 	FlagDBPath = FlagPrefixEvnode + "db_path"
+	// FlagClearCache is a flag for clearing the cache
+	FlagClearCache = FlagPrefixEvnode + "clear_cache"
 
 	// Node configuration flags
 
@@ -47,10 +49,11 @@ const (
 	FlagLazyBlockTime = FlagPrefixEvnode + "node.lazy_block_interval"
 	// FlagReadinessMaxBlocksBehind configures how many blocks behind best-known head is still considered ready
 	FlagReadinessMaxBlocksBehind = FlagPrefixEvnode + "node.readiness_max_blocks_behind"
-	// FlagClearCache is a flag for clearing the cache
-	FlagClearCache = FlagPrefixEvnode + "clear_cache"
-	// FlagP2PPrefer is a flag to prioritize p2p over da fetching
-	FlagP2PPrefer = FlagPrefixEvnode + "prefer_p2p"
+
+	// Sync configuration flags
+
+	// FlagSyncP2PPrefer is a flag to prioritize p2p over da fetching
+	FlagSyncP2PPrefer = FlagPrefixEvnode + "sync.prefer_p2p"
 
 	// Data Availability configuration flags
 
@@ -132,11 +135,14 @@ const (
 // Config stores Rollkit configuration.
 type Config struct {
 	RootDir    string `mapstructure:"-" yaml:"-" comment:"Root directory where rollkit files are located"`
-	ClearCache bool   `mapstructure:"-" yaml:"-" comment:"Clear the cache"`
-	PreferP2P  bool   `mapstructure:"-" yaml:"-" comment:"Prefer P2P over DA fetching"`
+	ClearCache bool   `mapstructure:"clear_cache" yaml:"-" comment:"Clear the cache"`
 
 	// Base configuration
 	DBPath string `mapstructure:"db_path" yaml:"db_path" comment:"Path inside the root directory where the database is located"`
+
+	// Sync configuration (not in yaml)
+	Sync SyncConfig `mapstructure:"sync" yaml:"-"`
+
 	// P2P configuration
 	P2P P2PConfig `mapstructure:"p2p" yaml:"p2p"`
 
@@ -159,6 +165,12 @@ type Config struct {
 	Signer SignerConfig `mapstructure:"signer" yaml:"signer"`
 }
 
+// SyncConfig contains synchronization configuration parameters
+// Those are used in flags are not set in the ev-node config.
+type SyncConfig struct {
+	PreferP2P bool `mapstructure:"prefer_p2p" yaml:"-" comment:"Prefer P2P over DA fetching"`
+}
+
 // DAConfig contains all Data Availability configuration parameters
 type DAConfig struct {
 	Address           string          `mapstructure:"address" yaml:"address" comment:"Address of the data availability layer service (host:port). This is the endpoint where Rollkit will connect to submit and retrieve data."`
@@ -311,7 +323,9 @@ func AddFlags(cmd *cobra.Command) {
 	// Add base flags
 	cmd.Flags().String(FlagDBPath, def.DBPath, "path for the node database")
 	cmd.Flags().Bool(FlagClearCache, def.ClearCache, "clear the cache")
-	cmd.Flags().Bool(FlagP2PPrefer, def.PreferP2P, "prefer P2P over DA fetching")
+
+	// Sync configuration flags
+	cmd.Flags().Bool(FlagSyncP2PPrefer, def.Sync.PreferP2P, "prefer P2P over DA fetching")
 
 	// Node configuration flags
 	cmd.Flags().Bool(FlagAggregator, def.Node.Aggregator, "run node in aggregator mode")
@@ -359,10 +373,8 @@ func AddFlags(cmd *cobra.Command) {
 	cmd.Flags().String(FlagSignerPath, def.Signer.SignerPath, "path to the signer file or address")
 	cmd.Flags().String(FlagSignerPassphrase, "", "passphrase for the signer (required for file signer and if aggregator is enabled)")
 
-	// Only sync nodes can use the FlagP2PPrefer
-	cmd.MarkFlagsMutuallyExclusive(FlagP2PPrefer, FlagAggregator)
-	// Clearing cache on an aggregator is dangerous, should be done manually
-	cmd.MarkFlagsMutuallyExclusive(FlagClearCache, FlagAggregator)
+	// FlagSyncP2PPrefer is only relevant for sync nodes.
+	cmd.MarkFlagsMutuallyExclusive(FlagSyncP2PPrefer, FlagAggregator)
 }
 
 // Load loads the node configuration in the following order of precedence: