Merge branch 'project/Channels' into XX-4228_clock_skew_tracker

bindings/timeNow.go

+///////////////////////////////////////////////////////////////////////////////
+// Copyright © 2020 xx network SEZC                                          //
+//                                                                           //
+// Use of this source code is governed by a license that can be found in the //
+// LICENSE file                                                              //
+///////////////////////////////////////////////////////////////////////////////
+
+package bindings
+
+import (
+	"gitlab.com/xx_network/primitives/netTime"
+	"time"
+)
+
+// SetTimeSource will set the time source that will be used when retrieving the
+// current time using [netTime.Now]. This should be called BEFORE Login()
+// and only be called once. Using this after Login is undefined behavior that
+// may result in a crash.
+//
+// Parameters:
+//  - timeNow is an object which adheres to [netTime.TimeSource]. Specifically,
+//    this object should a NowMs() method which return a 64-bit integer value.
+func SetTimeSource(timeNow netTime.TimeSource) {
+	netTime.SetTimeSource(timeNow)
+}
+
+// SetOffset will set an internal offset variable. All calls to [netTime.Now]
+// will have this offset applied to this value.
+//
+// Parameters:
+//  - offset is a time by which netTime.Now will be offset. This value may be
+//    negative or positive. This expects a 64-bit integer value which will
+//    represent the number in microseconds this offset will be.
+func SetOffset(offset int64) {
+	netTime.SetOffset(time.Duration(offset) * time.Microsecond)
+}

cmix/follow.go

@@ -340,7 +340,7 @@ func (c *client) follow(identity receptionID.IdentityUse,
 		if !hasMessage && c.verboseRounds != nil {
 			c.verboseRounds.denote(rid, RoundState(NoMessageAvailable))
 		}
-		jww.INFO.Printf("[LOOKUP] round %d checked for %d, has message: %s", rid, identity.EphId.Int64(), hasMessage)
+		//jww.INFO.Printf("[LOOKUP] round %d checked for %d, has message: %s", rid, identity.EphId.Int64(), hasMessage)
 		return hasMessage
 	}
 
@@ -391,7 +391,7 @@ func (c *client) follow(identity receptionID.IdentityUse,
 		gwRoundsState.RangeUnchecked(
 			updatedEarliestRound, c.param.KnownRoundsThreshold, roundChecker)
 
-	jww.DEBUG.Printf("Processed RangeUnchecked for %s, Oldest: %d, "+
+	jww.DEBUG.Printf("Processed RangeUnchecked for %d, Oldest: %d, "+
 		"firstUnchecked: %d, last Checked: %d, threshold: %d, "+
 		"NewEarliestRemaining: %d, NumWithMessages: %d, NumUnknown: %d",
 		identity.EphId.Int64(), updatedEarliestRound, gwRoundsState.GetFirstUnchecked(),

cmix/health/callback.go

+package health
+
+import "sync"
+
+type trackerCallback struct {
+	funcs   map[uint64]func(isHealthy bool)
+	funcsID uint64
+
+	mux sync.RWMutex
+}
+
+func initTrackerCallback() *trackerCallback {
+	return &trackerCallback{
+		funcs:   map[uint64]func(isHealthy bool){},
+		funcsID: 0,
+	}
+}
+
+// addHealthCallback adds a function to the list of tracker functions such that
+// each function can be run after network changes. Returns a unique ID for the
+// function.
+func (t *trackerCallback) addHealthCallback(f func(isHealthy bool), health bool) uint64 {
+	var currentID uint64
+
+	t.mux.Lock()
+	t.funcs[t.funcsID] = f
+	currentID = t.funcsID
+	t.funcsID++
+	t.mux.Unlock()
+
+	go f(health)
+
+	return currentID
+}
+
+// RemoveHealthCallback removes the function with the given ID from the list of
+// tracker functions so that it will no longer be run.
+func (t *trackerCallback) RemoveHealthCallback(chanID uint64) {
+	t.mux.Lock()
+	delete(t.funcs, chanID)
+	t.mux.Unlock()
+}
+
+// callback calls every function with the new health state
+func (t *trackerCallback) callback(health bool) {
+	t.mux.Lock()
+	defer t.mux.Unlock()
+
+	// Run all listening functions
+	for _, f := range t.funcs {
+		go f(health)
+	}
+}

cmix/health/tracker.go

@@ -11,8 +11,7 @@
 package health
 
 import (
-	"errors"
-	"sync"
+	"sync/atomic"
 	"time"
 
 	jww "github.com/spf13/jwalterweatherman"
@@ -29,108 +28,130 @@ type Monitor interface {
 }
 
 type tracker struct {
+	// timeout parameter describes how long
+	// without good news until the network is considered unhealthy
 	timeout time.Duration
 
+	// channel on which new status updates are received from the network handler
 	heartbeat chan network.Heartbeat
 
-	funcs      map[uint64]func(isHealthy bool)
-	channelsID uint64
-	funcsID    uint64
-
-	running bool
-
-	// Determines the current health status
-	isHealthy bool
+	// denotes the last time news was heard. Both hold ns since unix epoc
+	// in an atomic
+	lastCompletedRound *int64
+	lastWaitingRound   *int64
 
 	// Denotes that the past health status wasHealthy is true if isHealthy has
-	// ever been true
-	wasHealthy bool
-	mux        sync.RWMutex
+	// ever been true in an atomic.
+	wasHealthy *uint32
+
+	// stores registered callbacks to receive event updates
+	*trackerCallback
 }
 
 // Init creates a single HealthTracker thread, starts it, and returns a tracker
 // and a stoppable.
 func Init(instance *network.Instance, timeout time.Duration) Monitor {
-	tracker := newTracker(timeout)
-	instance.SetNetworkHealthChan(tracker.heartbeat)
 
-	return tracker
+	trkr := newTracker(timeout)
+	instance.SetNetworkHealthChan(trkr.heartbeat)
+
+	return trkr
 }
 
 // newTracker builds and returns a new tracker object given a Context.
 func newTracker(timeout time.Duration) *tracker {
-	return &tracker{
+
+	lastCompletedRound := int64(0)
+	lastWaitingRound := int64(0)
+
+	wasHealthy := uint32(0)
+
+	t := &tracker{
 		timeout:            timeout,
-		funcs:     map[uint64]func(isHealthy bool){},
 		heartbeat:          make(chan network.Heartbeat, 100),
-		isHealthy: false,
-		running:   false,
+		lastCompletedRound: &lastCompletedRound,
+		lastWaitingRound:   &lastWaitingRound,
+		wasHealthy:         &wasHealthy,
 	}
+	t.trackerCallback = initTrackerCallback()
+	return t
 }
 
-// AddHealthCallback adds a function to the list of tracker functions such that
-// each function can be run after network changes. Returns a unique ID for the
-// function.
-func (t *tracker) AddHealthCallback(f func(isHealthy bool)) uint64 {
-	var currentID uint64
+// getLastCompletedRoundTimestamp atomically loads the completed round timestamp
+// and converts it to a time object, then returns it
+func (t *tracker) getLastCompletedRoundTimestamp() time.Time {
+	return time.Unix(0, atomic.LoadInt64(t.lastCompletedRound))
+}
 
-	t.mux.Lock()
-	t.funcs[t.funcsID] = f
-	currentID = t.funcsID
-	t.funcsID++
-	t.mux.Unlock()
+// getLastWaitingRoundTimestamp atomically loads the waiting round timestamp
+// and converts it to a time object, then returns it
+func (t *tracker) getLastWaitingRoundTimestamp() time.Time {
+	return time.Unix(0, atomic.LoadInt64(t.lastWaitingRound))
+}
 
-	go f(t.IsHealthy())
+// IsHealthy returns true if the network is healthy, which is
+// defined as the client having knowledge of both valid queued rounds
+// and completed rounds within the last tracker.timeout seconds
+func (t *tracker) IsHealthy() bool {
+	// use the system time instead of netTime.Now() which can
+	// include an offset because local monotonicity is what
+	// matters here, not correctness relative to absolute time
+	now := time.Now()
 
-	return currentID
+	completedRecently := false
+	if now.Sub(t.getLastCompletedRoundTimestamp()) < t.timeout {
+		completedRecently = true
 	}
 
-// RemoveHealthCallback removes the function with the given ID from the list of
-// tracker functions so that it will no longer be run.
-func (t *tracker) RemoveHealthCallback(chanID uint64) {
-	t.mux.Lock()
-	delete(t.funcs, chanID)
-	t.mux.Unlock()
+	waitingRecently := false
+	if now.Sub(t.getLastWaitingRoundTimestamp()) < t.timeout {
+		waitingRecently = true
 	}
 
-func (t *tracker) IsHealthy() bool {
-	t.mux.RLock()
-	defer t.mux.RUnlock()
+	return completedRecently && waitingRecently
+}
+
+// updateHealth atomically updates the internal
+// timestamps to now if there are new waiting / completed
+// rounds
+func (t *tracker) updateHealth(hasWaiting, hasCompleted bool) {
+	// use the system time instead of netTime.Now() which can
+	// include an offset because local monotonicity is what
+	// matters here, not correctness relative to absolute time
+	now := time.Now().UnixNano()
 
-	return t.isHealthy
+	if hasWaiting {
+		atomic.StoreInt64(t.lastWaitingRound, now)
 	}
 
-// WasHealthy returns true if isHealthy has ever been true.
-func (t *tracker) WasHealthy() bool {
-	t.mux.RLock()
-	defer t.mux.RUnlock()
+	if hasCompleted {
+		atomic.StoreInt64(t.lastCompletedRound, now)
+	}
+}
 
-	return t.wasHealthy
+// forceUnhealthy cleats the internal timestamps, forcing the
+// tracker into unhealthy
+func (t *tracker) forceUnhealthy() {
+	atomic.StoreInt64(t.lastWaitingRound, 0)
+	atomic.StoreInt64(t.lastCompletedRound, 0)
 }
 
-func (t *tracker) setHealth(h bool) {
-	t.mux.Lock()
-	// Only set wasHealthy to true if either
-	//  wasHealthy is true or
-	//  wasHealthy is false but h value is true
-	t.wasHealthy = t.wasHealthy || h
-	t.isHealthy = h
-	t.mux.Unlock()
+// WasHealthy returns true if isHealthy has ever been true.
+func (t *tracker) WasHealthy() bool {
+	return atomic.LoadUint32(t.wasHealthy) == 1
+}
 
-	t.transmit(h)
+// AddHealthCallback adds a function to the list of tracker functions such that
+// each function can be run after network changes. Returns a unique ID for the
+// function.
+func (t *tracker) AddHealthCallback(f func(isHealthy bool)) uint64 {
+	return t.addHealthCallback(f, t.IsHealthy())
 }
 
+// StartProcesses starts running the
 func (t *tracker) StartProcesses() (stoppable.Stoppable, error) {
-	t.mux.Lock()
-	if t.running {
-		t.mux.Unlock()
-		return nil, errors.New(
-			"cannot start health tracker threads, they are already running")
-	}
-	t.running = true
 
-	t.isHealthy = false
-	t.mux.Unlock()
+	atomic.StoreUint32(t.wasHealthy, 0)
 
 	stop := stoppable.NewSingle("health tracker")
 
@@ -139,45 +160,68 @@ func (t *tracker) StartProcesses() (stoppable.Stoppable, error) {
 	return stop, nil
 }
 
-// start starts a long-running thread used to monitor and report on network
+// start begins a long-running thread used to monitor and report on network
 // health.
 func (t *tracker) start(stop *stoppable.Single) {
+
+	// ensures wasHealthy is only set once
+	hasSetWasHealthy := false
+
+	// denotation of the previous state in order to catch state changes
+	lastState := false
+
+	// flag denoting required exit, allows final signaling
+	quit := false
+
+	//ensured the timeout error is only printed once per timeout period
+	timedOut := true
+
 	for {
-		var heartbeat network.Heartbeat
+
+		/* wait for an event */
 		select {
 		case <-stop.Quit():
-			t.mux.Lock()
-			t.isHealthy = false
-			t.running = false
-			t.mux.Unlock()
+			t.forceUnhealthy()
 
-			t.transmit(false)
-			stop.ToStopped()
+			// flag the quit instead of quitting here so the
+			// joint signaling handler code can be triggered
+			quit = true
 
-			return
-		case heartbeat = <-t.heartbeat:
-			// FIXME: There's no transition to unhealthy here and there needs to
-			//  be after some number of bad polls
-			if healthy(heartbeat) {
-				t.setHealth(true)
-			}
+		case heartbeat := <-t.heartbeat:
+			t.updateHealth(heartbeat.HasWaitingRound, heartbeat.IsRoundComplete)
+			timedOut = false
 		case <-time.After(t.timeout):
-			if !t.isHealthy {
-				jww.WARN.Printf("Network health tracker timed out, network " +
-					"is no longer healthy...")
+			if !timedOut {
+				jww.ERROR.Printf("Network health tracker timed out, network " +
+					"is no longer healthy, follower likely has stopped...")
 			}
-			t.setHealth(false)
+			timedOut = true
+
+			// note: no need to force to unhealthy because by definition the
+			// timestamps will be stale
 		}
+
+		/* handle the state change resulting from an event */
+
+		// send signals if the state has changed
+		newHealthState := t.IsHealthy()
+		if newHealthState != lastState {
+			// set was healthy if we are healthy and it was never set before
+			if newHealthState && !hasSetWasHealthy {
+				atomic.StoreUint32(t.wasHealthy, 1)
+				hasSetWasHealthy = true
 			}
+
+			//trigger downstream events
+			t.callback(newHealthState)
+
+			lastState = newHealthState
 		}
 
-func (t *tracker) transmit(health bool) {
-	// Run all listening functions
-	for _, f := range t.funcs {
-		go f(health)
+		// quit if required to quit
+		if quit {
+			stop.ToStopped()
+			return
 		}
 	}
-
-func healthy(a network.Heartbeat) bool {
-	return a.IsRoundComplete
 }

cmix/params.go

@@ -94,7 +94,7 @@ func GetDefaultParams() Params {
 		TrackNetworkPeriod:        100 * time.Millisecond,
 		MaxCheckedRounds:          500,
 		RegNodesBufferLen:         1000,
-		NetworkHealthTimeout:      30 * time.Second,
+		NetworkHealthTimeout:      15 * time.Second,
 		ParallelNodeRegistrations: 20,
 		KnownRoundsThreshold:      1500, // 5 rounds/sec * 60 sec/min * 5 min
 		FastPolling:               true,

go.mod

@@ -14,13 +14,13 @@ require (
 	github.com/spf13/viper v1.12.0
 	github.com/stretchr/testify v1.8.0
 	gitlab.com/elixxir/bloomfilter v0.0.0-20211222005329-7d931ceead6f
-	gitlab.com/elixxir/comms v0.0.4-0.20221005205938-10f2defa5b33
+	gitlab.com/elixxir/comms v0.0.4-0.20221010233602-6ed8c94ddac0
 	gitlab.com/elixxir/crypto v0.0.7-0.20221003185354-b091598d2322
 	gitlab.com/elixxir/ekv v0.2.1
 	gitlab.com/elixxir/primitives v0.0.3-0.20220901220638-1acc75fabdc6
 	gitlab.com/xx_network/comms v0.0.4-0.20221005205845-b34d538ffd85
 	gitlab.com/xx_network/crypto v0.0.5-0.20220913213008-98764f5b3287
-	gitlab.com/xx_network/primitives v0.0.4-0.20220809193445-9fc0a5209548
+	gitlab.com/xx_network/primitives v0.0.4-0.20221010192422-3479f09b7769
 	go.uber.org/ratelimit v0.2.0
 	golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa
 	golang.org/x/net v0.0.0-20220822230855-b0a4917ee28c

go.sum

@@ -635,6 +635,8 @@ gitlab.com/elixxir/comms v0.0.4-0.20220916185715-f1e9a5eda939 h1:+VRx2ULHKs040bB
 gitlab.com/elixxir/comms v0.0.4-0.20220916185715-f1e9a5eda939/go.mod h1:AO6XkMhaHJW8eXlgL5m3UUcJqsSP8F5Wm1GX+wyq/rw=
 gitlab.com/elixxir/comms v0.0.4-0.20221005205938-10f2defa5b33 h1:mtn/b+/+cMoZNSEo6293U48uqz+aE0si90mPwlhh08w=
 gitlab.com/elixxir/comms v0.0.4-0.20221005205938-10f2defa5b33/go.mod h1:oRteMH+R5t1j/FZ+KJJnZUcqJO2sLXnWksN5HPkZUIo=
+gitlab.com/elixxir/comms v0.0.4-0.20221010233602-6ed8c94ddac0 h1:Z8VcCdfmA1AHlGdPe/L8QSGhbjSW2NyCNrDxzByfuqI=
+gitlab.com/elixxir/comms v0.0.4-0.20221010233602-6ed8c94ddac0/go.mod h1:oRteMH+R5t1j/FZ+KJJnZUcqJO2sLXnWksN5HPkZUIo=
 gitlab.com/elixxir/crypto v0.0.0-20200804182833-984246dea2c4/go.mod h1:ucm9SFKJo+K0N2GwRRpaNr+tKXMIOVWzmyUD0SbOu2c=
 gitlab.com/elixxir/crypto v0.0.3/go.mod h1:ZNgBOblhYToR4m8tj4cMvJ9UsJAUKq+p0gCp07WQmhA=
 gitlab.com/elixxir/crypto v0.0.7-0.20220913220142-ab0771bad0af/go.mod h1:QF8SzsrYh9Elip9EUYUDAhPjqO9DGrrrQxYHvn+VXok=
@@ -663,6 +665,7 @@ gitlab.com/xx_network/comms v0.0.4-0.20220913215811-c4bf83b27de3 h1:7mReTvEUVoI5
 gitlab.com/xx_network/comms v0.0.4-0.20220913215811-c4bf83b27de3/go.mod h1:E2QKOKyPKLRjLUwMxgZpTKueEsHDEqshfqOHJ54ttxU=
 gitlab.com/xx_network/comms v0.0.4-0.20220916185248-8a984b8594de h1:44VKuVgT6X1l+MX8/oNmYORA+pa4nkOWV8hYxi4SCzc=
 gitlab.com/xx_network/comms v0.0.4-0.20220916185248-8a984b8594de/go.mod h1:E2QKOKyPKLRjLUwMxgZpTKueEsHDEqshfqOHJ54ttxU=
+gitlab.com/xx_network/comms v0.0.4-0.20221005205845-b34d538ffd85 h1:bX2IYFnEbWTNGhZHfzHME19pkfD4Q7oTxFGI70PM2PM=
 gitlab.com/xx_network/comms v0.0.4-0.20221005205845-b34d538ffd85/go.mod h1:E2QKOKyPKLRjLUwMxgZpTKueEsHDEqshfqOHJ54ttxU=
 gitlab.com/xx_network/crypto v0.0.3/go.mod h1:DF2HYvvCw9wkBybXcXAgQMzX+MiGbFPjwt3t17VRqRE=
 gitlab.com/xx_network/crypto v0.0.4/go.mod h1:+lcQEy+Th4eswFgQDwT0EXKp4AXrlubxalwQFH5O0Mk=
@@ -674,6 +677,8 @@ gitlab.com/xx_network/primitives v0.0.2/go.mod h1:cs0QlFpdMDI6lAo61lDRH2JZz+3aVk
 gitlab.com/xx_network/primitives v0.0.4-0.20220222211843-901fa4a2d72b/go.mod h1:9imZHvYwNFobxueSvVtHneZLk9wTK7HQTzxPm+zhFhE=
 gitlab.com/xx_network/primitives v0.0.4-0.20220809193445-9fc0a5209548 h1:6orM1nSt9H/uN/oqbF8RYmwZ/OuWU6h79Fp+tWeHIPY=
 gitlab.com/xx_network/primitives v0.0.4-0.20220809193445-9fc0a5209548/go.mod h1:AXVVFt7dDAeIUpOGPiStCcUIKsBXLWbmV/BgZ4T+tOo=
+gitlab.com/xx_network/primitives v0.0.4-0.20221010192422-3479f09b7769 h1:BBYSog3VSKudey4rhWnCw54PcGcD4YRjTrnwzOhJ/GE=
+gitlab.com/xx_network/primitives v0.0.4-0.20221010192422-3479f09b7769/go.mod h1:AXVVFt7dDAeIUpOGPiStCcUIKsBXLWbmV/BgZ4T+tOo=
 gitlab.com/xx_network/ring v0.0.3-0.20220222211904-da613960ad93 h1:eJZrXqHsMmmejEPWw8gNAt0I8CGAMNO/7C339Zco3TM=
 gitlab.com/xx_network/ring v0.0.3-0.20220222211904-da613960ad93/go.mod h1:aLzpP2TiZTQut/PVHR40EJAomzugDdHXetbieRClXIM=
 go.etcd.io/bbolt v1.3.3/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=