status-go/timesource/timesource.go

package timesource

import (
	"bytes"
	"errors"
	"sort"
	"sync"
	"time"

	"go.uber.org/zap"

	"github.com/beevik/ntp"

	"github.com/ethereum/go-ethereum/log"
)

const (
	// DefaultMaxAllowedFailures defines how many failures will be tolerated.
	DefaultMaxAllowedFailures = 4

	// FastNTPSyncPeriod period between ntp synchronizations before the first
	// successful connection.
	FastNTPSyncPeriod = 2 * time.Minute

	// SlowNTPSyncPeriod period between ntp synchronizations after the first
	// successful connection.
	SlowNTPSyncPeriod = 1 * time.Hour

	// DefaultRPCTimeout defines write deadline for single ntp server request.
	DefaultRPCTimeout = 2 * time.Second
)

// defaultServers will be resolved to the closest available,
// and with high probability resolved to the different IPs
var defaultServers = []string{
	"time.apple.com",
	"pool.ntp.org",
	"time.cloudflare.com",
	"time.windows.com",
	"ntp.neu.edu.cn",
	"ntp.nict.jp",
	"amazon.pool.ntp.org",
	"android.pool.ntp.org",
}
var errUpdateOffset = errors.New("failed to compute offset")

var ntpTimeSource *NTPTimeSource
var ntpTimeSourceCreator func() *NTPTimeSource
var now func() time.Time

type ntpQuery func(string, ntp.QueryOptions) (*ntp.Response, error)

type queryResponse struct {
	Offset time.Duration
	Error  error
}

type multiRPCError []error

func (e multiRPCError) Error() string {
	var b bytes.Buffer
	b.WriteString("RPC failed: ")
	more := false
	for _, err := range e {
		if more {
			b.WriteString("; ")
		}
		b.WriteString(err.Error())
		more = true
	}
	b.WriteString(".")
	return b.String()
}

func init() {
	ntpTimeSourceCreator = func() *NTPTimeSource {
		if ntpTimeSource != nil {
			return ntpTimeSource
		}
		ntpTimeSource = &NTPTimeSource{
			servers:           defaultServers,
			allowedFailures:   DefaultMaxAllowedFailures,
			fastNTPSyncPeriod: FastNTPSyncPeriod,
			slowNTPSyncPeriod: SlowNTPSyncPeriod,
			timeQuery:         ntp.QueryWithOptions,
		}
		return ntpTimeSource
	}
	now = time.Now
}

func computeOffset(timeQuery ntpQuery, servers []string, allowedFailures int) (time.Duration, error) {
	if len(servers) == 0 {
		return 0, nil
	}
	responses := make(chan queryResponse, len(servers))
	for _, server := range servers {
		go func(server string) {
			response, err := timeQuery(server, ntp.QueryOptions{
				Timeout: DefaultRPCTimeout,
			})
			if err == nil {
				err = response.Validate()
			}
			if err != nil {
				responses <- queryResponse{Error: err}
				return
			}
			responses <- queryResponse{Offset: response.ClockOffset}
		}(server)
	}
	var (
		rpcErrors multiRPCError
		offsets   []time.Duration
		collected int
	)
	for response := range responses {
		if response.Error != nil {
			rpcErrors = append(rpcErrors, response.Error)
		} else {
			offsets = append(offsets, response.Offset)
		}
		collected++
		if collected == len(servers) {
			break
		}
	}
	if lth := len(rpcErrors); lth > allowedFailures {
		return 0, rpcErrors
	} else if lth == len(servers) {
		return 0, rpcErrors
	}
	sort.SliceStable(offsets, func(i, j int) bool {
		return offsets[i] > offsets[j]
	})
	mid := len(offsets) / 2
	if len(offsets)%2 == 0 {
		return (offsets[mid-1] + offsets[mid]) / 2, nil
	}
	return offsets[mid], nil
}

// Default initializes time source with default config values.
func Default() *NTPTimeSource {
	return ntpTimeSourceCreator()
}

type timeCallback struct {
	wg       *sync.WaitGroup
	callback func(time.Time)
}

// NTPTimeSource provides source of time that tries to be resistant to time skews.
// It does so by periodically querying time offset from ntp servers.
type NTPTimeSource struct {
	servers           []string
	allowedFailures   int
	fastNTPSyncPeriod time.Duration
	slowNTPSyncPeriod time.Duration
	timeQuery         ntpQuery // for ease of testing

	quit          chan struct{}
	wg            sync.WaitGroup
	started       bool
	updatedOffset bool
	callbacks     []timeCallback

	mu           sync.RWMutex
	latestOffset time.Duration
}

// AddCallback adds callback that will be called when offset is updated.
// If offset is already updated once, callback will be called immediately.
func (s *NTPTimeSource) AddCallback(callback func(time.Time)) *sync.WaitGroup {
	s.mu.Lock()
	defer s.mu.Unlock()
	var wg sync.WaitGroup
	if s.updatedOffset {
		callback(now().Add(s.latestOffset))
	} else {
		wg.Add(1)
		s.callbacks = append(s.callbacks, timeCallback{wg: &wg, callback: callback})
	}
	return &wg
}

// Now returns time adjusted by latest known offset
func (s *NTPTimeSource) Now() time.Time {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return now().Add(s.latestOffset)
}

func (s *NTPTimeSource) updateOffset() error {
	offset, err := computeOffset(s.timeQuery, s.servers, s.allowedFailures)
	if err != nil {
		log.Error("failed to compute offset", "error", err)
		s.mu.Lock()
		defer s.mu.Unlock()
		for _, c := range s.callbacks {
			c.callback(now())
			c.wg.Done()
		}
		return errUpdateOffset
	}
	log.Info("Difference with ntp servers", "offset", offset)
	s.mu.Lock()
	defer s.mu.Unlock()
	s.latestOffset = offset
	s.updatedOffset = true

	for _, c := range s.callbacks {
		c.callback(now().Add(s.latestOffset))
		c.wg.Done()
	}
	s.callbacks = nil
	return nil
}

// runPeriodically runs periodically the given function based on NTPTimeSource
// synchronization limits (fastNTPSyncPeriod / slowNTPSyncPeriod)
func (s *NTPTimeSource) runPeriodically(fn func() error) error {
	if s.started {
		return nil
	}

	var period time.Duration
	s.quit = make(chan struct{})
	// we try to do it synchronously so that user can have reliable messages right away
	s.wg.Add(1)
	go func() {
		for {
			select {
			case <-time.After(period):
				if err := fn(); err == nil {
					period = s.slowNTPSyncPeriod
				} else if period != s.slowNTPSyncPeriod {
					period = s.fastNTPSyncPeriod
				}

			case <-s.quit:
				s.wg.Done()
				return
			}
		}
	}()

	s.started = true
	return nil
}

// Start runs a goroutine that updates local offset every updatePeriod.
func (s *NTPTimeSource) Start() error {
	return s.runPeriodically(s.updateOffset)
}

// Stop goroutine that updates time source.
func (s *NTPTimeSource) Stop() error {
	if s.quit == nil {
		return nil
	}
	close(s.quit)
	s.wg.Wait()
	s.started = false
	return nil
}

func GetCurrentTimeInMillis() (uint64, error) {
	now, err := GetCurrentTime()
	if err != nil {
		return 0, err
	}
	return uint64(now.UnixNano() / int64(time.Millisecond)), nil
}

func GetCurrentTime() (*time.Time, error) {
	ts := Default()
	if err := ts.Start(); err != nil {
		return nil, err
	}
	var t *time.Time
	ts.AddCallback(func(now time.Time) {
		t = &now
	}).Wait()
	if ts.updatedOffset {
		return t, nil
	}
	return nil, errUpdateOffset
}

func Time() time.Time {
	now, err := GetCurrentTime()
	if err != nil {
		log.Error("[timesource] error when getting current time", zap.Error(err))
		return time.Now()
	}
	return *now
}