feat(sync): bifurcation for syncTarget (#219)

Fixes #217

Author: cristaloleg

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 behavior 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}
 	}

sync/metrics.go

@@ -26,6 +26,7 @@ type metrics struct {
 	trustedPeersOutOfSync metric.Int64Counter
 	outdatedHeader        metric.Int64Counter
 	subjectiveInit        metric.Int64Counter
+	failedBifurcations    metric.Int64Counter
 
 	subjectiveHead atomic.Uint64
 
@@ -75,6 +76,16 @@ func newMetrics() (*metrics, error) {
 		return nil, err
 	}
 
+	failedBifurcations, err := meter.Int64Counter(
+		"hdr_failed_bifurcations_total",
+		metric.WithDescription(
+			"tracks how many times bifurcation failed against subjective head",
+		),
+	)
+	if err != nil {
+		return nil, err
+	}
+
 	subjectiveHead, err := meter.Int64ObservableGauge(
 		"hdr_sync_subjective_head_gauge",
 		metric.WithDescription("subjective head height"),
@@ -116,6 +127,7 @@ func newMetrics() (*metrics, error) {
 		trustedPeersOutOfSync: trustedPeersOutOfSync,
 		outdatedHeader:        outdatedHeader,
 		subjectiveInit:        subjectiveInit,
+		failedBifurcations:    failedBifurcations,
 		syncLoopDurationHist:  syncLoopDurationHist,
 		syncLoopRunningInst:   syncLoopRunningInst,
 		requestRangeTimeHist:  requestRangeTimeHist,
@@ -198,6 +210,17 @@ func (m *metrics) newSubjectiveHead(ctx context.Context, height uint64, timestam
 	})
 }
 
+func (m *metrics) failedBifurcation(ctx context.Context, height uint64, hash string) {
+	m.observe(ctx, func(ctx context.Context) {
+		m.failedBifurcations.Add(ctx, 1,
+			metric.WithAttributes(
+				attribute.Int64("height", int64(height)), //nolint:gosec
+				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"
@@ -164,6 +165,7 @@ func (s *Syncer[H]) incomingNetworkHead(ctx context.Context, head H) error {
 }
 
 // verify verifies given network head candidate.
+// bool reports whether the returned error is a soft error.
 func (s *Syncer[H]) verify(ctx context.Context, newHead H) (bool, error) {
 	sbjHead, err := s.subjectiveHead(ctx)
 	if err != nil {
@@ -178,7 +180,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.verifyBifurcating(ctx, sbjHead, newHead)
+			return err != nil, err
+		}
+
 		logF := log.Warnw
 		if errors.Is(err, header.ErrKnownHeader) {
 			logF = log.Debugw
@@ -194,6 +201,69 @@ func (s *Syncer[H]) verify(ctx context.Context, newHead H) (bool, error) {
 	return verErr.SoftFailure, err
 }
 
+// verifyBifurcating verifies networkHead against subjHead via the interim headers when direct
+// verification is impossible.
+// 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.
+// A non-nil error is returned when networkHead can't be verified.
+func (s *Syncer[H]) verifyBifurcating(ctx context.Context, subjHead, networkHead H) error {
+	log.Warnw("header bifurcation started",
+		"height", networkHead.Height(),
+		"hash", networkHead.Hash().String(),
+	)
+
+	subjHeight := subjHead.Height()
+
+	diff := networkHead.Height() - subjHeight
+
+	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 {
+			var verErr *header.VerifyError
+			if errors.As(err, &verErr) && !verErr.SoftFailure {
+				return err
+			}
+
+			// candidate failed, go deeper in 1st half.
+			diff /= 2
+			continue
+		}
+
+		// candidate was validated properly, update subjHead.
+		subjHead = candidateHeader
+
+		if err := header.Verify(subjHead, networkHead); err == nil {
+			// network head validate properly, return success.
+			return nil
+		}
+
+		// new subjHead failed, go deeper in 2nd half.
+		subjHeight = subjHead.Height()
+		diff = networkHead.Height() - subjHeight
+	}
+
+	s.metrics.failedBifurcation(ctx, networkHead.Height(), networkHead.Hash().String())
+	log.Errorw("header bifurcation failed",
+		"height", networkHead.Height(),
+		"hash", networkHead.Hash().String(),
+	)
+
+	return &header.VerifyError{
+		Reason: fmt.Errorf("sync: header validation against subjHead height:%d hash:%s",
+			networkHead.Height(), networkHead.Hash().String(),
+		),
+		SoftFailure: false,
+	}
+}
+
 // isExpired checks if header is expired against trusting period.
 func isExpired[H header.Header[H]](header H, period time.Duration) bool {
 	expirationTime := header.Time().Add(period)