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

Author: jgimeno

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,87 @@ 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 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...))
+	localStore := &headertest.Store[*types.P2PData]{Headers: make(map[uint64]*types.P2PData)}
+	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]))
+	require.LessOrEqual(t, getter.getByHeightCalls.Load(), uint64(1),
+		"validating a distant head must not fetch intermediate blocks one by one")
+
+	require.Eventually(t, func() bool {
+		return syncer.State().ToHeight == targetHeight && localStore.Height() == targetHeight
+	}, time.Second, time.Millisecond)
+	require.Equal(t, targetHeight, localStore.Height())
+	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)