feat: p2p exchange wrapper

node/full.go

@@ -87,12 +87,12 @@ func newFullNode(
 	mainKV := newPrefixKV(database, EvPrefix)
 	rktStore := store.New(mainKV)
 
-	headerSyncService, err := initHeaderSyncService(mainKV, nodeConfig, genesis, p2pClient, logger)
+	headerSyncService, err := initHeaderSyncService(mainKV, rktStore, nodeConfig, genesis, p2pClient, logger)
 	if err != nil {
 		return nil, err
 	}
 
-	dataSyncService, err := initDataSyncService(mainKV, nodeConfig, genesis, p2pClient, logger)
+	dataSyncService, err := initDataSyncService(mainKV, rktStore, nodeConfig, genesis, p2pClient, logger)
 	if err != nil {
 		return nil, err
 	}
@@ -149,14 +149,15 @@ func newFullNode(
 
 func initHeaderSyncService(
 	mainKV ds.Batching,
+	daStore store.Store,
 	nodeConfig config.Config,
 	genesis genesispkg.Genesis,
 	p2pClient *p2p.Client,
 	logger zerolog.Logger,
 ) (*evsync.HeaderSyncService, error) {
 	componentLogger := logger.With().Str("component", "HeaderSyncService").Logger()
 
-	headerSyncService, err := evsync.NewHeaderSyncService(mainKV, nodeConfig, genesis, p2pClient, componentLogger)
+	headerSyncService, err := evsync.NewHeaderSyncService(mainKV, daStore, nodeConfig, genesis, p2pClient, componentLogger)
 	if err != nil {
 		return nil, fmt.Errorf("error while initializing HeaderSyncService: %w", err)
 	}
@@ -165,14 +166,15 @@ func initHeaderSyncService(
 
 func initDataSyncService(
 	mainKV ds.Batching,
+	daStore store.Store,
 	nodeConfig config.Config,
 	genesis genesispkg.Genesis,
 	p2pClient *p2p.Client,
 	logger zerolog.Logger,
 ) (*evsync.DataSyncService, error) {
 	componentLogger := logger.With().Str("component", "DataSyncService").Logger()
 
-	dataSyncService, err := evsync.NewDataSyncService(mainKV, nodeConfig, genesis, p2pClient, componentLogger)
+	dataSyncService, err := evsync.NewDataSyncService(mainKV, daStore, nodeConfig, genesis, p2pClient, componentLogger)
 	if err != nil {
 		return nil, fmt.Errorf("error while initializing DataSyncService: %w", err)
 	}

node/light.go

@@ -43,13 +43,13 @@ func newLightNode(
 	logger zerolog.Logger,
 ) (ln *LightNode, err error) {
 	componentLogger := logger.With().Str("component", "HeaderSyncService").Logger()
-	headerSyncService, err := sync.NewHeaderSyncService(database, conf, genesis, p2pClient, componentLogger)
+	store := store.New(database)
+
+	headerSyncService, err := sync.NewHeaderSyncService(database, store, conf, genesis, p2pClient, componentLogger)
 	if err != nil {
 		return nil, fmt.Errorf("error while initializing HeaderSyncService: %w", err)
 	}
 
-	store := store.New(database)
-
 	node := &LightNode{
 		P2P:          p2pClient,
 		hSyncService: headerSyncService,

pkg/sync/exchange_wrapper.go

@@ -0,0 +1,58 @@
+package sync
+
+import (
+	"context"
+
+	"github.com/celestiaorg/go-header"
+	"github.com/evstack/ev-node/pkg/store"
+	"github.com/evstack/ev-node/types"
+)
+
+type exchangeWrapper[H header.Header[H]] struct {
+	header.Exchange[H]
+	daStore store.Store
+}
+
+func (ew *exchangeWrapper[H]) Get(ctx context.Context, hash header.Hash) (H, error) {
+	// Check DA store first
+	var zero H
+	if ew.daStore != nil {
+		switch any(zero).(type) {
+		case *types.SignedHeader:
+			h, _, err := ew.daStore.GetBlockByHash(ctx, hash)
+			if err == nil && h != nil {
+				return any(h).(H), nil
+			}
+		case *types.Data:
+			_, d, err := ew.daStore.GetBlockByHash(ctx, hash)
+			if err == nil && d != nil {
+				return any(d).(H), nil
+			}
+		}
+	}
+
+	// Fallback to network exchange
+	return ew.Exchange.Get(ctx, hash)
+}
+
+func (ew *exchangeWrapper[H]) GetByHeight(ctx context.Context, height uint64) (H, error) {
+	// Check DA store first
+	var zero H
+	if ew.daStore != nil {
+		switch any(zero).(type) {
+		case *types.SignedHeader:
+			h, _, err := ew.daStore.GetBlockData(ctx, height)
+			if err == nil && h != nil {
+				return any(h).(H), nil
+			}
+		case *types.Data:
+			_, d, err := ew.daStore.GetBlockData(ctx, height)
+			if err == nil && d != nil {
+				return any(d).(H), nil
+			}
+		}
+	}
+
+	// Fallback to network exchange
+	return ew.Exchange.GetByHeight(ctx, height)
+}

pkg/sync/sync_service.go

@@ -23,6 +23,7 @@ import (
 	"github.com/evstack/ev-node/pkg/config"
 	"github.com/evstack/ev-node/pkg/genesis"
 	"github.com/evstack/ev-node/pkg/p2p"
+	"github.com/evstack/ev-node/pkg/store"
 	"github.com/evstack/ev-node/types"
 )
 
@@ -48,10 +49,12 @@ type SyncService[H header.Header[H]] struct {
 
 	p2p *p2p.Client
 
-	ex                *goheaderp2p.Exchange[H]
+	ex                header.Exchange[H]
+	p2pExchange       *goheaderp2p.Exchange[H]
 	sub               *goheaderp2p.Subscriber[H]
 	p2pServer         *goheaderp2p.ExchangeServer[H]
 	store             *goheaderstore.Store[H]
+	daStore           store.Store
 	syncer            *goheadersync.Syncer[H]
 	syncerStatus      *SyncerStatus
 	topicSubscription header.Subscription[H]
@@ -66,27 +69,30 @@ type HeaderSyncService = SyncService[*types.SignedHeader]
 // NewDataSyncService returns a new DataSyncService.
 func NewDataSyncService(
 	store ds.Batching,
+	daStore store.Store,
 	conf config.Config,
 	genesis genesis.Genesis,
 	p2p *p2p.Client,
 	logger zerolog.Logger,
 ) (*DataSyncService, error) {
-	return newSyncService[*types.Data](store, dataSync, conf, genesis, p2p, logger)
+	return newSyncService[*types.Data](store, daStore, dataSync, conf, genesis, p2p, logger)
 }
 
 // NewHeaderSyncService returns a new HeaderSyncService.
 func NewHeaderSyncService(
 	store ds.Batching,
+	daStore store.Store,
 	conf config.Config,
 	genesis genesis.Genesis,
 	p2p *p2p.Client,
 	logger zerolog.Logger,
 ) (*HeaderSyncService, error) {
-	return newSyncService[*types.SignedHeader](store, headerSync, conf, genesis, p2p, logger)
+	return newSyncService[*types.SignedHeader](store, daStore, headerSync, conf, genesis, p2p, logger)
 }
 
 func newSyncService[H header.Header[H]](
-	store ds.Batching,
+	dsStore ds.Batching,
+	daStore store.Store,
 	syncType syncType,
 	conf config.Config,
 	genesis genesis.Genesis,
@@ -98,7 +104,7 @@ func newSyncService[H header.Header[H]](
 	}
 
 	ss, err := goheaderstore.NewStore[H](
-		store,
+		dsStore,
 		goheaderstore.WithStorePrefix(string(syncType)),
 		goheaderstore.WithMetrics(),
 	)
@@ -111,6 +117,7 @@ func newSyncService[H header.Header[H]](
 		genesis:      genesis,
 		p2p:          p2p,
 		store:        ss,
+		daStore:      daStore,
 		syncType:     syncType,
 		logger:       logger,
 		syncerStatus: new(SyncerStatus),
@@ -276,12 +283,19 @@ func (syncService *SyncService[H]) setupP2PInfrastructure(ctx context.Context) (
 
 	peerIDs := syncService.getPeerIDs()
 
-	if syncService.ex, err = newP2PExchange[H](syncService.p2p.Host(), peerIDs, networkID, syncService.genesis.ChainID, syncService.p2p.ConnectionGater()); err != nil {
+	if syncService.p2pExchange, err = newP2PExchange[H](syncService.p2p.Host(), peerIDs, networkID, syncService.genesis.ChainID, syncService.p2p.ConnectionGater()); err != nil {
 		return nil, fmt.Errorf("error while creating exchange: %w", err)
 	}
-	if err := syncService.ex.Start(ctx); err != nil {
+	if err := syncService.p2pExchange.Start(ctx); err != nil {
 		return nil, fmt.Errorf("error while starting exchange: %w", err)
 	}
+
+	// Wrap the exchange with the DA store check
+	syncService.ex = &exchangeWrapper[H]{
+		Exchange: syncService.p2pExchange,
+		daStore:  syncService.daStore,
+	}
+
 	return peerIDs, nil
 }
 
@@ -375,7 +389,7 @@ func (syncService *SyncService[H]) Stop(ctx context.Context) error {
 	syncService.topicSubscription.Cancel()
 	err := errors.Join(
 		syncService.p2pServer.Stop(ctx),
-		syncService.ex.Stop(ctx),
+		syncService.p2pExchange.Stop(ctx),
 		syncService.sub.Stop(ctx),
 	)
 	if syncService.syncerStatus.isStarted() {

pkg/sync/sync_service_test.go

@@ -21,6 +21,8 @@ import (
 	mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/require"
+
+	"github.com/evstack/ev-node/pkg/store"
 )
 
 func TestHeaderSyncServiceRestart(t *testing.T) {
@@ -58,7 +60,8 @@ func TestHeaderSyncServiceRestart(t *testing.T) {
 	defer cancel()
 	require.NoError(t, p2pClient.Start(ctx))
 
-	svc, err := NewHeaderSyncService(mainKV, conf, genesisDoc, p2pClient, logger)
+	rktStore := store.New(mainKV)
+	svc, err := NewHeaderSyncService(mainKV, rktStore, conf, genesisDoc, p2pClient, logger)
 	require.NoError(t, err)
 	err = svc.Start(ctx)
 	require.NoError(t, err)
@@ -98,7 +101,7 @@ func TestHeaderSyncServiceRestart(t *testing.T) {
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = p2pClient.Close() })
 
-	svc, err = NewHeaderSyncService(mainKV, conf, genesisDoc, p2pClient, logger)
+	svc, err = NewHeaderSyncService(mainKV, rktStore, conf, genesisDoc, p2pClient, logger)
 	require.NoError(t, err)
 	err = svc.Start(ctx)
 	require.NoError(t, err)