feat(sync): bifurcation for syncTarget

header.go

@@ -12,7 +12,8 @@ type Header[H any] interface {
 	// New creates new instance of a header.
 	// It exists to overcome limitation of Go's type system.
 	// See:
-	// https://go.googlesource.com/proposal/+/refs/heads/master/design/43651-type-parameters.md#pointer-method-example
+	//
+	//https://go.googlesource.com/proposal/+/refs/heads/master/design/43651-type-parameters.md#pointer-method-example
 	New() H
 	// IsZero reports whether Header is a zero value of it's concrete type.
 	IsZero() bool

headertest/dummy_header.go

@@ -30,6 +30,9 @@ type DummyHeader struct {
 	// SoftFailure allows for testing scenarios where a header would fail
 	// verification with SoftFailure set to true
 	SoftFailure bool
+
+	// VerifyFn can be used to change header.Verify behaviour per header.
+	VerifyFn func(hdr *DummyHeader) error `json:"-"`
 }
 
 func RandDummyHeader(t *testing.T) *DummyHeader {
@@ -100,6 +103,9 @@ func (d *DummyHeader) IsExpired(period time.Duration) bool {
 }
 
 func (d *DummyHeader) Verify(hdr *DummyHeader) error {
+	if d.VerifyFn != nil {
+		return d.VerifyFn(hdr)
+	}
 	if hdr.VerifyFailure {
 		return &header.VerifyError{Reason: ErrDummyVerify, SoftFailure: hdr.SoftFailure}
 	}

p2p/server_test.go

@@ -176,7 +176,7 @@ func (timeoutStore[H]) Append(ctx context.Context, _ ...H) error {
 	return ctx.Err()
 }
 
-func (timeoutStore[H]) GetRange(ctx context.Context, _ uint64, _ uint64) ([]H, error) {
+func (timeoutStore[H]) GetRange(ctx context.Context, _, _ uint64) ([]H, error) {
 	<-ctx.Done()
 	return nil, ctx.Err()
 }

sync/metrics.go

@@ -22,6 +22,7 @@ type metrics struct {
 	trustedPeersOutOfSync metric.Int64Counter
 	outdatedHeader        metric.Int64Counter
 	subjectiveInit        metric.Int64Counter
+	failedAgainstSubjHead metric.Int64Counter
 
 	subjectiveHead atomic.Int64
 
@@ -71,6 +72,16 @@ func newMetrics() (*metrics, error) {
 		return nil, err
 	}
 
+	failedAgainstSubjHead, err := meter.Int64Counter(
+		"hdr_sync_subj_validation_failed",
+		metric.WithDescription(
+			"tracks how many times validation against subjective head failed",
+		),
+	)
+	if err != nil {
+		return nil, err
+	}
+
 	subjectiveHead, err := meter.Int64ObservableGauge(
 		"hdr_sync_subjective_head_gauge",
 		metric.WithDescription("subjective head height"),
@@ -112,6 +123,7 @@ func newMetrics() (*metrics, error) {
 		trustedPeersOutOfSync: trustedPeersOutOfSync,
 		outdatedHeader:        outdatedHeader,
 		subjectiveInit:        subjectiveInit,
+		failedAgainstSubjHead: failedAgainstSubjHead,
 		syncLoopDurationHist:  syncLoopDurationHist,
 		syncLoopRunningInst:   syncLoopRunningInst,
 		requestRangeTimeHist:  requestRangeTimeHist,
@@ -186,6 +198,17 @@ func (m *metrics) newSubjectiveHead(ctx context.Context, height uint64, timestam
 	})
 }
 
+func (m *metrics) failedValidationAgainstSubjHead(ctx context.Context, height int64, hash string) {
+	m.observe(ctx, func(ctx context.Context) {
+		m.failedAgainstSubjHead.Add(ctx, 1,
+			metric.WithAttributes(
+				attribute.Int64("height", height),
+				attribute.String("hash", hash),
+			),
+		)
+	})
+}
+
 func (m *metrics) rangeRequestStart() {
 	if m == nil {
 		return

sync/sync_head.go

@@ -3,6 +3,7 @@ package sync
 import (
 	"context"
 	"errors"
+	"fmt"
 	"time"
 
 	"github.com/celestiaorg/go-header"
@@ -173,7 +174,12 @@ func (s *Syncer[H]) verify(ctx context.Context, newHead H) (bool, error) {
 	}
 
 	var verErr *header.VerifyError
-	if errors.As(err, &verErr) && !verErr.SoftFailure {
+	if errors.As(err, &verErr) {
+		if verErr.SoftFailure {
+			err := s.verifySkipping(ctx, sbjHead, newHead)
+			return err != nil, err
+		}
+
 		logF := log.Warnw
 		if errors.Is(err, header.ErrKnownHeader) {
 			logF = log.Debugw
@@ -186,7 +192,59 @@ func (s *Syncer[H]) verify(ctx context.Context, newHead H) (bool, error) {
 			"reason", verErr.Reason)
 	}
 
-	return verErr.SoftFailure, err
+	return false, err
+}
+
+// verifySkipping performs a bifurcated header verification process such that
+// it tries to find a header (or several headers if necessary)
+// between the networkHead and the subjectiveHead such that non-adjacent
+// (or in the worst case adjacent) verification passes and the networkHead
+// can be verified as a valid sync target against the syncer's subjectiveHead.
+//
+// When such candidates cannot be found [NewValidatorSetCantBeTrustedError] will be returned.
+func (s *Syncer[H]) verifySkipping(ctx context.Context, subjHead, networkHeader H) error {
+	subjHeight := subjHead.Height()
+
+	diff := networkHeader.Height() - subjHeight
+	if diff <= 0 {
+		panic(fmt.Sprintf("implementation bug:\n subjective head height %d, hash %X,\n network head height %d, hash %X",
+			subjHeight, subjHead.Hash(),
+			networkHeader.Height(), networkHeader.Hash(),
+		))
+	}
+
+	for diff > 1 {
+		candidateHeight := subjHeight + diff/2
+
+		candidateHeader, err := s.getter.GetByHeight(ctx, candidateHeight)
+		if err != nil {
+			return err
+		}
+
+		if err := header.Verify(subjHead, candidateHeader); err != nil {
+			// candidate failed, go deeper in 1st half.
+			diff = diff / 2
+			continue
+		}
+
+		// candidate was validated properly, update subjHead.
+		subjHead = candidateHeader
+		s.setSubjectiveHead(ctx, subjHead)
+
+		if err := header.Verify(subjHead, networkHeader); err == nil {
+			// network head validate properly, return success.
+			return nil
+		}
+
+		// new subjHead failed, go deeper in 2nd half.
+		subjHeight = subjHead.Height()
+		diff = networkHeader.Height() - subjHeight
+	}
+
+	s.metrics.failedValidationAgainstSubjHead(ctx, int64(networkHeader.Height()), networkHeader.Hash().String())
+	log.Warnw("sync: header validation against subjHead", "height", networkHeader.Height(), "hash", networkHeader.Hash().String())
+
+	return fmt.Errorf("sync: header validation against subjHead height:%d hash:%x", networkHeader.Height(), networkHeader.Hash().String())
 }
 
 // isExpired checks if header is expired against trusting period.