client/daemon: route liveness probing / tests cleanup

Author: snormore

client/doublezerod/internal/probing/limiter_test.go

@@ -26,7 +26,6 @@ func TestProbing_SemaphoreLimiter_BasicAcquireRelease(t *testing.T) {
 	l, err := NewSemaphoreLimiter(2)
 	require.NoError(t, err)
 
-	// Acquire twice should succeed immediately.
 	rel1, ok := l.Acquire(context.Background())
 	require.True(t, ok)
 	require.NotNil(t, rel1)
@@ -35,7 +34,6 @@ func TestProbing_SemaphoreLimiter_BasicAcquireRelease(t *testing.T) {
 	require.True(t, ok)
 	require.NotNil(t, rel2)
 
-	// Third acquire should block until a release happens.
 	acquired := make(chan struct{})
 	var rel3 func()
 
@@ -47,25 +45,30 @@ func TestProbing_SemaphoreLimiter_BasicAcquireRelease(t *testing.T) {
 		}
 	}()
 
-	// Ensure it does NOT acquire within a short window (still blocked).
+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+	defer cancel()
+	done := make(chan struct{})
+	go func() {
+		select {
+		case <-acquired:
+		case <-ctx.Done():
+		}
+		close(done)
+	}()
 	select {
-	case <-acquired:
-		t.Fatalf("third Acquire should be blocked until a release")
-	case <-time.After(50 * time.Millisecond):
-		// expected: still blocked
+	case <-done:
+		require.NotNil(t, ctx.Err())
+	default:
 	}
 
-	// Release one permit; now the goroutine should acquire shortly.
 	rel1()
 
 	select {
 	case <-acquired:
-		// got the third permit
-	case <-time.After(250 * time.Millisecond):
+	case <-time.After(500 * time.Millisecond):
 		t.Fatalf("third Acquire did not succeed after a release")
 	}
 
-	// Cleanup remaining permits
 	rel2()
 	if rel3 != nil {
 		rel3()
@@ -78,33 +81,30 @@ func TestProbing_SemaphoreLimiter_CancelWhileBlocked(t *testing.T) {
 	l, err := NewSemaphoreLimiter(1)
 	require.NoError(t, err)
 
-	// Fill the single permit.
 	rel, ok := l.Acquire(context.Background())
 	require.True(t, ok)
 	require.NotNil(t, rel)
 
-	// Attempt another acquire with a short timeout; it should fail when context expires.
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
 	defer cancel()
 
-	done := make(chan struct{})
+	got := make(chan struct{})
 	var gotOK bool
 	var gotRel func()
 
 	go func() {
-		defer close(done)
+		defer close(got)
 		gotRel, gotOK = l.Acquire(ctx)
 	}()
 
 	select {
-	case <-done:
-		require.False(t, gotOK, "Acquire should return ok=false when context times out")
+	case <-got:
+		require.False(t, gotOK)
 		require.Nil(t, gotRel)
 	case <-time.After(200 * time.Millisecond):
 		t.Fatalf("Acquire did not return after context timeout")
 	}
 
-	// Release original permit to avoid leaks.
 	rel()
 }
 
@@ -123,21 +123,21 @@ func TestProbing_SemaphoreLimiter_SequentialAcquireReleaseCycles(t *testing.T) {
 		require.True(t, ok)
 		require.NotNil(t, relB)
 
-		// A third should block; verify it doesn’t acquire within 20ms.
-		acquired := make(chan struct{})
+		blockCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+		defer cancel()
+		acqDone := make(chan struct{})
+		var acqOK bool
 		go func() {
-			if r, ok := l.Acquire(context.Background()); ok {
-				r() // immediately release if we ever acquire (we shouldn't before the test releases)
-				close(acquired)
-			}
+			defer close(acqDone)
+			_, acqOK = l.Acquire(blockCtx)
 		}()
 		select {
-		case <-acquired:
-			t.Fatalf("Acquire should not succeed before a release (iteration %d)", i)
-		case <-time.After(20 * time.Millisecond):
+		case <-acqDone:
+			require.False(t, acqOK, "Acquire should not succeed before a release (iteration %d)", i)
+		case <-time.After(200 * time.Millisecond):
+			t.Fatalf("Acquire did not return within timeout (iteration %d)", i)
 		}
 
-		// Release both and ensure we can acquire twice again.
 		relA()
 		relB()
 

client/doublezerod/internal/probing/main_test.go

@@ -342,42 +342,6 @@ func (s *fakeScheduler) Complete(k RouteKey, _ ProbeOutcome) {
 	s.mu.Unlock()
 }
 
-//nolint:unused
-func waitForProbe(t *testing.T, ch <-chan struct{}, sched *fakeScheduler, d time.Duration) {
-	t.Helper()
-	deadline := time.After(d)
-	tick := time.NewTicker(5 * time.Millisecond)
-	defer tick.Stop()
-
-	for {
-		select {
-		case <-ch:
-			return
-		case <-tick.C:
-			sched.Trigger() // keep nudging the worker
-		case <-deadline:
-			t.Fatalf("probe did not start within %v", d)
-		}
-	}
-}
-
-func startNudger(t *testing.T, sched *fakeScheduler, every time.Duration) (stop func()) {
-	done := make(chan struct{})
-	go func() {
-		tk := time.NewTicker(every)
-		defer tk.Stop()
-		for {
-			select {
-			case <-tk.C:
-				sched.Trigger()
-			case <-done:
-				return
-			}
-		}
-	}()
-	return func() { close(done) }
-}
-
 func waitEdge(t *testing.T, ch <-chan struct{}, d time.Duration, msg string) {
 	t.Helper()
 	select {
@@ -419,24 +383,6 @@ want:
 	t.Logf("\n==== BEGIN FILTERED GOROUTINES ====\n%s==== END FILTERED GOROUTINES ====\n", out.String())
 }
 
-//nolint:unused
-func requireEventuallyDump(t *testing.T, cond func() bool, wait, tick time.Duration, why string) {
-	deadline := time.Now().Add(wait)
-	for time.Now().Before(deadline) {
-		if cond() {
-			return
-		}
-		time.Sleep(tick)
-	}
-	// last chance
-	if cond() {
-		return
-	}
-	dumpGoroutines(t)
-	dumpGoroutinesFiltered(t, "/internal/probing", "probingWorker", "fakeScheduler")
-	t.Fatalf("eventually failed: %s (wait=%v tick=%v)", why, wait, tick)
-}
-
 type memoryNetlinker struct {
 	routesByDst map[string][]*routing.Route
 

client/doublezerod/internal/probing/manager_test.go

@@ -14,8 +14,6 @@ import (
 )
 
 func TestProbing_RouteManager_PeerLifecycle_StartsAndStopsWorker_Idempotent(t *testing.T) {
-	t.Parallel()
-
 	cfg := newTestConfig(t, func(c *Config) {
 		c.Limiter, _ = NewSemaphoreLimiter(4)
 		c.Scheduler = newFakeScheduler()
@@ -38,8 +36,6 @@ func TestProbing_RouteManager_PeerLifecycle_StartsAndStopsWorker_Idempotent(t *t
 }
 
 func TestProbing_RouteManager_RouteAdd_WorkerRunning_StoresButNoKernelAdd(t *testing.T) {
-	t.Parallel()
-
 	var addCalls int64
 	cfg := newTestConfig(t, func(c *Config) {
 		c.Limiter, _ = NewSemaphoreLimiter(4)
@@ -61,11 +57,8 @@ func TestProbing_RouteManager_RouteAdd_WorkerRunning_StoresButNoKernelAdd(t *tes
 }
 
 func TestProbing_RouteManager_RouteAdd_WorkerStopped_CallsKernelAdd(t *testing.T) {
-	t.Parallel()
-
 	var addCalls int64
 	cfg := newTestConfig(t, func(c *Config) {
-		// No scheduler/limiter needed since worker won't be started
 		c.Netlink = &MockNetlinker{
 			RouteAddFunc:    func(*routing.Route) error { atomic.AddInt64(&addCalls, 1); return nil },
 			RouteDeleteFunc: func(*routing.Route) error { return nil },
@@ -81,22 +74,18 @@ func TestProbing_RouteManager_RouteAdd_WorkerStopped_CallsKernelAdd(t *testing.T
 }
 
 func TestProbing_RouteManager_RouteAdd_InvalidRoute_Err(t *testing.T) {
-	t.Parallel()
-
 	cfg := newTestConfig(t, func(c *Config) {
 		c.Limiter, _ = NewSemaphoreLimiter(1)
 		c.Scheduler = newFakeScheduler()
 	})
 	m, _ := NewRouteManager(cfg)
 	_ = m.PeerOnEstablished()
-	err := m.RouteAdd(&routing.Route{}) // invalid
+	err := m.RouteAdd(&routing.Route{})
 	require.Error(t, err)
 	require.Equal(t, 0, m.store.Len())
 }
 
 func TestProbing_RouteManager_RouteDelete_WorkerRunning_RemovesAndKernelDelete(t *testing.T) {
-	t.Parallel()
-
 	var delCalls int64
 	cfg := newTestConfig(t, func(c *Config) {
 		c.Limiter, _ = NewSemaphoreLimiter(4)
@@ -118,8 +107,6 @@ func TestProbing_RouteManager_RouteDelete_WorkerRunning_RemovesAndKernelDelete(t
 }
 
 func TestProbing_RouteManager_RouteDelete_WorkerRunning_RouteNotFoundInKernel_DoesNotReturnError(t *testing.T) {
-	t.Parallel()
-
 	var delCalls int64
 	cfg := newTestConfig(t, func(c *Config) {
 		c.Limiter, _ = NewSemaphoreLimiter(4)
@@ -141,8 +128,6 @@ func TestProbing_RouteManager_RouteDelete_WorkerRunning_RouteNotFoundInKernel_Do
 }
 
 func TestProbing_RouteManager_RouteDelete_WorkerStopped_CallsKernelDelete(t *testing.T) {
-	t.Parallel()
-
 	var delCalls int64
 	cfg := newTestConfig(t, func(c *Config) {
 		c.Netlink = &MockNetlinker{
@@ -158,8 +143,6 @@ func TestProbing_RouteManager_RouteDelete_WorkerStopped_CallsKernelDelete(t *tes
 }
 
 func TestProbing_RouteManager_RouteDelete_InvalidRoute_Err(t *testing.T) {
-	t.Parallel()
-
 	cfg := newTestConfig(t, func(c *Config) {
 		c.Limiter, _ = NewSemaphoreLimiter(1)
 		c.Scheduler = newFakeScheduler()
@@ -172,8 +155,6 @@ func TestProbing_RouteManager_RouteDelete_InvalidRoute_Err(t *testing.T) {
 }
 
 func TestProbing_RouteManager_RouteByProtocol_Passthrough(t *testing.T) {
-	t.Parallel()
-
 	r := newTestRouteWithDst(net.IPv4(10, 0, 0, 30))
 
 	cfg := newTestConfig(t, func(c *Config) {
@@ -193,16 +174,12 @@ func TestProbing_RouteManager_RouteByProtocol_Passthrough(t *testing.T) {
 }
 
 func TestProbing_RouteManager_NewRouteManager_ConfigValidateErrorBubbles(t *testing.T) {
-	t.Parallel()
-
 	cfg := newTestConfig(t, func(c *Config) { c.Logger = nil })
 	_, err := NewRouteManager(cfg)
 	require.Error(t, err)
 }
 
-func TestProbing_RouteManager_PeerOnEstablished_StartsWorkerAndProbes(t *testing.T) {
-	t.Parallel()
-
+func TestProbing_RouteManager_PeerOnEstablished_StartsAndProbes(t *testing.T) {
 	probed := make(chan struct{}, 1)
 	sched := newFakeScheduler()
 	cfg := newTestConfig(t, func(c *Config) {
@@ -227,17 +204,14 @@ func TestProbing_RouteManager_PeerOnEstablished_StartsWorkerAndProbes(t *testing
 	r2 := newTestRouteWithDst(net.IPv4(10, 0, 0, 41))
 	require.NoError(t, m.RouteAdd(r2))
 
-	stop := startNudger(t, sched, 5*time.Millisecond)
-	t.Cleanup(stop)
+	sched.Trigger()
 	waitEdge(t, probed, 2*time.Second, "probe did not start")
 
 	require.NoError(t, m.PeerOnClose())
 	require.Equal(t, 0, m.store.Len())
 }
 
 func TestProbing_RouteManager_PeerOnEstablished_ClearsStore(t *testing.T) {
-	t.Parallel()
-
 	cfg := newTestConfig(t, func(c *Config) {
 		c.Limiter, _ = NewSemaphoreLimiter(1)
 		c.Scheduler = newFakeScheduler()
@@ -251,8 +225,6 @@ func TestProbing_RouteManager_PeerOnEstablished_ClearsStore(t *testing.T) {
 }
 
 func TestProbing_RouteManager_PeerOnClose_ClearsStore(t *testing.T) {
-	t.Parallel()
-
 	cfg := newTestConfig(t, func(c *Config) {
 		c.Limiter, _ = NewSemaphoreLimiter(1)
 		c.Scheduler = newFakeScheduler()