feat: DA hints in p2p

.mockery.yaml

@@ -49,6 +49,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

@@ -5,12 +5,12 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/evstack/ev-node/types"
 	"github.com/spf13/cobra"
 
 	goheaderstore "github.com/celestiaorg/go-header/store"
 	rollcmd "github.com/evstack/ev-node/pkg/cmd"
 	"github.com/evstack/ev-node/pkg/store"
-	"github.com/evstack/ev-node/types"
 )
 
 // NewRollbackCmd creates a command to rollback ev-node state by one height.
@@ -64,7 +64,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(),
@@ -73,7 +73,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

@@ -5,10 +5,10 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/evstack/ev-node/pkg/sync"
 	"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,7 +20,6 @@ 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/types"
 )
 
 // Components represents the block-related components
@@ -121,8 +120,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,
@@ -152,7 +151,7 @@ func NewSyncComponents(
 	)
 
 	// Create submitter for sync nodes (no signer, only DA inclusion processing)
-	daSubmitter := submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger)
+	daSubmitter := submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger, headerStore, dataStore)
 	submitter := submitting.NewSubmitter(
 		store,
 		exec,
@@ -186,8 +185,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,
@@ -242,7 +241,7 @@ func NewAggregatorComponents(
 		}, nil
 	}
 
-	daSubmitter := submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger)
+	daSubmitter := submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger, headerBroadcaster, dataBroadcaster)
 	submitter := submitting.NewSubmitter(
 		store,
 		exec,

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

@@ -38,8 +38,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
@@ -79,8 +79,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,
@@ -432,8 +432,12 @@ func (e *Executor) produceBlock() error {
 
 	// broadcast header and data to P2P network
 	g, ctx := errgroup.WithContext(e.ctx)
-	g.Go(func() error { return e.headerBroadcaster.WriteToStoreAndBroadcast(ctx, header) })
-	g.Go(func() error { return e.dataBroadcaster.WriteToStoreAndBroadcast(ctx, data) })
+	g.Go(func() error {
+		return e.headerBroadcaster.WriteToStoreAndBroadcast(ctx, &types.P2PSignedHeader{Message: header})
+	})
+	g.Go(func() error {
+		return e.dataBroadcaster.WriteToStoreAndBroadcast(ctx, &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(