feat: DA hints in p2p

.mockery.yaml

@@ -66,6 +66,11 @@ packages:
           dir: ./block/internal/syncing
           pkgname: syncing
           filename: syncer_mock.go
+      HeightStore:
+        config:
+          dir: ./block/internal/syncing
+          pkgname: syncing
+          filename: height_store_mock.go
   github.com/evstack/ev-node/block/internal/common:
     interfaces:
       Broadcaster:

apps/evm/cmd/rollback.go

@@ -80,7 +80,7 @@ func NewRollbackCmd() *cobra.Command {
 			}
 
 			// rollback ev-node goheader state
-			headerStore, err := goheaderstore.NewStore[*types.SignedHeader](
+			headerStore, err := goheaderstore.NewStore[*types.P2PSignedHeader](
 				evolveDB,
 				goheaderstore.WithStorePrefix("headerSync"),
 				goheaderstore.WithMetrics(),
@@ -89,7 +89,7 @@ func NewRollbackCmd() *cobra.Command {
 				return err
 			}
 
-			dataStore, err := goheaderstore.NewStore[*types.Data](
+			dataStore, err := goheaderstore.NewStore[*types.P2PData](
 				evolveDB,
 				goheaderstore.WithStorePrefix("dataSync"),
 				goheaderstore.WithMetrics(),

apps/testapp/cmd/rollback.go

@@ -70,7 +70,7 @@ func NewRollbackCmd() *cobra.Command {
 			}
 
 			// rollback ev-node goheader state
-			headerStore, err := goheaderstore.NewStore[*types.SignedHeader](
+			headerStore, err := goheaderstore.NewStore[*types.P2PSignedHeader](
 				evolveDB,
 				goheaderstore.WithStorePrefix("headerSync"),
 				goheaderstore.WithMetrics(),
@@ -79,7 +79,7 @@ func NewRollbackCmd() *cobra.Command {
 				return err
 			}
 
-			dataStore, err := goheaderstore.NewStore[*types.Data](
+			dataStore, err := goheaderstore.NewStore[*types.P2PData](
 				evolveDB,
 				goheaderstore.WithStorePrefix("dataSync"),
 				goheaderstore.WithMetrics(),

block/components.go

@@ -8,7 +8,6 @@ import (
 	"github.com/rs/zerolog"
 
 	"github.com/evstack/ev-node/block/internal/cache"
-	"github.com/evstack/ev-node/block/internal/common"
 	da "github.com/evstack/ev-node/block/internal/da"
 	"github.com/evstack/ev-node/block/internal/executing"
 	"github.com/evstack/ev-node/block/internal/reaping"
@@ -20,8 +19,8 @@ import (
 	"github.com/evstack/ev-node/pkg/genesis"
 	"github.com/evstack/ev-node/pkg/signer"
 	"github.com/evstack/ev-node/pkg/store"
+	"github.com/evstack/ev-node/pkg/sync"
 	"github.com/evstack/ev-node/pkg/telemetry"
-	"github.com/evstack/ev-node/types"
 )
 
 // Components represents the block-related components
@@ -122,8 +121,8 @@ func NewSyncComponents(
 	store store.Store,
 	exec coreexecutor.Executor,
 	daClient da.Client,
-	headerStore common.Broadcaster[*types.SignedHeader],
-	dataStore common.Broadcaster[*types.Data],
+	headerStore *sync.HeaderSyncService,
+	dataStore *sync.DataSyncService,
 	logger zerolog.Logger,
 	metrics *Metrics,
 	blockOpts BlockOptions,
@@ -157,7 +156,7 @@ func NewSyncComponents(
 	}
 
 	// Create submitter for sync nodes (no signer, only DA inclusion processing)
-	var daSubmitter submitting.DASubmitterAPI = submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger)
+	var daSubmitter submitting.DASubmitterAPI = submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger, headerStore, dataStore)
 	if config.Instrumentation.IsTracingEnabled() {
 		daSubmitter = submitting.WithTracingDASubmitter(daSubmitter)
 	}
@@ -194,8 +193,8 @@ func NewAggregatorComponents(
 	sequencer coresequencer.Sequencer,
 	daClient da.Client,
 	signer signer.Signer,
-	headerBroadcaster common.Broadcaster[*types.SignedHeader],
-	dataBroadcaster common.Broadcaster[*types.Data],
+	headerBroadcaster *sync.HeaderSyncService,
+	dataBroadcaster *sync.DataSyncService,
 	logger zerolog.Logger,
 	metrics *Metrics,
 	blockOpts BlockOptions,
@@ -259,7 +258,7 @@ func NewAggregatorComponents(
 		}, nil
 	}
 
-	var daSubmitter submitting.DASubmitterAPI = submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger)
+	var daSubmitter submitting.DASubmitterAPI = submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger, headerBroadcaster, dataBroadcaster)
 	if config.Instrumentation.IsTracingEnabled() {
 		daSubmitter = submitting.WithTracingDASubmitter(daSubmitter)
 	}

block/internal/common/event.go

@@ -20,4 +20,7 @@ type DAHeightEvent struct {
 	DaHeight uint64
 	// Source indicates where this event originated from (DA or P2P)
 	Source EventSource
+
+	// Optional DA height hints from P2P. first is the DA height hint for the header, second is the DA height hint for the data
+	DaHeightHints [2]uint64
 }

block/internal/common/expected_interfaces.go

@@ -3,13 +3,20 @@ package common
 import (
 	"context"
 
+	"github.com/evstack/ev-node/types"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 
 	"github.com/celestiaorg/go-header"
 )
 
-// broadcaster interface for P2P broadcasting
+type (
+	HeaderP2PBroadcaster = Broadcaster[*types.P2PSignedHeader]
+	DataP2PBroadcaster   = Broadcaster[*types.P2PData]
+)
+
+// Broadcaster interface for P2P broadcasting
 type Broadcaster[H header.Header[H]] interface {
 	WriteToStoreAndBroadcast(ctx context.Context, payload H, opts ...pubsub.PubOpt) error
-	Store() header.Store[H]
+	AppendDAHint(ctx context.Context, daHeight uint64, hashes ...types.Hash) error
+	GetByHeight(ctx context.Context, height uint64) (H, uint64, error)
 }

block/internal/executing/executor.go

@@ -40,8 +40,8 @@ type Executor struct {
 	metrics *common.Metrics
 
 	// Broadcasting
-	headerBroadcaster common.Broadcaster[*types.SignedHeader]
-	dataBroadcaster   common.Broadcaster[*types.Data]
+	headerBroadcaster common.HeaderP2PBroadcaster
+	dataBroadcaster   common.DataP2PBroadcaster
 
 	// Configuration
 	config  config.Config
@@ -85,8 +85,8 @@ func NewExecutor(
 	metrics *common.Metrics,
 	config config.Config,
 	genesis genesis.Genesis,
-	headerBroadcaster common.Broadcaster[*types.SignedHeader],
-	dataBroadcaster common.Broadcaster[*types.Data],
+	headerBroadcaster common.HeaderP2PBroadcaster,
+	dataBroadcaster common.DataP2PBroadcaster,
 	logger zerolog.Logger,
 	options common.BlockOptions,
 	errorCh chan<- error,
@@ -453,9 +453,13 @@ func (e *Executor) ProduceBlock(ctx context.Context) error {
 	e.setLastState(newState)
 
 	// broadcast header and data to P2P network
-	g, broadcastCtx := errgroup.WithContext(ctx)
-	g.Go(func() error { return e.headerBroadcaster.WriteToStoreAndBroadcast(broadcastCtx, header) })
-	g.Go(func() error { return e.dataBroadcaster.WriteToStoreAndBroadcast(broadcastCtx, data) })
+	g, broadcastCtx := errgroup.WithContext(e.ctx)
+	g.Go(func() error {
+		return e.headerBroadcaster.WriteToStoreAndBroadcast(broadcastCtx, &types.P2PSignedHeader{Message: header})
+	})
+	g.Go(func() error {
+		return e.dataBroadcaster.WriteToStoreAndBroadcast(broadcastCtx, &types.P2PData{Message: data})
+	})
 	if err := g.Wait(); err != nil {
 		e.logger.Error().Err(err).Msg("failed to broadcast header and/data")
 		// don't fail block production on broadcast error

block/internal/executing/executor_lazy_test.go

@@ -47,9 +47,9 @@ func TestLazyMode_ProduceBlockLogic(t *testing.T) {
 
 	mockExec := testmocks.NewMockExecutor(t)
 	mockSeq := testmocks.NewMockSequencer(t)
-	hb := common.NewMockBroadcaster[*types.SignedHeader](t)
+	hb := common.NewMockBroadcaster[*types.P2PSignedHeader](t)
 	hb.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
-	db := common.NewMockBroadcaster[*types.Data](t)
+	db := common.NewMockBroadcaster[*types.P2PData](t)
 	db.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
 
 	exec, err := NewExecutor(
@@ -162,9 +162,9 @@ func TestRegularMode_ProduceBlockLogic(t *testing.T) {
 
 	mockExec := testmocks.NewMockExecutor(t)
 	mockSeq := testmocks.NewMockSequencer(t)
-	hb := common.NewMockBroadcaster[*types.SignedHeader](t)
+	hb := common.NewMockBroadcaster[*types.P2PSignedHeader](t)
 	hb.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
-	db := common.NewMockBroadcaster[*types.Data](t)
+	db := common.NewMockBroadcaster[*types.P2PData](t)
 	db.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
 
 	exec, err := NewExecutor(