From 8ee9980b9629d392f201caee84e308db62b515c2 Mon Sep 17 00:00:00 2001 From: Kristoffer Dalby Date: Wed, 28 Oct 2026 18:27:54 +0000 Subject: [PATCH] mapper: count only the batcher's goroutines in lifecycle tests runtime.NumGoroutine deltas moved when goroutines left by earlier tests exited mid-test. Label the batcher's goroutines and count those. --- hscontrol/mapper/batcher_concurrency_test.go | 43 +++-------- hscontrol/mapper/batcher_unit_test.go | 76 +++++++++++++------- 2 files changed, 60 insertions(+), 59 deletions(-) diff --git a/hscontrol/mapper/batcher_concurrency_test.go b/hscontrol/mapper/batcher_concurrency_test.go index e84ba08a8..2985c64b2 100644 --- a/hscontrol/mapper/batcher_concurrency_test.go +++ b/hscontrol/mapper/batcher_concurrency_test.go @@ -12,7 +12,6 @@ package mapper import ( "fmt" - "runtime" "sync" "sync/atomic" "testing" @@ -1150,52 +1149,30 @@ func TestBug6_StartCalledMultipleTimes_GoroutineLeak(t *testing.T) { lb := setupLightweightBatcher(t, 3, 10) lb.b.workers = 2 - goroutinesBefore := runtime.NumGoroutine() - // Call Start() once - this should launch (workers + 1) goroutines // (1 for doWork + workers for worker()) - lb.b.Start() - time.Sleep(50 * time.Millisecond) //nolint:forbidigo // concurrency test coordination - - goroutinesAfterFirst := runtime.NumGoroutine() - firstStartDelta := goroutinesAfterFirst - goroutinesBefore - t.Logf("goroutines: before=%d, after_first_Start=%d, delta=%d", - goroutinesBefore, goroutinesAfterFirst, firstStartDelta) + startLabelled(lb.b, t.Name()) + require.Eventually(t, func() bool { + return batcherGoroutines(t.Name()) == 3 + }, time.Second, 5*time.Millisecond, "first Start() should run doWork + 2 workers") // Call Start() again - this SHOULD be a no-op // BUG: it creates a NEW done channel (orphaning goroutines listening on the old one) // and launches another doWork()+workers set - lb.b.Start() + startLabelled(lb.b, t.Name()) + startLabelled(lb.b, t.Name()) time.Sleep(50 * time.Millisecond) //nolint:forbidigo // concurrency test coordination - goroutinesAfterSecond := runtime.NumGoroutine() - secondStartDelta := goroutinesAfterSecond - goroutinesAfterFirst - t.Logf("goroutines: after_second_Start=%d, delta=%d (should be 0)", - goroutinesAfterSecond, secondStartDelta) + extra := batcherGoroutines(t.Name()) - 3 - // Call Start() a third time - lb.b.Start() - time.Sleep(50 * time.Millisecond) //nolint:forbidigo // concurrency test coordination - - goroutinesAfterThird := runtime.NumGoroutine() - thirdStartDelta := goroutinesAfterThird - goroutinesAfterSecond - t.Logf("goroutines: after_third_Start=%d, delta=%d (should be 0)", - goroutinesAfterThird, thirdStartDelta) - - // Close() only closes the LAST done channel, leaving earlier goroutines leaked lb.b.Close() - time.Sleep(100 * time.Millisecond) //nolint:forbidigo // concurrency test coordination - - goroutinesAfterClose := runtime.NumGoroutine() - t.Logf("goroutines after Close: %d (leaked: %d)", - goroutinesAfterClose, goroutinesAfterClose-goroutinesBefore) // Second Start() should NOT have created new goroutines - assert.Zero(t, secondStartDelta, - "BUG #6: second Start() call leaked %d goroutines. "+ + assert.Zero(t, extra, + "BUG #6: repeated Start() calls leaked %d goroutines. "+ "Start() has no idempotency guard, each call spawns new goroutines. "+ "Fix: add sync.Once or atomic.Bool to prevent multiple Start() calls", - secondStartDelta) + extra) } // TestBug7_CleanupOfflineNodes_PendingChangesCleanedStructurally verifies that diff --git a/hscontrol/mapper/batcher_unit_test.go b/hscontrol/mapper/batcher_unit_test.go index c02f9c4a6..f50b08574 100644 --- a/hscontrol/mapper/batcher_unit_test.go +++ b/hscontrol/mapper/batcher_unit_test.go @@ -5,9 +5,13 @@ package mapper // handleNodeChange. import ( + "bytes" + "context" "errors" "fmt" - "runtime" + "runtime/pprof" + "strconv" + "strings" "sync" "sync/atomic" "testing" @@ -1450,28 +1454,18 @@ func TestMultiChannelSend_ConcurrentRemoveAndSend(t *testing.T) { func TestBatcher_CloseWaitsForWorkers(t *testing.T) { b := NewBatcher(50*time.Millisecond, 4, nil) - goroutinesBefore := runtime.NumGoroutine() + startLabelled(b, t.Name()) - b.Start() - - // Give workers time to start. - time.Sleep(20 * time.Millisecond) //nolint:forbidigo // test timing - - goroutinesDuring := runtime.NumGoroutine() - - // We expect at least 5 new goroutines: 1 doWork + 4 workers. - assert.GreaterOrEqual(t, goroutinesDuring-goroutinesBefore, 5, - "expected doWork + 4 workers to be running") + // 1 doWork + 4 workers. + require.Eventually(t, func() bool { + return batcherGoroutines(t.Name()) == 5 + }, time.Second, 5*time.Millisecond, "expected doWork + 4 workers to be running") // Close should block until all workers have exited. b.Close() - // After Close returns, goroutines should have dropped back. - // Allow a small margin for runtime goroutines. - goroutinesAfter := runtime.NumGoroutine() - assert.InDelta(t, goroutinesBefore, goroutinesAfter, 3, - "goroutines should return to baseline after Close(); before=%d after=%d", - goroutinesBefore, goroutinesAfter) + assert.Zero(t, batcherGoroutines(t.Name()), + "batcher goroutines still running after Close()") } // TestBatcher_CloseThenStartIsNoop verifies the lifecycle contract: @@ -1480,22 +1474,52 @@ func TestBatcher_CloseWaitsForWorkers(t *testing.T) { func TestBatcher_CloseThenStartIsNoop(t *testing.T) { b := NewBatcher(50*time.Millisecond, 2, nil) - b.Start() + startLabelled(b, t.Name()) b.Close() - goroutinesBefore := runtime.NumGoroutine() - // Second Start should be a no-op because started is already true. - b.Start() + startLabelled(b, t.Name()) // Allow a moment for any hypothetical goroutine to appear. <-time.After(10 * time.Millisecond) //nolint:forbidigo // test timing - goroutinesAfter := runtime.NumGoroutine() + assert.Zero(t, batcherGoroutines(t.Name()), + "Start() after Close() should not spawn new goroutines") +} - assert.InDelta(t, goroutinesBefore, goroutinesAfter, 1, - "Start() after Close() should not spawn new goroutines; before=%d after=%d", - goroutinesBefore, goroutinesAfter) +// startLabelled starts b under a pprof label that its goroutines inherit, so +// batcherGoroutines counts them and nothing else. +func startLabelled(b *Batcher, label string) { + pprof.Do(context.Background(), pprof.Labels("batcher", label), func(context.Context) { + b.Start() + }) +} + +// batcherGoroutines counts goroutines in Batcher code started by startLabelled +// with label. A runtime.NumGoroutine delta also moves when goroutines other +// tests left behind exit. +func batcherGoroutines(label string) int { + var buf bytes.Buffer + + _ = pprof.Lookup("goroutine").WriteTo(&buf, 1) + + want := fmt.Sprintf(`"batcher":%q`, label) + total := 0 + + for group := range strings.SplitSeq(buf.String(), "\n\n") { + if !strings.Contains(group, want) || !strings.Contains(group, ".(*Batcher).") { + continue + } + + for line := range strings.Lines(group) { + if n, _, ok := strings.Cut(line, " @ "); ok { + c, _ := strconv.Atoi(n) + total += c + } + } + } + + return total } // TestBatcher_CloseStopsTicker verifies that Close() stops the internal