feat: p2p exchange wrapper

.mockery.yaml

@@ -41,6 +41,11 @@ packages:
           filename: store.go
   github.com/celestiaorg/go-header:
     interfaces:
+      Exchange:
+        config:
+          dir: ./test/mocks
+          pkgname: mocks
+          filename: external/hexchange.go
       Store:
         config:
           dir: ./test/mocks

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,42 @@
+package sync
+
+import (
+	"context"
+
+	"github.com/celestiaorg/go-header"
+	"github.com/evstack/ev-node/pkg/store"
+)
+
+type storeGetter[H header.Header[H]] func(context.Context, store.Store, header.Hash) (H, error)
+type storeGetterByHeight[H header.Header[H]] func(context.Context, store.Store, uint64) (H, error)
+
+type exchangeWrapper[H header.Header[H]] struct {
+	header.Exchange[H]
+	daStore        store.Store
+	getter         storeGetter[H]
+	getterByHeight storeGetterByHeight[H]
+}
+
+func (ew *exchangeWrapper[H]) Get(ctx context.Context, hash header.Hash) (H, error) {
+	// Check DA store first
+	if ew.daStore != nil && ew.getter != nil {
+		if h, err := ew.getter(ctx, ew.daStore, hash); err == nil && !h.IsZero() {
+			return 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
+	if ew.daStore != nil && ew.getterByHeight != nil {
+		if h, err := ew.getterByHeight(ctx, ew.daStore, height); err == nil && !h.IsZero() {
+			return h, nil
+		}
+	}
+
+	// Fallback to network exchange
+	return ew.Exchange.GetByHeight(ctx, height)
+}

pkg/sync/exchange_wrapper_test.go

@@ -0,0 +1,116 @@
+package sync
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/celestiaorg/go-header"
+	"github.com/evstack/ev-node/pkg/store"
+	"github.com/evstack/ev-node/test/mocks"
+	extmocks "github.com/evstack/ev-node/test/mocks/external"
+	"github.com/evstack/ev-node/types"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestExchangeWrapper_Get(t *testing.T) {
+	ctx := context.Background()
+	hash := header.Hash([]byte("test-hash"))
+	expectedHeader := &types.SignedHeader{} // Just a dummy
+
+	t.Run("Hit in Store", func(t *testing.T) {
+		mockEx := extmocks.NewMockExchange[*types.SignedHeader](t)
+		// Exchange should NOT be called
+
+		getter := func(ctx context.Context, s store.Store, h header.Hash) (*types.SignedHeader, error) {
+			return expectedHeader, nil
+		}
+
+		ew := &exchangeWrapper[*types.SignedHeader]{
+			Exchange: mockEx,
+			daStore:  mocks.NewMockStore(t),
+			getter:   getter,
+		}
+
+		h, err := ew.Get(ctx, hash)
+		assert.NoError(t, err)
+		assert.Equal(t, expectedHeader, h)
+	})
+
+	t.Run("Miss in Store", func(t *testing.T) {
+		mockEx := extmocks.NewMockExchange[*types.SignedHeader](t)
+		mockEx.On("Get", ctx, hash).Return(expectedHeader, nil)
+
+		getter := func(ctx context.Context, s store.Store, h header.Hash) (*types.SignedHeader, error) {
+			return nil, errors.New("not found")
+		}
+
+		ew := &exchangeWrapper[*types.SignedHeader]{
+			Exchange: mockEx,
+			daStore:  mocks.NewMockStore(t),
+			getter:   getter,
+		}
+
+		h, err := ew.Get(ctx, hash)
+		assert.NoError(t, err)
+		assert.Equal(t, expectedHeader, h)
+	})
+
+	t.Run("Store Not Configured", func(t *testing.T) {
+		mockEx := extmocks.NewMockExchange[*types.SignedHeader](t)
+		mockEx.On("Get", ctx, hash).Return(expectedHeader, nil)
+
+		ew := &exchangeWrapper[*types.SignedHeader]{
+			Exchange: mockEx,
+			daStore:  nil, // No store
+			getter:   nil,
+		}
+
+		h, err := ew.Get(ctx, hash)
+		assert.NoError(t, err)
+		assert.Equal(t, expectedHeader, h)
+	})
+}
+
+func TestExchangeWrapper_GetByHeight(t *testing.T) {
+	ctx := context.Background()
+	height := uint64(10)
+	expectedHeader := &types.SignedHeader{}
+
+	t.Run("Hit in Store", func(t *testing.T) {
+		mockEx := extmocks.NewMockExchange[*types.SignedHeader](t)
+
+		getterByHeight := func(ctx context.Context, s store.Store, h uint64) (*types.SignedHeader, error) {
+			return expectedHeader, nil
+		}
+
+		ew := &exchangeWrapper[*types.SignedHeader]{
+			Exchange:       mockEx,
+			daStore:        mocks.NewMockStore(t),
+			getterByHeight: getterByHeight,
+		}
+
+		h, err := ew.GetByHeight(ctx, height)
+		assert.NoError(t, err)
+		assert.Equal(t, expectedHeader, h)
+	})
+
+	t.Run("Miss in Store", func(t *testing.T) {
+		mockEx := extmocks.NewMockExchange[*types.SignedHeader](t)
+		mockEx.On("GetByHeight", ctx, height).Return(expectedHeader, nil)
+
+		getterByHeight := func(ctx context.Context, s store.Store, h uint64) (*types.SignedHeader, error) {
+			return nil, errors.New("not found")
+		}
+
+		ew := &exchangeWrapper[*types.SignedHeader]{
+			Exchange:       mockEx,
+			daStore:        mocks.NewMockStore(t),
+			getterByHeight: getterByHeight,
+		}
+
+		h, err := ew.GetByHeight(ctx, height)
+		assert.NoError(t, err)
+		assert.Equal(t, expectedHeader, h)
+	})
+}

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,14 +49,19 @@ 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]
-	storeInitialized  atomic.Bool
+
+	getter           storeGetter[H]
+	getterByHeight   storeGetterByHeight[H]
+	storeInitialized atomic.Bool
 }
 
 // DataSyncService is the P2P Sync Service for blocks.
@@ -66,28 +72,48 @@ type HeaderSyncService = SyncService[*types.SignedHeader]
 
 // NewDataSyncService returns a new DataSyncService.
 func NewDataSyncService(
-	store ds.Batching,
+	dsStore 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)
+	getter := func(ctx context.Context, s store.Store, hash header.Hash) (*types.Data, error) {
+		_, d, err := s.GetBlockByHash(ctx, hash)
+		return d, err
+	}
+	getterByHeight := func(ctx context.Context, s store.Store, height uint64) (*types.Data, error) {
+		_, d, err := s.GetBlockData(ctx, height)
+		return d, err
+	}
+	return newSyncService[*types.Data](dsStore, daStore, getter, getterByHeight, dataSync, conf, genesis, p2p, logger)
 }
 
 // NewHeaderSyncService returns a new HeaderSyncService.
 func NewHeaderSyncService(
-	store ds.Batching,
+	dsStore 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)
+	getter := func(ctx context.Context, s store.Store, hash header.Hash) (*types.SignedHeader, error) {
+		h, _, err := s.GetBlockByHash(ctx, hash)
+		return h, err
+	}
+	getterByHeight := func(ctx context.Context, s store.Store, height uint64) (*types.SignedHeader, error) {
+		return s.GetHeader(ctx, height)
+	}
+	return newSyncService[*types.SignedHeader](dsStore, daStore, getter, getterByHeight, headerSync, conf, genesis, p2p, logger)
 }
 
 func newSyncService[H header.Header[H]](
-	store ds.Batching,
+	dsStore ds.Batching,
+	daStore store.Store,
+	getter storeGetter[H],
+	getterByHeight storeGetterByHeight[H],
 	syncType syncType,
 	conf config.Config,
 	genesis genesis.Genesis,
@@ -99,7 +125,7 @@ func newSyncService[H header.Header[H]](
 	}
 
 	ss, err := goheaderstore.NewStore[H](
-		store,
+		dsStore,
 		goheaderstore.WithStorePrefix(string(syncType)),
 		goheaderstore.WithMetrics(),
 	)
@@ -108,13 +134,16 @@ func newSyncService[H header.Header[H]](
 	}
 
 	svc := &SyncService[H]{
-		conf:         conf,
-		genesis:      genesis,
-		p2p:          p2p,
-		store:        ss,
-		syncType:     syncType,
-		logger:       logger,
-		syncerStatus: new(SyncerStatus),
+		conf:           conf,
+		genesis:        genesis,
+		p2p:            p2p,
+		store:          ss,
+		daStore:        daStore,
+		getter:         getter,
+		getterByHeight: getterByHeight,
+		syncType:       syncType,
+		logger:         logger,
+		syncerStatus:   new(SyncerStatus),
 	}
 
 	return svc, nil
@@ -284,12 +313,21 @@ 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,
+		getter:         syncService.getter,
+		getterByHeight: syncService.getterByHeight,
+	}
+
 	return peerIDs, nil
 }
 
@@ -388,7 +426,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() {