header: Extract Head method into separate Head interface, make Syncer implement it

header/interface.go

@@ -104,8 +104,7 @@ type Store interface {
 // Getter contains the behavior necessary for a component to retrieve
 // headers that have been processed during header sync.
 type Getter interface {
-	// Head returns the ExtendedHeader of the chain head.
-	Head(context.Context) (*ExtendedHeader, error)
+	Head
 
 	// Get returns the ExtendedHeader corresponding to the given hash.
 	Get(context.Context, tmbytes.HexBytes) (*ExtendedHeader, error)
@@ -116,3 +115,11 @@ type Getter interface {
 	// GetRangeByHeight returns the given range [from:to) of ExtendedHeaders.
 	GetRangeByHeight(ctx context.Context, from, to uint64) ([]*ExtendedHeader, error)
 }
+
+// Head contains the behavior necessary for a component to retrieve
+// the chain head. Note that "chain head" is subjective to the component
+// reporting it.
+type Head interface {
+	// Head returns the ExtendedHeader of the chain head.
+	Head(context.Context) (*ExtendedHeader, error)
+}

header/sync/sync.go

@@ -3,6 +3,7 @@ package sync
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 	"time"
 
@@ -35,6 +36,10 @@ type Syncer struct {
 	exchange header.Exchange
 	store    header.Store
 
+	// blockTime provides a reference point for the Syncer to determine
+	// whether its subjective head is outdated
+	blockTime time.Duration
+
 	// stateLk protects state which represents the current or latest sync
 	stateLk sync.RWMutex
 	state   State
@@ -47,11 +52,12 @@ type Syncer struct {
 }
 
 // NewSyncer creates a new instance of Syncer.
-func NewSyncer(exchange header.Exchange, store header.Store, sub header.Subscriber) *Syncer {
+func NewSyncer(exchange header.Exchange, store header.Store, sub header.Subscriber, blockTime time.Duration) *Syncer {
 	return &Syncer{
 		sub:         sub,
 		exchange:    exchange,
 		store:       store,
+		blockTime:   blockTime,
 		triggerSync: make(chan struct{}, 1), // should be buffered
 	}
 }
@@ -88,6 +94,12 @@ func (s *Syncer) WaitSync(ctx context.Context) error {
 	return err
 }
 
+// Head tries to return the Syncer's view of the objective head of the
+// network.
+func (s *Syncer) Head(ctx context.Context) (*header.ExtendedHeader, error) {
+	return s.trustedHead(ctx)
+}
+
 // State collects all the information about a sync.
 type State struct {
 	ID                   uint64 // incrementing ID of a sync
@@ -133,16 +145,23 @@ func (s *Syncer) trustedHead(ctx context.Context) (*header.ExtendedHeader, error
 		return nil, err
 	}
 
-	// check if our subjective header is not expired and use it
-	if !sbj.IsExpired() {
+	// check if our subjective header is not expired and not too outdated (relative to the
+	// network's block time) and use it
+	if !sbj.IsExpired() && (time.Now().Sub(sbj.Time) <= s.blockTime) {
 		return sbj, nil
 	}
 
-	// otherwise, request head from a trustedPeer or, in other words, do automatic subjective initialization
+	// otherwise, attempt to request head from a trustedPeer or, in other words, do
+	// automatic subjective initialization
 	objHead, err := s.exchange.Head(ctx)
 	if err != nil {
 		return nil, err
 	}
+	// ensure that head returned from trustedPeer is more recent than subjective head
+	// as it is possible that the trustedPeer's reported head is not current.
+	if objHead.Height < sbj.Height {
+		return nil, fmt.Errorf("trusted peer failed to return current network head")
+	}
 
 	s.pending.Add(objHead)
 	return objHead, nil

header/sync/sync_test.go

@@ -14,6 +14,8 @@ import (
 	"github.com/celestiaorg/celestia-node/header/store"
 )
 
+var blockTime = time.Minute
+
 func TestSyncSimpleRequestingHead(t *testing.T) {
 	// this way we force local head of Syncer to expire, so it requests a new one from trusted peer
 	header.TrustingPeriod = time.Microsecond
@@ -33,7 +35,7 @@ func TestSyncSimpleRequestingHead(t *testing.T) {
 	require.NoError(t, err)
 
 	localStore := store.NewTestStore(ctx, t, head)
-	syncer := NewSyncer(local.NewExchange(remoteStore), localStore, &header.DummySubscriber{})
+	syncer := NewSyncer(local.NewExchange(remoteStore), localStore, &header.DummySubscriber{}, blockTime)
 	err = syncer.Start(ctx)
 	require.NoError(t, err)
 
@@ -67,7 +69,7 @@ func TestSyncCatchUp(t *testing.T) {
 
 	remoteStore := store.NewTestStore(ctx, t, head)
 	localStore := store.NewTestStore(ctx, t, head)
-	syncer := NewSyncer(local.NewExchange(remoteStore), localStore, &header.DummySubscriber{})
+	syncer := NewSyncer(local.NewExchange(remoteStore), localStore, &header.DummySubscriber{}, blockTime)
 	// 1. Initial sync
 	err := syncer.Start(ctx)
 	require.NoError(t, err)
@@ -111,7 +113,7 @@ func TestSyncPendingRangesWithMisses(t *testing.T) {
 
 	remoteStore := store.NewTestStore(ctx, t, head)
 	localStore := store.NewTestStore(ctx, t, head)
-	syncer := NewSyncer(local.NewExchange(remoteStore), localStore, &header.DummySubscriber{})
+	syncer := NewSyncer(local.NewExchange(remoteStore), localStore, &header.DummySubscriber{}, blockTime)
 	err := syncer.Start(ctx)
 	require.NoError(t, err)
 
@@ -153,3 +155,44 @@ func TestSyncPendingRangesWithMisses(t *testing.T) {
 	assert.Equal(t, exp.Height, have.Height)
 	assert.Empty(t, syncer.pending.Head()) // assert all cache from pending is used
 }
+
+// TestSyncHead tests the Syncer's Head method.
+func TestSyncHead(t *testing.T) {
+	// this way we force local head of Syncer to expire, so it requests a new one from trusted peer
+	header.TrustingPeriod = time.Microsecond
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	t.Cleanup(cancel)
+
+	suite := header.NewTestSuite(t, 3)
+	head := suite.Head()
+
+	remoteStore := store.NewTestStore(ctx, t, head)
+	_, err := remoteStore.Append(ctx, suite.GenExtendedHeaders(100)...)
+	require.NoError(t, err)
+
+	_, err = remoteStore.GetByHeight(ctx, 100)
+	require.NoError(t, err)
+
+	localStore := store.NewTestStore(ctx, t, head)
+	syncer := NewSyncer(local.NewExchange(remoteStore), localStore, &header.DummySubscriber{}, blockTime)
+	err = syncer.Start(ctx)
+	require.NoError(t, err)
+
+	_, err = localStore.GetByHeight(ctx, 100)
+	require.NoError(t, err)
+
+	// send some headers through "headersub" to increase pending cache
+	headers := suite.GenExtendedHeaders(10)
+	res := syncer.processIncoming(ctx, headers[len(headers)-1])
+	require.Equal(t, pubsub.ValidationAccept, res)
+
+	// compare Syncer's reported head against local store's head
+	storeHead, err := localStore.Head(ctx)
+	require.NoError(t, err)
+	syncHead, err := syncer.Head(ctx)
+	require.NoError(t, err)
+
+	// ensure Syncer reports its head from pending cache rather than the store
+	assert.Greater(t, syncHead.Height, storeHead.Height)
+}

node/components.go

@@ -88,7 +88,7 @@ func baseComponents(cfg *Config, store Store) fx.Option {
 		fx.Provide(services.HeaderStore),
 		fx.Invoke(services.HeaderStoreInit(&cfg.Services)),
 		fxutil.ProvideAs(services.FraudService, new(fraud.Service), new(fraud.Subscriber)),
-		fx.Provide(services.HeaderSyncer),
+		fx.Provide(services.HeaderSyncer(cfg.Services)),
 		fxutil.ProvideAs(services.P2PSubscriber, new(header.Broadcaster), new(header.Subscriber)),
 		fx.Provide(services.HeaderP2PExchangeServer),
 		// p2p components

node/config_opts.go

@@ -54,6 +54,14 @@ func WithTrustedPeers(addr ...string) Option {
 	}
 }
 
+// WithBlockTime overrides the default block time with the given
+// duration.
+func WithBlockTime(blockTime time.Duration) Option {
+	return func(sets *settings) {
+		sets.cfg.Services.BlockTime = blockTime
+	}
+}
+
 // WithPeersLimit overrides default peer limit for peers found during discovery.
 func WithPeersLimit(limit uint) Option {
 	return func(sets *settings) {

node/rpc_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/celestiaorg/celestia-node/header/local"
 	"github.com/celestiaorg/celestia-node/header/store"
 	"github.com/celestiaorg/celestia-node/header/sync"
+	"github.com/celestiaorg/celestia-node/params"
 	service "github.com/celestiaorg/celestia-node/service/header"
 	"github.com/celestiaorg/celestia-node/service/rpc"
 	"github.com/celestiaorg/celestia-node/service/share"
@@ -242,7 +243,7 @@ func setupHeaderService(ctx context.Context, t *testing.T) *service.Service {
 	_, err := localStore.Append(ctx, suite.GenExtendedHeaders(5)...)
 	require.NoError(t, err)
 	// create syncer
-	syncer := sync.NewSyncer(local.NewExchange(remoteStore), localStore, &header.DummySubscriber{})
+	syncer := sync.NewSyncer(local.NewExchange(remoteStore), localStore, &header.DummySubscriber{}, params.BlockTime)
 
 	return service.NewHeaderService(syncer, nil, nil, nil, localStore)
 }

node/services/config.go

@@ -22,6 +22,8 @@ type Config struct {
 	// Note: The trusted does *not* imply Headers are not verified, but trusted as reliable to fetch headers
 	// at any moment.
 	TrustedPeers []string
+	// BlockTime is the block time of the network.
+	BlockTime time.Duration
 	// NOTE: All further fields related to share/discovery.
 	// PeersLimit defines how many peers will be added during discovery.
 	PeersLimit uint
@@ -36,6 +38,7 @@ func DefaultConfig() Config {
 	return Config{
 		TrustedHash:       "",
 		TrustedPeers:      make([]string, 0),
+		BlockTime:         params.BlockTime,
 		PeersLimit:        3,
 		DiscoveryInterval: time.Second * 30,
 		AdvertiseInterval: time.Second * 30,

node/services/service.go

@@ -26,24 +26,31 @@ import (
 )
 
 // HeaderSyncer creates a new Syncer.
-func HeaderSyncer(
+func HeaderSyncer(cfg Config) func(
 	ctx context.Context,
 	lc fx.Lifecycle,
 	ex header.Exchange,
 	store header.Store,
 	sub header.Subscriber,
 	fservice fraud.Service,
 ) (*sync.Syncer, error) {
-	syncer := sync.NewSyncer(ex, store, sub)
-	lifecycleCtx := fxutil.WithLifecycle(ctx, lc)
-	lc.Append(fx.Hook{
-		OnStart: func(startCtx context.Context) error {
-			return FraudLifecycle(startCtx, lifecycleCtx, fraud.BadEncoding, fservice, syncer.Start, syncer.Stop)
-		},
-		OnStop: syncer.Stop,
-	})
-
-	return syncer, nil
+	return func(ctx context.Context,
+		lc fx.Lifecycle,
+		ex header.Exchange,
+		store header.Store,
+		sub header.Subscriber,
+		fservice fraud.Service,
+	) (*sync.Syncer, error) {
+		syncer := sync.NewSyncer(ex, store, sub, cfg.BlockTime)
+		lifecycleCtx := fxutil.WithLifecycle(ctx, lc)
+		lc.Append(fx.Hook{
+			OnStart: func(startCtx context.Context) error {
+				return FraudLifecycle(startCtx, lifecycleCtx, fraud.BadEncoding, fservice, syncer.Start, syncer.Stop)
+			},
+			OnStop: syncer.Stop,
+		})
+		return syncer, nil
+	}
 }
 
 // P2PSubscriber creates a new p2p.Subscriber.

node/testing.go

@@ -42,6 +42,7 @@ func TestNode(t *testing.T, tp Type, opts ...Option) *Node {
 		WithNetwork(params.Private),
 		WithRPCPort("0"),
 		WithKeyringSigner(TestKeyringSigner(t)),
+		WithBlockTime(time.Microsecond),
 	)
 	nd, err := New(tp, store, opts...)
 	require.NoError(t, err)

params/network.go

@@ -2,6 +2,7 @@ package params
 
 import (
 	"errors"
+	"time"
 
 	"github.com/libp2p/go-libp2p-core/peer"
 )
@@ -13,6 +14,8 @@ const (
 	// Private can be used to set up any private network, including local testing setups.
 	// Use CELESTIA_PRIVATE_GENESIS env var to enable Private by specifying its genesis block hash.
 	Private Network = "private"
+	// TODO @renaynay @Wondertan: Set this once it's agreed upon, ideally this could point at a core const
+	BlockTime = time.Second * 30
 )
 
 // Network is a type definition for DA network run by Celestia Node.