estimation test

Author: Wondertan

headertest/dummy_suite.go

@@ -5,6 +5,8 @@ import (
 	"time"
 )
 
+const HeaderTime = time.Nanosecond
+
 // DummySuite provides everything you need to test chain of DummyHeaders.
 // If not, please don't hesitate to extend it for your case.
 type DummySuite struct {
@@ -42,7 +44,7 @@ func (s *DummySuite) NextHeader() *DummyHeader {
 	}
 
 	dh := RandDummyHeader(s.t)
-	dh.Timestamp = s.head.Time().Add(time.Nanosecond)
+	dh.Timestamp = s.head.Time().Add(HeaderTime)
 	dh.HeightI = s.head.Height() + 1
 	dh.PreviousHash = s.head.Hash()
 	dh.Chainid = s.head.ChainID()

sync/sync_head.go

@@ -50,13 +50,7 @@ func (s *Syncer[H]) Head(ctx context.Context, _ ...header.HeadOption[H]) (H, err
 	// NOTE: We could trust the netHead like we do during 'automatic subjective initialization'
 	// but in this case our subjective head is not expired, so we should verify netHead
 	// and only if it is valid, set it as new head
-	err = s.incomingNetworkHead(ctx, netHead)
-	if err != nil {
-		log.Errorw("incoming network head failed",
-			"height", netHead.Height(),
-			"hash", netHead.Hash().String(),
-			"err", err)
-	}
+	_ = s.incomingNetworkHead(ctx, netHead)
 	// netHead was either accepted or rejected as the new subjective
 	// anyway return most current known subjective head
 	return s.subjectiveHead(ctx)
@@ -155,6 +149,11 @@ func (s *Syncer[H]) incomingNetworkHead(ctx context.Context, head H) error {
 	//  To be reworked by bsync.
 	_, err := s.subjectiveTail(ctx, head)
 	if err != nil {
+		log.Errorw("subjective tail failed",
+			"new_head", head.Height(),
+			"hash", head.Hash().String(),
+			"err", err,
+		)
 		return err
 	}
 

sync/sync_tail.go

@@ -12,7 +12,6 @@ import (
 
 // TODO:
 //  * Refactor tests
-//  * Write tests for estimation
 //  * Ensure sync always happen on start
 
 // subjectiveTail returns the current Tail header.
@@ -26,19 +25,23 @@ func (s *Syncer[H]) subjectiveTail(ctx context.Context, head H) (H, error) {
 
 	var fetched bool
 	if tailHash, outdated := s.isTailHashOutdated(tail); outdated {
+		log.Debugw("tail hash updated", "hash", tailHash)
 		tail, err = s.store.Get(ctx, tailHash)
 		if err != nil {
+			log.Debugw("tail hash not available locally, fetching...", "hash", tailHash)
 			tail, err = s.getter.Get(ctx, tailHash)
 			if err != nil {
 				return tail, fmt.Errorf("getting SyncFromHash tail(%x): %w", tailHash, err)
 			}
 			fetched = true
 		}
 	} else if tailHeight, outdated := s.isTailHeightOutdated(tail); outdated {
+		log.Debugw("tail height updated", "height", tailHeight)
 		if tailHeight <= s.store.Height() {
 			tail, err = s.store.GetByHeight(ctx, tailHeight)
 		}
 		if err != nil || tailHeight != tail.Height() {
+			log.Debugw("tail height not available locally, fetching...", "height", tailHeight)
 			tail, err = s.getter.GetByHeight(ctx, tailHeight)
 			if err != nil {
 				return tail, fmt.Errorf("getting SyncFromHeight tail(%d): %w", tailHeight, err)
@@ -62,22 +65,26 @@ func (s *Syncer[H]) subjectiveTail(ctx context.Context, head H) (H, error) {
 				// current tail is relevant as is
 				return tail, nil
 			}
+			log.Debugw("current tail is beyond pruning window", "current_height", tail.Height(), "diff", diff.String())
 
 			toDeleteEstimate := uint64(diff / s.Params.blockTime) //nolint:gosec
 			estimatedNewTail := tail.Height() + toDeleteEstimate
 
 			for {
 				tail, err = s.store.GetByHeight(ctx, estimatedNewTail)
 				if err != nil {
-					log.Errorw("getting estimated tail from store ", "error", err)
+					log.Errorw("getting estimated tail from store", "height", estimatedNewTail, "error", err)
 					return tail, err
 				}
-				if tail.Time().UTC().Before(cutoffTime) {
+				if tail.Time().UTC().Compare(cutoffTime) <= 0 {
+					// tail before or equal to cutoffTime
 					break
 				}
 
 				estimatedNewTail++
 			}
+
+			log.Debugw("estimated new tail", "new_height", tail.Height())
 		}
 	}
 

sync/sync_tail_test.go

@@ -14,6 +14,51 @@ import (
 	"github.com/celestiaorg/go-header/store"
 )
 
+func TestSyncer_TailEstimation(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	t.Cleanup(cancel)
+
+	remoteStore := headertest.NewStore[*headertest.DummyHeader](t, headertest.NewTestSuite(t), 100)
+
+	ds := dssync.MutexWrap(datastore.NewMapDatastore())
+	localStore, err := store.NewStore[*headertest.DummyHeader](
+		ds,
+		store.WithWriteBatchSize(1),
+	)
+	require.NoError(t, err)
+	err = localStore.Start(ctx)
+	require.NoError(t, err)
+
+	syncer, err := NewSyncer[*headertest.DummyHeader](
+		remoteStore,
+		localStore,
+		headertest.NewDummySubscriber(),
+		WithBlockTime(headertest.HeaderTime),
+		WithPruningWindow(time.Nanosecond*50),
+	)
+	require.NoError(t, err)
+
+	err = syncer.Start(ctx)
+	require.NoError(t, err)
+	time.Sleep(time.Millisecond * 10)
+	err = syncer.SyncWait(ctx)
+	require.NoError(t, err)
+	require.EqualValues(t, 100, syncer.State().Height)
+
+	tail, err := localStore.Tail(ctx)
+	require.NoError(t, err)
+	require.EqualValues(t, tail.Height(), 1)
+
+	// simulate new header arrival by triggering recency check
+	head, err := syncer.Head(ctx)
+	require.NoError(t, err)
+	require.Equal(t, head.Height(), remoteStore.Height())
+
+	tail, err = localStore.Tail(ctx)
+	require.NoError(t, err)
+	require.EqualValues(t, 50, tail.Height())
+}
+
 func TestSyncer_TailReconfiguration(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
 	t.Cleanup(cancel)