feat(sync): non-recursive singleflight protection for Head (#229)

Closes: #154

This PR adds singleflight protection to prevent duplicate header
requests in the Exchange component. The focus is on minimizing redundant
requests, especially around the Head method, which is most likely to
experience simultaneous calls.

I moved out single-flight protection out of exchange to be in sync.Syncer. This way the single-flight protection works for every Head implementation: p2p, core. Besides, I avoid using the x/sync/singleflight as it does not handle context cancellation.

This also fixes an extremely rare but possible issue with subjective initialization similar to #247 (cc @walldiss). This issue is caused by to be removed syncGetter when used together with recursion. The fix is to move singleflight handling out of sync.Syncer, it should not be responsible for this. This way we fix the issue, without the need for read/write refactoring in sync_head.go we discussed.

---------

Co-authored-by: Wondertan <[email protected]>
Co-authored-by: IliaBulavintsev <[email protected]>

Author: 3 people

sync/head_sync.go

@@ -0,0 +1,51 @@
+package sync
+
+import (
+	"context"
+	"sync"
+
+	"github.com/celestiaorg/go-header"
+)
+
+// syncHead is a wrapper around the header.Head that provides single-flight access to the Head.
+type syncHead[H header.Header[H]] struct {
+	headMu sync.Mutex
+	headCh chan struct{}
+	head   header.Head[H]
+
+	resHead H
+	resErr  error
+}
+
+func (sh *syncHead[H]) Head(ctx context.Context, opts ...header.HeadOption[H]) (H, error) {
+	sh.headMu.Lock()
+	doneCh := sh.headCh
+	acquired := doneCh == nil
+	if acquired {
+		doneCh = make(chan struct{})
+		sh.headCh = doneCh
+	}
+	sh.headMu.Unlock()
+
+	if acquired {
+		head, err := sh.head.Head(ctx, opts...)
+
+		sh.headMu.Lock()
+		sh.resHead, sh.resErr = head, err
+		sh.headCh = nil
+		sh.headMu.Unlock()
+
+		close(doneCh)
+		return head, err
+	}
+
+	select {
+	case <-doneCh:
+		sh.headMu.Lock()
+		defer sh.headMu.Unlock()
+		return sh.resHead, sh.resErr
+	case <-ctx.Done():
+		var zero H
+		return zero, ctx.Err()
+	}
+}

sync/head_sync_test.go

@@ -0,0 +1,48 @@
+package sync
+
+import (
+	"context"
+	"math/rand/v2"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	libhead "github.com/celestiaorg/go-header"
+	"github.com/celestiaorg/go-header/headertest"
+)
+
+func Test_syncHead(t *testing.T) {
+	counter := &headCounter{}
+	sh := &syncHead[*headertest.DummyHeader]{head: counter}
+
+	callsN := 1000
+	wg := sync.WaitGroup{}
+	wg.Add(callsN)
+	for i := 0; i < callsN; i++ {
+		go func() {
+			defer wg.Done()
+			time.Sleep(time.Duration(rand.IntN(1000)) * time.Microsecond)
+			_, _ = sh.Head(context.Background())
+		}()
+	}
+
+	wg.Wait()
+
+	assert.Less(t, int(counter.cntr.Load()), callsN/100) // <1% of calls should go through
+}
+
+type headCounter struct {
+	cntr atomic.Int64
+}
+
+func (h *headCounter) Head(
+	ctx context.Context,
+	h2 ...libhead.HeadOption[*headertest.DummyHeader],
+) (*headertest.DummyHeader, error) {
+	time.Sleep(time.Millisecond * 1)
+	h.cntr.Add(1)
+	return &headertest.DummyHeader{}, nil
+}

sync/sync.go

@@ -33,7 +33,8 @@ var log = logging.Logger("header/sync")
 type Syncer[H header.Header[H]] struct {
 	sub     header.Subscriber[H] // to subscribe for new Network Heads
 	store   syncStore[H]         // to store all the headers to
-	getter  syncGetter[H]        // to fetch headers from
+	head    syncHead[H]
+	getter  header.Getter[H] // to fetch headers from
 	metrics *metrics
 
 	// stateLk protects state which represents the current or latest sync
@@ -80,7 +81,8 @@ func NewSyncer[H header.Header[H]](
 	return &Syncer[H]{
 		sub:         sub,
 		store:       syncStore[H]{Store: store},
-		getter:      syncGetter[H]{Getter: getter},
+		head:        syncHead[H]{head: getter},
+		getter:      getter,
 		metrics:     metrics,
 		triggerSync: make(chan struct{}, 1), // should be buffered
 		Params:      &params,

sync/sync_getter.go

@@ -30,18 +30,11 @@ func (s *Syncer[H]) Head(ctx context.Context, _ ...header.HeadOption[H]) (H, err
 		return sbjHead, nil
 	}
 
-	// single-flight protection ensure only one Head is requested at the time
-	if !s.getter.Lock() {
-		// means that other routine held the lock and set the subjective head
-		return s.subjectiveHead(ctx)
-	}
-	defer s.getter.Unlock()
-
 	s.metrics.outdatedHead(s.ctx)
 
 	reqCtx, cancel := context.WithTimeout(ctx, headRequestTimeout)
 	defer cancel()
-	netHead, err := s.getter.Head(reqCtx, header.WithTrustedHead[H](sbjHead))
+	netHead, err := s.head.Head(reqCtx, header.WithTrustedHead[H](sbjHead))
 	if err != nil {
 		log.Warnw(
 			"failed to get recent head, returning current subjective",
@@ -86,16 +79,8 @@ func (s *Syncer[H]) subjectiveHead(ctx context.Context) (H, error) {
 	}
 	// otherwise, request head from a trusted peer
 	log.Infow("stored head header expired", "height", storeHead.Height())
-	// single-flight protection
-	// ensure only one Head is requested at the time
-	if !s.getter.Lock() {
-		// means that other routine held the lock and set the subjective head for us,
-		// so just recursively get it
-		return s.subjectiveHead(ctx)
-	}
-	defer s.getter.Unlock()
 
-	trustHead, err := s.getter.Head(ctx)
+	trustHead, err := s.head.Head(ctx)
 	if err != nil {
 		return trustHead, err
 	}