client/daemon: route liveness probing / listenfunc with retry

Author: snormore

client/doublezerod/cmd/doublezerod/main.go

@@ -183,7 +183,7 @@ func main() {
 					Liveness:   liveness,
 					Limiter:    limiter,
 					Scheduler:  scheduler,
-					ListenFunc: probing.DefaultListenFunc(logger, iface, src),
+					ListenFunc: probing.DefaultListenFuncWithRetry(logger, iface, src),
 					ProbeFunc:  probing.DefaultProbeFunc(logger, iface, *routeProbingProbeTimeout),
 				})
 			} else {

client/doublezerod/internal/probing/config.go

@@ -6,20 +6,10 @@ import (
 	"context"
 	"errors"
 	"log/slog"
-	"time"
 
 	"github.com/malbeclabs/doublezero/client/doublezerod/internal/routing"
 )
 
-var (
-	// defaultListenBackoff is used when no ListenBackoff is provided.
-	defaultListenBackoff = ListenBackoffConfig{
-		Initial:    1 * time.Second,
-		Max:        30 * time.Second,
-		Multiplier: 2,
-	}
-)
-
 // ListenFunc starts a long-lived listener for probe responses or control-plane
 // events. It should block until the context is canceled or an unrecoverable
 // error occurs, returning nil on clean shutdown (ctx canceled).
@@ -31,14 +21,6 @@ type ListenFunc func(context.Context) error
 // the error is due to worker stop (ctx canceled by parent).
 type ProbeFunc func(context.Context, *routing.Route) (ProbeResult, error)
 
-// ListenBackoffConfig controls exponential backoff for ListenFunc retries.
-// Multiplier is applied to the previous backoff duration, capped at Max.
-type ListenBackoffConfig struct {
-	Initial    time.Duration // starting delay before first retry
-	Max        time.Duration // upper bound for backoff delay
-	Multiplier float64       // growth factor per retry, e.g. 2.0 for doubling
-}
-
 // ProbeResult contains probe outcome and basic packet counts.
 // OK should reflect end-to-end success criteria for a single probe wave.
 type ProbeResult struct {
@@ -60,9 +42,6 @@ type Config struct {
 	Scheduler  Scheduler         // scheduler for route probing
 	ListenFunc ListenFunc        // long-lived listener (with retry/backoff)
 	ProbeFunc  ProbeFunc         // per-route probe function
-
-	// Required scalar fields.
-	ListenBackoff ListenBackoffConfig // retry policy for ListenFunc errors; defaulted if zero
 }
 
 // Validate verifies required fields and applies defaults for optional fields.
@@ -94,10 +73,5 @@ func (cfg *Config) Validate() error {
 		return errors.New("probe func is required")
 	}
 
-	// Default values.
-	if cfg.ListenBackoff == (ListenBackoffConfig{}) {
-		cfg.ListenBackoff = defaultListenBackoff
-	}
-
 	return nil
 }

client/doublezerod/internal/probing/default.go

@@ -9,25 +9,50 @@ import (
 	"net"
 	"time"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/malbeclabs/doublezero/client/doublezerod/internal/routing"
 	"github.com/malbeclabs/doublezero/tools/uping/pkg/uping"
 	promprobing "github.com/prometheus-community/pro-bing"
 )
 
-// DefaultListenFunc returns a ListenFunc that starts an ICMP listener
-// using the uping package bound to the given interface and source IP.
+// DefaultListenFunc returns a ListenFunc that starts an ICMP listener bound to iface/src.
 // It blocks until the context is canceled or a fatal error occurs.
 func DefaultListenFunc(log *slog.Logger, iface string, src net.IP) ListenFunc {
 	return func(ctx context.Context) error {
-		listener, err := uping.NewListener(uping.ListenerConfig{
+		l, err := uping.NewListener(uping.ListenerConfig{
 			Logger:    log,
 			Interface: iface,
 			IP:        src,
 		})
 		if err != nil {
 			return err
 		}
-		return listener.Listen(ctx)
+		return l.Listen(ctx)
+	}
+}
+func DefaultListenFuncWithRetry(log *slog.Logger, iface string, src net.IP, opts ...backoff.ExponentialBackOffOpts) ListenFunc {
+	base := DefaultListenFunc(log, iface, src)
+
+	opts = append([]backoff.ExponentialBackOffOpts{
+		backoff.WithInitialInterval(100 * time.Millisecond),
+		backoff.WithMultiplier(2.0),
+		backoff.WithMaxInterval(5 * time.Second),
+		backoff.WithMaxElapsedTime(1 * time.Minute), // stop retrying after a minute…
+		backoff.WithRandomizationFactor(0),          // deterministic (no jitter)
+	}, opts...)
+	return func(ctx context.Context) error {
+		b := backoff.NewExponentialBackOff(opts...)
+
+		bo := backoff.WithContext(b, ctx) // …and also stop on ctx cancel/timeout
+
+		op := func() error {
+			err := base(ctx)
+			// If you detect a non-retryable error, wrap it:
+			// if errors.Is(err, uping.ErrInvalidInterface) { return backoff.Permanent(err) }
+			return err
+		}
+
+		return backoff.Retry(op, bo)
 	}
 }
 

client/doublezerod/internal/probing/main_test.go

@@ -90,11 +90,6 @@ func newTestConfig(t *testing.T, mutate func(*Config)) *Config {
 			time.Sleep(1 * time.Millisecond)
 			return ProbeResult{OK: true, Sent: 1, Received: 1}, nil
 		},
-		ListenBackoff: ListenBackoffConfig{
-			Initial:    10 * time.Millisecond,
-			Max:        100 * time.Millisecond,
-			Multiplier: 2,
-		},
 	}
 	if mutate != nil {
 		mutate(&cfg)

client/doublezerod/internal/probing/worker.go

@@ -41,14 +41,14 @@ func newWorker(log *slog.Logger, cfg *Config, store *routeStore) *probingWorker
 // the provided context is canceled or Stop is called. Safe to call concurrently
 // with IsRunning/Stop.
 func (w *probingWorker) Start(ctx context.Context) {
-	if w.IsRunning() {
+	if !w.running.CompareAndSwap(false, true) {
 		return
 	}
 	ctx, cancel := context.WithCancel(ctx)
 	w.cancelMu.Lock()
 	w.cancel = cancel
 	w.cancelMu.Unlock()
-	w.running.Store(true)
+
 	w.wg.Add(1)
 	go func() {
 		defer w.wg.Done()
@@ -87,11 +87,21 @@ func (w *probingWorker) Run(ctx context.Context) {
 		"scheduler", w.cfg.Scheduler.String(),
 	)
 
-	// Listener runs in parallel and is retried with backoff on failure.
+	// Listener runs in parallel.
 	w.wg.Add(1)
 	go func() {
 		defer w.wg.Done()
-		w.listen(ctx)
+		err := w.cfg.ListenFunc(ctx)
+		if err != nil {
+			w.log.Error("listener error", "error", err)
+
+			// Cancel the worker to stop the run loop.
+			w.cancelMu.Lock()
+			if w.cancel != nil {
+				w.cancel()
+			}
+			w.cancelMu.Unlock()
+		}
 	}()
 
 	// Single reusable timer; we re-arm it whenever the earliest due changes.
@@ -222,50 +232,6 @@ func (w *probingWorker) runProbe(parent context.Context, rk RouteKey, mr managed
 	w.applyProbeResult(&mr, outcome.OK)
 }
 
-// listen runs cfg.ListenFunc until it returns nil or ctx is canceled, retrying
-// with exponential backoff on error. Backoff sleeps are ctx-cancelable.
-// Contract: ListenFunc should return nil on a clean, permanent exit; transient
-// failures should be reported as errors to trigger backoff/retry.
-func (w *probingWorker) listen(ctx context.Context) {
-	backoff := w.cfg.ListenBackoff
-	attempt := 0
-	for {
-		if ctx.Err() != nil {
-			return
-		}
-
-		if err := w.cfg.ListenFunc(ctx); err == nil {
-			// Listener exited cleanly; we’re done.
-			return
-		} else {
-			w.log.Error("listener error", "error", err)
-		}
-
-		// Calculate the backoff duration.
-		attempt++
-		d := backoff.Initial
-		for i := 1; i < attempt; i++ {
-			d = time.Duration(float64(d) * backoff.Multiplier)
-			if d > backoff.Max {
-				d = backoff.Max
-				break
-			}
-		}
-
-		// Cancelable sleep between retries.
-		t := time.NewTimer(d)
-		select {
-		case <-t.C:
-			// Backoff timer fired; retry.
-		case <-ctx.Done():
-			if !t.Stop() {
-				<-t.C
-			}
-			return
-		}
-	}
-}
-
 // applyProbeResult updates liveness state and reconciles kernel routes:
 //
 //	Up   -> Netlink.RouteAdd
@@ -309,8 +275,7 @@ func (w *probingWorker) applyProbeResult(mr *managedRoute, ok bool) {
 		// No kernel operation required.
 	}
 
-	// Persist the updated snapshot back into the store.
-	w.store.Set(key, cur)
+	// No need to store.Set since we mutated the managedRoute in place.
 }
 
 // validateRoute enforces IPv4-only Src/Dst/NextHop. This worker currently

client/doublezerod/internal/probing/worker_test.go

@@ -330,57 +330,6 @@ func TestProbing_Worker_ContextCancelIsNoop(t *testing.T) {
 	require.True(t, hasRouteLiveness(w, r, LivenessStatusDown, 0, 0))
 }
 
-func TestProbing_Worker_ListenRetry_UntilContextDone(t *testing.T) {
-	t.Parallel()
-
-	fourthStarted := make(chan struct{}, 1)
-	var listenCalls atomic.Int64
-
-	sched := newFakeScheduler()
-	cfg := newTestConfig(t, func(c *Config) {
-		c.Liveness = seqPolicy([]LivenessTransition{LivenessTransitionNoChange})
-		c.ProbeFunc = func(context.Context, *routing.Route) (ProbeResult, error) { return ProbeResult{OK: true}, nil }
-		c.ListenFunc = func(ctx context.Context) error {
-			n := listenCalls.Add(1)
-			if n <= 3 {
-				return errors.New("synthetic listen error")
-			}
-			select {
-			case fourthStarted <- struct{}{}:
-			default:
-			}
-			<-ctx.Done()
-			return nil
-		}
-		c.ListenBackoff = ListenBackoffConfig{Initial: time.Millisecond, Max: time.Millisecond, Multiplier: 1}
-		c.Limiter, _ = NewSemaphoreLimiter(2)
-		c.Scheduler = sched
-	})
-
-	w := newWorker(cfg.Logger, cfg, newRouteStore())
-	r := newTestRouteWithDst(net.IPv4(10, 0, 5, 10))
-	rk := newRouteKey(r)
-	w.store.Set(rk, managedRoute{route: r, liveness: cfg.Liveness.NewTracker()})
-	sched.Add(rk, time.Now())
-
-	w.Start(cfg.Context)
-	t.Cleanup(w.Stop)
-
-	requireEventuallyDump(t, func() bool { return w.IsRunning() }, 2*time.Second, 10*time.Millisecond, "worker did not start")
-
-	// Wait until the 4th listen attempt has occurred
-	requireEventuallyDump(t, func() bool {
-		select {
-		case <-fourthStarted:
-			return true
-		default:
-			return false
-		}
-	}, 5*time.Second, 5*time.Millisecond, "4th listen attempt did not start")
-
-	require.True(t, w.IsRunning())
-}
-
 func TestProbing_Worker_KernelError_DoesNotBlockStateAdvance(t *testing.T) {
 	t.Parallel()