rfc: client route liveness probing / rtt and nowfunc

Author: snormore

client/doublezerod/internal/probing/config.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"errors"
 	"log/slog"
+	"time"
 
 	"github.com/malbeclabs/doublezero/client/doublezerod/internal/routing"
 )
@@ -25,10 +26,14 @@ type ProbeFunc func(context.Context, *routing.Route) (ProbeResult, error)
 // OK should reflect end-to-end success criteria for a single probe wave.
 type ProbeResult struct {
 	OK       bool
-	Sent     int
-	Received int
+	Sent     int           // number of probes sent in the wave
+	Received int           // number of probes received in the wave
+	RTTMean  time.Duration // mean RTT of all probes in the wave
 }
 
+// NowFunc returns the current time in UTC.
+type NowFunc func() time.Time
+
 // Config provides all dependencies and tunables for the probing system.
 // Fields marked “Required” must be set; Validate enforces this and applies
 // defaults where appropriate.
@@ -42,6 +47,7 @@ type Config struct {
 	Scheduler  Scheduler         // scheduler for route probing
 	ListenFunc ListenFunc        // long-lived listener (with retry/backoff)
 	ProbeFunc  ProbeFunc         // per-route probe function
+	NowFunc    NowFunc           // function to get the current time
 }
 
 // Validate verifies required fields and applies defaults for optional fields.
@@ -72,6 +78,11 @@ func (cfg *Config) Validate() error {
 	if cfg.ProbeFunc == nil {
 		return errors.New("probe func is required")
 	}
+	if cfg.NowFunc == nil {
+		cfg.NowFunc = func() time.Time {
+			return time.Now().UTC()
+		}
+	}
 
 	return nil
 }

client/doublezerod/internal/probing/default.go

@@ -83,6 +83,7 @@ func DefaultProbeFunc(log *slog.Logger, iface string, timeout time.Duration) Pro
 			OK:       ok,
 			Sent:     stats.PacketsSent,
 			Received: stats.PacketsRecv,
+			RTTMean:  stats.AvgRtt,
 		}, nil
 	}
 }

client/doublezerod/internal/probing/main_test.go

@@ -90,6 +90,7 @@ func newTestConfig(t *testing.T, mutate func(*Config)) *Config {
 			time.Sleep(1 * time.Millisecond)
 			return ProbeResult{OK: true, Sent: 1, Received: 1}, nil
 		},
+		NowFunc: now,
 	}
 	if mutate != nil {
 		mutate(&cfg)
@@ -293,7 +294,7 @@ func (s *fakeScheduler) Peek() (time.Time, bool) {
 	p := s.wavePending
 	s.mu.Unlock()
 	if p {
-		return time.Now(), true
+		return now(), true
 	}
 	return time.Time{}, false
 }
@@ -569,3 +570,7 @@ func (m *mockLivenessTracker) ConsecutiveOK() uint {
 func (m *mockLivenessTracker) ConsecutiveFail() uint {
 	return m.ConsecutiveFailFunc()
 }
+
+func now() time.Time {
+	return time.Now().UTC()
+}

client/doublezerod/internal/probing/manager.go

@@ -6,7 +6,6 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
-	"time"
 
 	"github.com/malbeclabs/doublezero/client/doublezerod/internal/routing"
 )
@@ -99,7 +98,7 @@ func (m *RouteManager) handleRouteAdd(route *routing.Route) error {
 
 	// Add the route to managed route store.
 	key := newRouteKey(route)
-	now := time.Now()
+	now := m.cfg.NowFunc()
 	m.cfg.Scheduler.Add(key, now)
 	m.store.Set(key, managedRoute{
 		route:    route,

client/doublezerod/internal/probing/scheduler.go

@@ -39,6 +39,7 @@ type IntervalConfig struct {
 	Interval time.Duration // base interval between probes
 	Jitter   time.Duration // max absolute jitter (+/-) applied to the interval/phase anchor
 	Phase    bool          // whether to phase routes deterministically
+	NowFunc  NowFunc       // function to get the current time
 }
 
 // Validate ensures the configuration is usable.
@@ -49,6 +50,11 @@ func (cfg *IntervalConfig) Validate() error {
 	if cfg.Jitter < 0 {
 		return errors.New("jitter must be >= 0")
 	}
+	if cfg.NowFunc == nil {
+		cfg.NowFunc = func() time.Time {
+			return time.Now().UTC()
+		}
+	}
 	return nil
 }
 
@@ -105,7 +111,7 @@ func (s *IntervalScheduler) Add(k RouteKey, base time.Time) {
 		return
 	}
 	seed := hash64(k)
-	due := firstDue(base, s.cfg.Interval, s.cfg.Jitter, s.cfg.Phase, k, seed)
+	due := firstDue(base, s.cfg.Interval, s.cfg.Jitter, s.cfg.Phase, k, seed, s.cfg.NowFunc)
 	s.routes[k] = &schedulerRouteState{seed: seed, nextDue: due}
 	s.maybeSignalLocked()
 }
@@ -250,9 +256,9 @@ func phaseOffset(iv time.Duration, k RouteKey) time.Duration {
 
 // firstDue computes the initial due time for a route given the config.
 // If Phase is enabled, routes are staggered deterministically by key.
-func firstDue(base time.Time, iv time.Duration, jitter time.Duration, phase bool, k RouteKey, seed uint64) time.Time {
+func firstDue(base time.Time, iv time.Duration, jitter time.Duration, phase bool, k RouteKey, seed uint64, nowFunc NowFunc) time.Time {
 	if base.IsZero() {
-		base = time.Now()
+		base = nowFunc()
 	}
 	if phase {
 		d := base.Truncate(iv).Add(phaseOffset(iv, k))

client/doublezerod/internal/probing/worker.go

@@ -143,8 +143,8 @@ func (w *probingWorker) Run(ctx context.Context) {
 	for {
 		// Handle “due now” immediately to avoid Reset(0) races.
 		if next, ok := w.cfg.Scheduler.Peek(); ok {
-			if !next.After(time.Now()) {
-				launchDue(time.Now())
+			if !next.After(w.cfg.NowFunc()) {
+				launchDue(w.cfg.NowFunc())
 				// After processing, loop to re-peek and re-arm.
 				continue
 			}
@@ -164,7 +164,7 @@ func (w *probingWorker) Run(ctx context.Context) {
 			wakeCh = w.cfg.Scheduler.Wake()
 
 		case <-tc:
-			launchDue(time.Now())
+			launchDue(w.cfg.NowFunc())
 		}
 	}
 }
@@ -180,7 +180,7 @@ func (w *probingWorker) runProbe(parent context.Context, rk RouteKey, mr managed
 	ctx, cancel := context.WithTimeout(parent, 10*time.Second)
 	defer cancel()
 
-	outcome := ProbeOutcome{When: time.Now()}
+	outcome := ProbeOutcome{When: w.cfg.NowFunc()}
 	defer func() {
 		if r := recover(); r != nil {
 			outcome.OK = false
@@ -197,14 +197,11 @@ func (w *probingWorker) runProbe(parent context.Context, rk RouteKey, mr managed
 	}
 	defer rel()
 
-	start := time.Now()
 	res, err := w.cfg.ProbeFunc(ctx, mr.route)
-	outcome.When = time.Now()
+	outcome.When = w.cfg.NowFunc()
 	outcome.OK = (err == nil && res.OK)
 	outcome.Err = err
-	if err == nil {
-		outcome.RTT = outcome.When.Sub(start)
-	}
+	outcome.RTT = res.RTTMean
 
 	if ctx.Err() != nil {
 		return

client/doublezerod/internal/probing/worker_test.go

@@ -57,8 +57,8 @@ func TestProbing_Worker_SuccessThenFailure_TransitionsAndKernel(t *testing.T) {
 	k1, k2 := newRouteKey(r1), newRouteKey(r2)
 	w.store.Set(k1, managedRoute{route: r1, liveness: cfg.Liveness.NewTracker()})
 	w.store.Set(k2, managedRoute{route: r2, liveness: cfg.Liveness.NewTracker()})
-	sched.Add(k1, time.Now())
-	sched.Add(k2, time.Now())
+	sched.Add(k1, now())
+	sched.Add(k2, now())
 
 	w.Start(cfg.Context)
 	t.Cleanup(w.Stop)
@@ -110,8 +110,8 @@ func TestProbing_Worker_ErrorCountsAsFailure(t *testing.T) {
 	r2 := newTestRouteWithDst(net.IPv4(10, 0, 1, 2))
 	w.store.Set(newRouteKey(r1), managedRoute{route: r1, liveness: cfg.Liveness.NewTracker()})
 	w.store.Set(newRouteKey(r2), managedRoute{route: r2, liveness: cfg.Liveness.NewTracker()})
-	sched.Add(newRouteKey(r1), time.Now())
-	sched.Add(newRouteKey(r2), time.Now())
+	sched.Add(newRouteKey(r1), now())
+	sched.Add(newRouteKey(r2), now())
 
 	w.Start(cfg.Context)
 	t.Cleanup(w.Stop)
@@ -182,7 +182,7 @@ func TestProbing_Worker_RespectsLimiterConcurrency(t *testing.T) {
 		r := newTestRouteWithDst(net.IPv4(10, 0, 2, byte(i+1)))
 		rk := newRouteKey(r)
 		w.store.Set(rk, managedRoute{route: r, liveness: cfg.Liveness.NewTracker()})
-		sched.Add(rk, time.Now())
+		sched.Add(rk, now())
 	}
 
 	w.Start(cfg.Context)
@@ -239,7 +239,7 @@ func TestProbing_Worker_IgnoresResultIfRouteRemoved(t *testing.T) {
 	r := newTestRouteWithDst(net.IPv4(10, 0, 3, 1))
 	key := newRouteKey(r)
 	w.store.Set(key, managedRoute{route: r, liveness: cfg.Liveness.NewTracker()})
-	sched.Add(key, time.Now())
+	sched.Add(key, now())
 
 	w.Start(cfg.Context)
 	t.Cleanup(w.Stop)
@@ -284,7 +284,7 @@ func TestProbing_Worker_ContextCancelIsNoop(t *testing.T) {
 	r := newTestRouteWithDst(net.IPv4(10, 0, 4, 1))
 	key := newRouteKey(r)
 	w.store.Set(key, managedRoute{route: r, liveness: cfg.Liveness.NewTracker()})
-	sched.Add(key, time.Now())
+	sched.Add(key, now())
 
 	ctx, cancel := context.WithCancel(cfg.Context)
 	cfg.Context = ctx
@@ -318,7 +318,7 @@ func TestProbing_Worker_KernelError_DoesNotBlockStateAdvance(t *testing.T) {
 	r := newTestRouteWithDst(net.IPv4(10, 0, 6, 1))
 	key := newRouteKey(r)
 	w.store.Set(key, managedRoute{route: r, liveness: cfg.Liveness.NewTracker()})
-	sched.Add(key, time.Now())
+	sched.Add(key, now())
 
 	w.Start(cfg.Context)
 	defer w.Stop()