x

Author: alpe

block/components.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/evstack/ev-node/pkg/sync"
 	"github.com/rs/zerolog"
 
 	"github.com/evstack/ev-node/block/internal/cache"
@@ -132,8 +133,8 @@ func NewSyncComponents(
 	store store.Store,
 	exec coreexecutor.Executor,
 	da coreda.DA,
-	headerStore common.Broadcaster[*types.SignedHeaderWithDAHint],
-	dataStore common.Broadcaster[*types.Data],
+	headerStore *sync.HeaderSyncService,
+	dataStore *sync.DataSyncService,
 	logger zerolog.Logger,
 	metrics *Metrics,
 	blockOpts BlockOptions,
@@ -157,15 +158,15 @@ func NewSyncComponents(
 		metrics,
 		config,
 		genesis,
-		headerStore,
-		dataStore,
+		common.NewDecorator[*types.SignedHeader](headerStore),
+		common.NewDecorator[*types.Data](dataStore),
 		logger,
 		blockOpts,
 		errorCh,
 	)
 
 	// Create submitter for sync nodes (no signer, only DA inclusion processing)
-	daSubmitter := submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger, nil) // todo (Alex): use a noop
+	daSubmitter := submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger, headerStore, dataStore)
 	submitter := submitting.NewSubmitter(
 		store,
 		exec,
@@ -198,8 +199,8 @@ func NewAggregatorComponents(
 	sequencer coresequencer.Sequencer,
 	da coreda.DA,
 	signer signer.Signer,
-	headerBroadcaster common.Broadcaster[*types.SignedHeaderWithDAHint],
-	dataBroadcaster common.Broadcaster[*types.Data],
+	headerBroadcaster *sync.HeaderSyncService,
+	dataBroadcaster *sync.DataSyncService,
 	logger zerolog.Logger,
 	metrics *Metrics,
 	blockOpts BlockOptions,
@@ -222,8 +223,8 @@ func NewAggregatorComponents(
 		metrics,
 		config,
 		genesis,
-		headerBroadcaster,
-		dataBroadcaster,
+		common.NewDecorator[*types.SignedHeader](headerBroadcaster),
+		common.NewDecorator[*types.Data](dataBroadcaster),
 		logger,
 		blockOpts,
 		errorCh,
@@ -247,7 +248,7 @@ func NewAggregatorComponents(
 
 	// Create DA client and submitter for aggregator nodes (with signer for submission)
 	daClient := NewDAClient(da, config, logger)
-	daSubmitter := submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger, headerBroadcaster)
+	daSubmitter := submitting.NewDASubmitter(daClient, config, genesis, blockOpts, metrics, logger, headerBroadcaster, dataBroadcaster)
 	submitter := submitting.NewSubmitter(
 		store,
 		exec,

block/internal/common/broadcaster_mock.go

@@ -21,6 +21,6 @@ type DAHeightEvent struct {
 	// Source indicates where this event originated from (DA or P2P)
 	Source EventSource
 
-	// Optional DA height hint from P2P
-	DaHeightHint uint64
+	// 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/event.go

@@ -3,14 +3,63 @@ 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 = Decorator[*types.SignedHeader]
+	DataP2PBroadcaster   = Decorator[*types.Data]
+)
+
+// 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]
-	XXX(ctx context.Context, headerOrData H) error
+	AppendDAHint(ctx context.Context, daHeight uint64, hashes ...types.Hash) error
+}
+
+// Decorator to access the the payload type without the container
+type Decorator[H header.Header[H]] struct {
+	nested Broadcaster[*types.DAHeightHintContainer[H]]
+}
+
+func NewDecorator[H header.Header[H]](nested Broadcaster[*types.DAHeightHintContainer[H]]) Decorator[H] {
+	return Decorator[H]{nested: nested}
+}
+
+func (d Decorator[H]) WriteToStoreAndBroadcast(ctx context.Context, payload H, opts ...pubsub.PubOpt) error {
+	return d.nested.WriteToStoreAndBroadcast(ctx, &types.DAHeightHintContainer[H]{Entry: payload}, opts...)
+}
+
+func (d Decorator[H]) Store() HeightStore[H] {
+	return HeightStoreImpl[H]{store: d.nested.Store()}
+}
+func (d Decorator[H]) XStore() header.Store[*types.DAHeightHintContainer[H]] {
+	return d.nested.Store()
+}
+
+func (d Decorator[H]) AppendDAHint(ctx context.Context, daHeight uint64, hashes ...types.Hash) error {
+	return d.nested.AppendDAHint(ctx, daHeight, hashes...)
+}
+
+// HeightStore is a subset of goheader.Store
+type HeightStore[H header.Header[H]] interface {
+	GetByHeight(context.Context, uint64) (H, error)
+}
+
+type HeightStoreImpl[H header.Header[H]] struct {
+	store header.Store[*types.DAHeightHintContainer[H]]
+}
+
+func (s HeightStoreImpl[H]) GetByHeight(ctx context.Context, height uint64) (H, error) {
+	var zero H
+	v, err := s.store.GetByHeight(ctx, height)
+	if err != nil {
+		return zero, err
+	}
+	return v.Entry, nil
+
 }

block/internal/common/expected_interfaces.go

@@ -37,8 +37,8 @@ type Executor struct {
 	metrics *common.Metrics
 
 	// Broadcasting
-	headerBroadcaster common.Broadcaster[*types.SignedHeaderWithDAHint]
-	dataBroadcaster   common.Broadcaster[*types.Data]
+	headerBroadcaster common.HeaderP2PBroadcaster
+	dataBroadcaster   common.DataP2PBroadcaster
 
 	// Configuration
 	config  config.Config
@@ -76,8 +76,8 @@ func NewExecutor(
 	metrics *common.Metrics,
 	config config.Config,
 	genesis genesis.Genesis,
-	headerBroadcaster common.Broadcaster[*types.SignedHeaderWithDAHint],
-	dataBroadcaster common.Broadcaster[*types.Data],
+	headerBroadcaster common.HeaderP2PBroadcaster,
+	dataBroadcaster common.DataP2PBroadcaster,
 	logger zerolog.Logger,
 	options common.BlockOptions,
 	errorCh chan<- error,
@@ -420,7 +420,7 @@ 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, &types.SignedHeaderWithDAHint{SignedHeader: header}) })
+	g.Go(func() error { return e.headerBroadcaster.WriteToStoreAndBroadcast(ctx, header) })
 	g.Go(func() error { return e.dataBroadcaster.WriteToStoreAndBroadcast(ctx, data) })
 	if err := g.Wait(); err != nil {
 		e.logger.Error().Err(err).Msg("failed to broadcast header and/data")