feat: DA hints in p2p

.mockery.yaml

@@ -61,6 +61,7 @@ packages:
           dir: ./block/internal/syncing
           pkgname: syncing
           filename: syncer_mock.go
+
   github.com/evstack/ev-node/block/internal/common:
     interfaces:
       Broadcaster:

apps/evm/go.mod

@@ -2,10 +2,10 @@ module github.com/evstack/ev-node/apps/evm
 
 go 1.25.0
 
-//replace (
-//	github.com/evstack/ev-node => ../../
-//	github.com/evstack/ev-node/execution/evm => ../../execution/evm
-//)
+replace (
+	github.com/evstack/ev-node => ../../
+	github.com/evstack/ev-node/execution/evm => ../../execution/evm
+)
 
 require (
 	github.com/ethereum/go-ethereum v1.16.8

apps/evm/go.sum

@@ -409,12 +409,8 @@ github.com/ethereum/go-ethereum v1.16.8 h1:LLLfkZWijhR5m6yrAXbdlTeXoqontH+Ga2f9i
 github.com/ethereum/go-ethereum v1.16.8/go.mod h1:Fs6QebQbavneQTYcA39PEKv2+zIjX7rPUZ14DER46wk=
 github.com/ethereum/go-verkle v0.2.2 h1:I2W0WjnrFUIzzVPwm8ykY+7pL2d4VhlsePn4j7cnFk8=
 github.com/ethereum/go-verkle v0.2.2/go.mod h1:M3b90YRnzqKyyzBEWJGqj8Qff4IDeXnzFw0P9bFw3uk=
-github.com/evstack/ev-node v1.0.0-rc.2 h1:gUQzLTkCj6D751exm/FIR/yw2aXWiW2aEREEwtxMvw0=
-github.com/evstack/ev-node v1.0.0-rc.2/go.mod h1:Qa2nN1D6PJQRU2tiarv6X5Der5OZg/+2QGY/K2mA760=
 github.com/evstack/ev-node/core v1.0.0-rc.1 h1:Dic2PMUMAYUl5JW6DkDj6HXDEWYzorVJQuuUJOV0FjE=
 github.com/evstack/ev-node/core v1.0.0-rc.1/go.mod h1:n2w/LhYQTPsi48m6lMj16YiIqsaQw6gxwjyJvR+B3sY=
-github.com/evstack/ev-node/execution/evm v1.0.0-rc.2 h1:t7os7ksmPhf2rWY2psVBowyc+iuneMDPwBGQaxSckus=
-github.com/evstack/ev-node/execution/evm v1.0.0-rc.2/go.mod h1:ahxKQfPlJ5C7g15Eq9Mjn2tQnn59T0kIm9B10zDhcTI=
 github.com/fatih/color v1.10.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGEBuJM=
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
 github.com/fatih/color v1.15.0/go.mod h1:0h5ZqXfHYED7Bhv2ZJamyIOUej9KtShiJESRwBDUSsw=

apps/grpc/go.mod

@@ -2,10 +2,10 @@ module github.com/evstack/ev-node/apps/grpc
 
 go 1.25.0
 
-//replace (
-//	github.com/evstack/ev-node => ../../
-//	github.com/evstack/ev-node/execution/grpc => ../../execution/grpc
-//)
+replace (
+	github.com/evstack/ev-node => ../../
+	github.com/evstack/ev-node/execution/grpc => ../../execution/grpc
+)
 
 require (
 	github.com/evstack/ev-node v1.0.0-rc.2

apps/grpc/go.sum

@@ -365,12 +365,8 @@ github.com/envoyproxy/protoc-gen-validate v0.10.0/go.mod h1:DRjgyB0I43LtJapqN6Ni
 github.com/envoyproxy/protoc-gen-validate v0.10.1/go.mod h1:DRjgyB0I43LtJapqN6NiRwroiAU2PaFuvk/vjgh61ss=
 github.com/envoyproxy/protoc-gen-validate v1.0.1/go.mod h1:0vj8bNkYbSTNS2PIyH87KZaeN4x9zpL9Qt8fQC7d+vs=
 github.com/envoyproxy/protoc-gen-validate v1.0.2/go.mod h1:GpiZQP3dDbg4JouG/NNS7QWXpgx6x8QiMKdmN72jogE=
-github.com/evstack/ev-node v1.0.0-rc.2 h1:gUQzLTkCj6D751exm/FIR/yw2aXWiW2aEREEwtxMvw0=
-github.com/evstack/ev-node v1.0.0-rc.2/go.mod h1:Qa2nN1D6PJQRU2tiarv6X5Der5OZg/+2QGY/K2mA760=
 github.com/evstack/ev-node/core v1.0.0-rc.1 h1:Dic2PMUMAYUl5JW6DkDj6HXDEWYzorVJQuuUJOV0FjE=
 github.com/evstack/ev-node/core v1.0.0-rc.1/go.mod h1:n2w/LhYQTPsi48m6lMj16YiIqsaQw6gxwjyJvR+B3sY=
-github.com/evstack/ev-node/execution/grpc v1.0.0-rc.1 h1:OzrWLDDY6/9+LWx0XmUqPzxs/CHZRJICOwQ0Me/i6dY=
-github.com/evstack/ev-node/execution/grpc v1.0.0-rc.1/go.mod h1:Pr/sF6Zx8am9ZeWFcoz1jYPs0kXmf+OmL8Tz2Gyq7E4=
 github.com/fatih/color v1.10.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGEBuJM=
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
 github.com/fatih/color v1.15.0 h1:kOqh6YHBtK8aywxGerMG2Eq3H6Qgoqeo13Bk2Mv/nBs=

block/components.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/celestiaorg/go-header"
 	"github.com/rs/zerolog"
 
 	"github.com/evstack/ev-node/block/internal/cache"
@@ -20,6 +21,7 @@ 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"
 )
@@ -127,8 +129,10 @@ func NewSyncComponents(
 	store store.Store,
 	exec coreexecutor.Executor,
 	daClient da.Client,
-	headerStore common.Broadcaster[*types.SignedHeader],
-	dataStore common.Broadcaster[*types.Data],
+	headerStore header.Store[*types.P2PSignedHeader],
+	dataStore header.Store[*types.P2PData],
+	headerDAHintAppender submitting.DAHintAppender,
+	dataDAHintAppender submitting.DAHintAppender,
 	logger zerolog.Logger,
 	metrics *Metrics,
 	blockOpts BlockOptions,
@@ -163,7 +167,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, headerDAHintAppender, dataDAHintAppender)
 	if config.Instrumentation.IsTracingEnabled() {
 		daSubmitter = submitting.WithTracingDASubmitter(daSubmitter)
 	}
@@ -200,8 +204,8 @@ func NewAggregatorComponents(
 	sequencer coresequencer.Sequencer,
 	daClient da.Client,
 	signer signer.Signer,
-	headerBroadcaster common.Broadcaster[*types.SignedHeader],
-	dataBroadcaster common.Broadcaster[*types.Data],
+	headerSyncService *sync.HeaderSyncService,
+	dataSyncService *sync.DataSyncService,
 	logger zerolog.Logger,
 	metrics *Metrics,
 	blockOpts BlockOptions,
@@ -229,8 +233,8 @@ func NewAggregatorComponents(
 		metrics,
 		config,
 		genesis,
-		headerBroadcaster,
-		dataBroadcaster,
+		headerSyncService,
+		dataSyncService,
 		logger,
 		blockOpts,
 		errorCh,
@@ -266,7 +270,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, headerSyncService, dataSyncService)
 	if config.Instrumentation.IsTracingEnabled() {
 		daSubmitter = submitting.WithTracingDASubmitter(daSubmitter)
 	}

block/components_test.go

@@ -22,8 +22,17 @@ import (
 	"github.com/evstack/ev-node/pkg/signer/noop"
 	"github.com/evstack/ev-node/pkg/store"
 	testmocks "github.com/evstack/ev-node/test/mocks"
+	extmocks "github.com/evstack/ev-node/test/mocks/external"
+	"github.com/evstack/ev-node/types"
 )
 
+// noopDAHintAppender is a no-op implementation of DAHintAppender for testing
+type noopDAHintAppender struct{}
+
+func (n noopDAHintAppender) AppendDAHint(ctx context.Context, daHeight uint64, heights ...uint64) error {
+	return nil
+}
+
 func TestBlockComponents_ExecutionClientFailure_StopsNode(t *testing.T) {
 	// Test the error channel mechanism works as intended
 
@@ -86,6 +95,14 @@ func TestNewSyncComponents_Creation(t *testing.T) {
 	daClient.On("GetForcedInclusionNamespace").Return([]byte(nil)).Maybe()
 	daClient.On("HasForcedInclusionNamespace").Return(false).Maybe()
 
+	// Create mock P2P stores
+	mockHeaderStore := extmocks.NewMockStore[*types.P2PSignedHeader](t)
+	mockDataStore := extmocks.NewMockStore[*types.P2PData](t)
+
+	// Create noop DAHintAppenders for testing
+	headerHintAppender := noopDAHintAppender{}
+	dataHintAppender := noopDAHintAppender{}
+
 	// Just test that the constructor doesn't panic - don't start the components
 	// to avoid P2P store dependencies
 	components, err := NewSyncComponents(
@@ -94,8 +111,10 @@ func TestNewSyncComponents_Creation(t *testing.T) {
 		memStore,
 		mockExec,
 		daClient,
-		nil,
-		nil,
+		mockHeaderStore,
+		mockDataStore,
+		headerHintAppender,
+		dataHintAppender,
 		zerolog.Nop(),
 		NopMetrics(),
 		DefaultBlockOptions(),

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,11 +3,17 @@ package common
 import (
 	"context"
 
+	"github.com/evstack/ev-node/types"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 
 	"github.com/celestiaorg/go-header"
 )
 
+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

block/internal/executing/executor.go

@@ -42,8 +42,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
@@ -90,8 +90,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,
@@ -547,9 +547,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{SignedHeader: header})
+	})
+	g.Go(func() error {
+		return e.dataBroadcaster.WriteToStoreAndBroadcast(broadcastCtx, &types.P2PData{Data: 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(
@@ -163,9 +163,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(

block/internal/executing/executor_logic_test.go

@@ -69,9 +69,9 @@ func TestProduceBlock_EmptyBatch_SetsEmptyDataHash(t *testing.T) {
 	mockSeq := testmocks.NewMockSequencer(t)
 
 	// Broadcasters are required by produceBlock; use generated mocks
-	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(
@@ -160,9 +160,9 @@ func TestPendingLimit_SkipsProduction(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(

block/internal/executing/executor_restart_test.go

@@ -48,9 +48,9 @@ func TestExecutor_RestartUsesPendingHeader(t *testing.T) {
 	// Create first executor instance
 	mockExec1 := testmocks.NewMockExecutor(t)
 	mockSeq1 := testmocks.NewMockSequencer(t)
-	hb1 := common.NewMockBroadcaster[*types.SignedHeader](t)
+	hb1 := common.NewMockBroadcaster[*types.P2PSignedHeader](t)
 	hb1.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
-	db1 := common.NewMockBroadcaster[*types.Data](t)
+	db1 := common.NewMockBroadcaster[*types.P2PData](t)
 	db1.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
 
 	exec1, err := NewExecutor(
@@ -171,9 +171,9 @@ func TestExecutor_RestartUsesPendingHeader(t *testing.T) {
 	// Create second executor instance (restart scenario)
 	mockExec2 := testmocks.NewMockExecutor(t)
 	mockSeq2 := testmocks.NewMockSequencer(t)
-	hb2 := common.NewMockBroadcaster[*types.SignedHeader](t)
+	hb2 := common.NewMockBroadcaster[*types.P2PSignedHeader](t)
 	hb2.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
-	db2 := common.NewMockBroadcaster[*types.Data](t)
+	db2 := common.NewMockBroadcaster[*types.P2PData](t)
 	db2.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
 
 	exec2, err := NewExecutor(
@@ -275,9 +275,9 @@ func TestExecutor_RestartNoPendingHeader(t *testing.T) {
 	// Create first executor and produce one block
 	mockExec1 := testmocks.NewMockExecutor(t)
 	mockSeq1 := testmocks.NewMockSequencer(t)
-	hb1 := common.NewMockBroadcaster[*types.SignedHeader](t)
+	hb1 := common.NewMockBroadcaster[*types.P2PSignedHeader](t)
 	hb1.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
-	db1 := common.NewMockBroadcaster[*types.Data](t)
+	db1 := common.NewMockBroadcaster[*types.P2PData](t)
 	db1.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
 
 	exec1, err := NewExecutor(
@@ -336,9 +336,9 @@ func TestExecutor_RestartNoPendingHeader(t *testing.T) {
 	// Create second executor (restart)
 	mockExec2 := testmocks.NewMockExecutor(t)
 	mockSeq2 := testmocks.NewMockSequencer(t)
-	hb2 := common.NewMockBroadcaster[*types.SignedHeader](t)
+	hb2 := common.NewMockBroadcaster[*types.P2PSignedHeader](t)
 	hb2.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
-	db2 := common.NewMockBroadcaster[*types.Data](t)
+	db2 := common.NewMockBroadcaster[*types.P2PData](t)
 	db2.EXPECT().WriteToStoreAndBroadcast(mock.Anything, mock.Anything).Return(nil).Maybe()
 
 	exec2, err := NewExecutor(