fix(sync): sync distant P2P data heads by range

pkg/sync/sync_service_test.go

@@ -5,9 +5,14 @@ import (
 	cryptoRand "crypto/rand"
 	"math/rand"
 	"path/filepath"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	goheader "github.com/celestiaorg/go-header"
+	"github.com/celestiaorg/go-header/headertest"
+	goheaderlocal "github.com/celestiaorg/go-header/local"
+	goheadersync "github.com/celestiaorg/go-header/sync"
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/sync"
 	"github.com/libp2p/go-libp2p/core/crypto"
@@ -26,6 +31,118 @@ import (
 	"github.com/evstack/ev-node/types"
 )
 
+type countingP2PDataGetter struct {
+	goheader.Getter[*types.P2PData]
+	getByHeightCalls atomic.Uint64
+	rangeCalls       atomic.Uint64
+}
+
+func (g *countingP2PDataGetter) GetByHeight(ctx context.Context, height uint64) (*types.P2PData, error) {
+	g.getByHeightCalls.Add(1)
+	return g.Getter.GetByHeight(ctx, height)
+}
+
+func (g *countingP2PDataGetter) GetRangeByHeight(
+	ctx context.Context,
+	from *types.P2PData,
+	to uint64,
+) ([]*types.P2PData, error) {
+	g.rangeCalls.Add(1)
+	return g.Getter.GetRangeByHeight(ctx, from, to)
+}
+
+type signalingP2PDataStore struct {
+	goheader.Store[*types.P2PData]
+	targetHeight uint64
+	syncComplete chan struct{}
+	signaled     atomic.Bool
+}
+
+func (s *signalingP2PDataStore) Append(ctx context.Context, data ...*types.P2PData) error {
+	if err := s.Store.Append(ctx, data...); err != nil {
+		return err
+	}
+
+	for _, item := range data {
+		if item.Height() == s.targetHeight && s.signaled.CompareAndSwap(false, true) {
+			close(s.syncComplete)
+			break
+		}
+	}
+
+	return nil
+}
+
+type verifierCapturingP2PDataSubscriber struct {
+	*headertest.Subscriber[*types.P2PData]
+	verifier func(context.Context, *types.P2PData) error
+}
+
+func (s *verifierCapturingP2PDataSubscriber) SetVerifier(
+	verifier func(context.Context, *types.P2PData) error,
+) error {
+	s.verifier = verifier
+	return nil
+}
+
+func TestDataSyncerDistantHeadUsesRangeSync(t *testing.T) {
+	const targetHeight uint64 = 64
+
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+
+	chain := make([]*types.P2PData, 0, targetHeight)
+	blockTime := time.Now().Add(-time.Duration(targetHeight) * time.Millisecond)
+	var previousHash types.Hash
+	for height := uint64(1); height <= targetHeight; height++ {
+		_, data := types.GetRandomBlock(height, 1, "data-sync-catchup")
+		data.Metadata.Time = uint64(blockTime.Add(time.Duration(height) * time.Millisecond).UnixNano())
+		data.LastDataHash = previousHash
+		previousHash = data.Hash()
+		chain = append(chain, &types.P2PData{Data: data})
+	}
+
+	remoteStore := &headertest.Store[*types.P2PData]{Headers: make(map[uint64]*types.P2PData)}
+	require.NoError(t, remoteStore.Append(ctx, chain...))
+	localKV := sync.MutexWrap(datastore.NewMapDatastore())
+	localStore := &signalingP2PDataStore{
+		Store: store.NewDataStoreAdapter(store.New(localKV), genesispkg.Genesis{
+			ChainID:       "data-sync-catchup",
+			InitialHeight: 1,
+		}),
+		targetHeight: targetHeight,
+		syncComplete: make(chan struct{}),
+	}
+	require.NoError(t, localStore.Append(ctx, chain[0]))
+
+	getter := &countingP2PDataGetter{Getter: goheaderlocal.NewExchange(remoteStore)}
+	subscriber := &verifierCapturingP2PDataSubscriber{
+		Subscriber: &headertest.Subscriber[*types.P2PData]{},
+	}
+	syncer, err := goheadersync.NewSyncer(
+		getter,
+		localStore,
+		subscriber,
+		goheadersync.WithBlockTime(time.Second),
+	)
+	require.NoError(t, err)
+	require.NoError(t, syncer.Start(ctx))
+	t.Cleanup(func() { _ = syncer.Stop(context.Background()) })
+
+	getter.getByHeightCalls.Store(0)
+	getter.rangeCalls.Store(0)
+	require.NoError(t, subscriber.verifier(ctx, chain[targetHeight-1]))
+
+	select {
+	case <-localStore.syncComplete:
+	case <-ctx.Done():
+		require.NoError(t, ctx.Err(), "waiting for data synchronization to complete")
+	}
+	require.LessOrEqual(t, getter.getByHeightCalls.Load(), uint64(1),
+		"validating a distant head must not fetch intermediate blocks one by one")
+	require.Positive(t, getter.rangeCalls.Load())
+}
+
 func TestHeaderSyncServiceStartForPublishingWithPeers(t *testing.T) {
 	mainKV := sync.MutexWrap(datastore.NewMapDatastore())
 	pk, _, err := crypto.GenerateEd25519Key(cryptoRand.Reader)

types/p2p_envelope.go

@@ -109,8 +109,12 @@ func (p *P2PData) DAHint() uint64 {
 	return p.DAHeightHint
 }
 
-// Verify verifies against untrusted data.
+// Verify verifies the data hash linkage for adjacent data.
 func (p *P2PData) Verify(untrusted *P2PData) error {
+	if p.Height()+1 != untrusted.Height() {
+		return nil
+	}
+
 	return p.Data.Verify(untrusted.Data)
 }
 

types/p2p_envelope_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	goheader "github.com/celestiaorg/go-header"
 	"github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -42,6 +43,28 @@ func TestP2PEnvelope_MarshalUnmarshal(t *testing.T) {
 	assert.Equal(t, envelope.Txs, newEnvelope.Txs)
 }
 
+func TestP2PDataVerifyAdjacentHeads(t *testing.T) {
+	now := time.Now()
+	_, trustedData := GetRandomBlock(10, 1, "test-chain")
+	trustedData.Metadata.Time = uint64(now.UnixNano())
+
+	_, validData := GetRandomBlock(11, 1, "test-chain")
+	validData.Metadata.Time = uint64(now.Add(time.Second).UnixNano())
+	validData.LastDataHash = trustedData.Hash()
+	require.NoError(t, goheader.Verify(
+		&P2PData{Data: trustedData},
+		&P2PData{Data: validData},
+	))
+
+	_, invalidData := GetRandomBlock(11, 1, "test-chain")
+	invalidData.Metadata.Time = uint64(now.Add(time.Second).UnixNano())
+	invalidData.LastDataHash = bytes.Repeat([]byte{0x1}, 32)
+	require.Error(t, goheader.Verify(
+		&P2PData{Data: trustedData},
+		&P2PData{Data: invalidData},
+	))
+}
+
 func TestP2PSignedHeader_MarshalUnmarshal(t *testing.T) {
 	_, pubKey, err := crypto.GenerateEd25519Key(nil)
 	require.NoError(t, err)