From 6dd94a11f53f948c287b7afa3378a19a057b9cbe Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 16 May 2026 15:20:33 +0000 Subject: [PATCH] add concurrent-code seams and hermetic tests for statsd, WaitLoop, Monitor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI runs `go test -race` but until now no test exercised the package's concurrent code paths, so the race detector had nothing to observe. This adds the minimum testability seams to drive each concurrent surface from a hermetic test: - statsd.go: extract a `chan<- string` parameter on `count`, `timer`, `gauge`, `increment`, and `statsdSender`. Public `Count`/`Timer`/ `Gauge`/`Increment`/`StatsdSender` route through the new helpers with the package-level `queue` as the default. Tests inject their own buffered channel. - destinations.go: factor `Monitor`'s loop body into `monitorWithCheck` (one iteration, returns next confidence) and `WaitFor`'s loop into `waitForWithCheck(check, sleep)`. The infinite `for {}` in `Monitor` stays in the public method. - connectivity.go: add `waitLoop(destinations, checks, sleep)` so the goroutine fan-out is driveable with stub checks and a zero sleep. New tests in concurrency_test.go: - TestStatsdCountBlocksWhenQueueFull pins the #11 back-pressure hazard by saturating an injected queue and asserting the next send blocks. Marked `Refs #11 -- flip when fixed` for when non-blocking sends land. - TestStatsdSenderDeliversToUDP runs the sender goroutine against a local UDP listener with M concurrent producers — real producer/consumer exercise under -race. - TestWaitLoopCompletesWhenChecksSucceed drives the goroutine fan-out with a stub check that succeeds on the second attempt, asserts wg.Wait() returns and brackets NumGoroutine to catch leaks. - TestMonitorWithCheckResetsConfidenceOnFailure pins the post-#16 confidence reset: after saturating at 10, one failure snaps the next sleep back to 1 minute. Scope is testability only: this does NOT fix the #11 back-pressure bug (producers still block), the #17 lifecycle gaps (no signal handling, no statsd drain, no panic recovery, no context.Context), or the #18 wait-timeout flag. `see #17` / `see #18` doc comments mark the spots where the substrate work needs to land. Each new test was verified by mutation: temporarily breaking the production code (e.g., making `count` non-blocking, dropping bytes in `statsdSender`, skipping the retry in `waitForWithCheck`, dropping the confidence reset) causes the matching assertion to fail. The tests are not checklist theater. Fixes #49 https://claude.ai/code/session_01WjHPSobuzrRkjwUgjAJWMk --- concurrency_test.go | 230 ++++++++++++++++++++++++++++++++++++++++++++ connectivity.go | 24 ++++- destinations.go | 51 +++++++--- statsd.go | 41 +++++++- 4 files changed, 322 insertions(+), 24 deletions(-) create mode 100644 concurrency_test.go diff --git a/concurrency_test.go b/concurrency_test.go new file mode 100644 index 0000000..fbaa7d8 --- /dev/null +++ b/concurrency_test.go @@ -0,0 +1,230 @@ +package main + +import ( + "net" + "runtime" + "strconv" + "sync" + "testing" + "time" +) + +// This file holds hermetic tests for the concurrent surfaces of the package: +// the statsd producer/consumer goroutines, WaitLoop's per-destination +// goroutines, and the Monitor confidence loop. Each test exercises a seam +// (lowercase variant of the public function) so it can drive the concurrent +// code path synchronously without spinning up production goroutines. +// +// Refs #49 — these tests are the reason -race in CI is no longer a false +// signal: they actually execute the concurrent producer/consumer surface. + +// TestStatsdCountBlocksWhenQueueFull pins the back-pressure hazard documented +// in #11: when StatsdSender cannot drain (statsd down, slow TCP, etc.) the +// 100-slot queue fills up and every metric helper blocks. The test uses an +// injected queue with capacity 1 so it can be saturated cheaply. +// +// Refs #11 -- flip when fixed: once the producers become non-blocking +// (select-with-default + drop_and_count_drops), this test should be updated +// to assert the drop counter increments and Count returns promptly. +func TestStatsdCountBlocksWhenQueueFull(t *testing.T) { + q := make(chan string, 1) + // Saturate the queue. No consumer is running, so the next send will block. + count(q, "preload", 1, []string{"t:v"}) + + done := make(chan struct{}) + go func() { + count(q, "should-block", 1, []string{"t:v"}) + close(done) + }() + + select { + case <-done: + t.Fatal("count(q, ...) returned while queue was full; want it to block " + + "(this is the bug from #11; flip the assertion when #11 is fixed)") + case <-time.After(50 * time.Millisecond): + // Expected: the send is blocked. Drain the preload so the goroutine + // can complete and not leak past the test. + } + + // Drain one slot so the blocked send can complete, then wait for the + // goroutine to finish so we don't leak it across tests. + <-q + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("blocked count goroutine did not complete after draining a slot") + } + // Drain the second message the goroutine enqueued. + <-q +} + +// TestStatsdSenderDeliversToUDP runs the statsd consumer goroutine end-to-end +// against a local UDP listener. It exercises the consumer half of the queue +// (the goroutine started from main as `go StatsdSender(config)`) and the +// wire-format helpers under -race conditions: M producer goroutines push N +// metrics each through the injected queue while the sender reads them. +func TestStatsdSenderDeliversToUDP(t *testing.T) { + pc, err := net.ListenPacket("udp", "127.0.0.1:0") + if err != nil { + t.Fatalf("ListenPacket: %v", err) + } + t.Cleanup(func() { pc.Close() }) + + host, port, err := net.SplitHostPort(pc.LocalAddr().String()) + if err != nil { + t.Fatalf("SplitHostPort(%q): %v", pc.LocalAddr().String(), err) + } + portNum, err := strconv.Atoi(port) + if err != nil { + t.Fatalf("strconv.Atoi(%q): %v", port, err) + } + cfg := &Config{StatsdHost: host, StatsdPort: portNum, StatsdProtocol: "udp"} + + const producers = 4 + const perProducer = 5 + const total = producers * perProducer + + q := make(chan string, total) + + senderDone := make(chan struct{}) + go func() { + defer close(senderDone) + statsdSender(cfg, q) + }() + + var wg sync.WaitGroup + for i := 0; i < producers; i++ { + wg.Add(1) + go func(id int) { + defer wg.Done() + for j := 0; j < perProducer; j++ { + increment(q, "connectivity.test", []string{"producer:p", "j:v"}) + } + }(i) + } + wg.Wait() + // Close the queue so statsdSender's range loop terminates, then wait + // for the goroutine to exit so the -race detector sees no leaked + // consumer running into subsequent tests. + close(q) + select { + case <-senderDone: + case <-time.After(2 * time.Second): + t.Fatal("statsdSender did not exit after queue close") + } + + // Read all expected datagrams from the listener. + if err := pc.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatalf("SetReadDeadline: %v", err) + } + got := 0 + buf := make([]byte, 1024) + wantPayload := "connectivity.test:1|c|#producer:p,j:v" + for got < total { + n, _, err := pc.ReadFrom(buf) + if err != nil { + t.Fatalf("ReadFrom after %d/%d datagrams: %v", got, total, err) + } + if string(buf[:n]) != wantPayload { + t.Fatalf("datagram[%d] = %q; want %q", got, string(buf[:n]), wantPayload) + } + got++ + } +} + +// TestWaitLoopCompletesWhenChecksSucceed exercises the WaitLoop goroutine +// fan-out with a stub check that succeeds on its second invocation per +// destination. The assertion is that wg.Wait() in WaitLoop returns (the test +// itself would time out otherwise) and that every destination's check was +// invoked at least twice — proving the retry loop in WaitFor actually ran. +// +// NumGoroutine bracketing catches the regression of a leaked WaitFor +// goroutine that would silently spin past the test. +func TestWaitLoopCompletesWhenChecksSucceed(t *testing.T) { + before := runtime.NumGoroutine() + + const destCount = 3 + + type destState struct { + mu sync.Mutex + attempts int + } + states := make([]*destState, destCount) + dests := make([]*Destination, destCount) + checks := make([]func() bool, destCount) + for i := 0; i < destCount; i++ { + s := &destState{} + states[i] = s + dests[i] = &Destination{Label: "stub", Host: "host", Port: 1} + checks[i] = func() bool { + s.mu.Lock() + defer s.mu.Unlock() + s.attempts++ + return s.attempts >= 2 + } + } + + // Drive WaitLoop via the seam: a zero sleep keeps the test fast. + waitLoop(dests, checks, 0) + + for i, s := range states { + s.mu.Lock() + got := s.attempts + s.mu.Unlock() + if got < 2 { + t.Errorf("destination[%d] attempts = %d; want >= 2", i, got) + } + } + + // Give scheduler a moment to reap goroutines, then verify no leak. + // runtime.Gosched is enough — every goroutine returned before waitLoop + // did. + runtime.Gosched() + if after := runtime.NumGoroutine(); after > before { + t.Errorf("goroutines leaked: before=%d after=%d", before, after) + } +} + +// TestMonitorWithCheckResetsConfidenceOnFailure pins the post-#16 behavior: +// after a success run that saturates confidence at 10 (so the sleep is 10 +// minutes), a single failed check snaps the next sleep back to 1 minute. The +// seam exposes one iteration at a time so the test can drive the loop +// synchronously without an injected wall-clock dependency. +// +// Refs #16 — the production logic for the reset already merged; this test +// locks it in so future refactors don't regress it. +func TestMonitorWithCheckResetsConfidenceOnFailure(t *testing.T) { + dest := &Destination{Label: "stub", Host: "host", Port: 1} + + var sleeps []time.Duration + sleep := func(d time.Duration) { sleeps = append(sleeps, d) } + + // 12 successes drive confidence past 10 and pin it at the cap, then one + // failure resets it to 1. + results := []bool{true, true, true, true, true, true, true, true, true, true, true, true, false} + + confidence := 1 + for _, r := range results { + r := r + check := func() bool { return r } + confidence = dest.monitorWithCheck(confidence, check, sleep) + } + + if got := len(sleeps); got != len(results) { + t.Fatalf("sleep call count = %d; want %d", got, len(results)) + } + // After 10+ successes the saturated sleep should be 10 minutes. + if got, want := sleeps[10], 10*time.Minute; got != want { + t.Errorf("sleeps[10] (post-saturation) = %v; want %v", got, want) + } + if got, want := sleeps[11], 10*time.Minute; got != want { + t.Errorf("sleeps[11] (still saturated) = %v; want %v", got, want) + } + // The failure at index 12 snaps the next sleep to 1 minute. + if got, want := sleeps[12], 1*time.Minute; got != want { + t.Errorf("sleeps[12] (after failure) = %v; want %v", got, want) + } + if confidence != 1 { + t.Errorf("confidence after failure = %d; want 1", confidence) + } +} diff --git a/connectivity.go b/connectivity.go index f7442a6..a755048 100644 --- a/connectivity.go +++ b/connectivity.go @@ -4,6 +4,7 @@ import ( "log" "os" "sync" + "time" ) /* @@ -153,13 +154,28 @@ func CheckLoop(destinations []*Destination) bool { } func WaitLoop(destinations []*Destination) { + checks := make([]func() bool, len(destinations)) + for i, dest := range destinations { + checks[i] = dest.Check + } + waitLoop(destinations, checks, 15*time.Second) +} + +// waitLoop is the testable form of WaitLoop: each destination is polled by +// the corresponding entry in checks, sleeping `sleep` between attempts. +// Tests pass deterministic stub checks and a zero sleep so the goroutine +// fan-out can be exercised hermetically. The two slices must be the same +// length; in production they're built from the destinations themselves. +// +// see #17 -- still no context.Context for cancellation. +func waitLoop(destinations []*Destination, checks []func() bool, sleep time.Duration) { var wg sync.WaitGroup - for _, dest := range destinations { + for i, dest := range destinations { wg.Add(1) - go func(dest *Destination) { + go func(dest *Destination, check func() bool) { defer wg.Done() - dest.WaitFor() - }(dest) + dest.waitForWithCheck(check, sleep) + }(dest, checks[i]) } wg.Wait() diff --git a/destinations.go b/destinations.go index 60438df..415e9a2 100644 --- a/destinations.go +++ b/destinations.go @@ -198,30 +198,51 @@ func (dest *Destination) Monitor() { confidence := 1 for { - reachable := dest.Check() + confidence = dest.monitorWithCheck(confidence, dest.Check, time.Sleep) + } +} - if reachable { - confidence += 1 - if confidence > 10 { - confidence = 10 - } - } else { - confidence = 1 +// monitorWithCheck runs one iteration of the Monitor loop: invoke check, +// adjust confidence per the #16 reset-on-failure rule, then sleep. The +// confidence value is threaded through the caller (rather than held in a +// closure) so a test can drive a deterministic sequence of iterations +// without spawning a goroutine. The injected sleep lets the test observe +// the chosen sleep duration without waiting on a real clock. +// +// see #17 -- Monitor itself still has no termination condition, no panic +// recovery, and no context.Context; those land with the lifecycle work. +func (dest *Destination) monitorWithCheck(confidence int, check func() bool, sleep func(time.Duration)) int { + if check() { + confidence += 1 + if confidence > 10 { + confidence = 10 } - - time.Sleep(time.Duration(confidence) * time.Minute) + } else { + confidence = 1 } + + sleep(time.Duration(confidence) * time.Minute) + return confidence } func (dest *Destination) WaitFor() { - for { - reachable := dest.Check() + // see #18 -- the 15s flat poll has no overall deadline and no + // exponential backoff; that lands with the wait-timeout flag. + dest.waitForWithCheck(dest.Check, 15*time.Second) +} - if reachable { +// waitForWithCheck polls check until it returns true, sleeping `sleep` +// between attempts. The seam keeps WaitFor's public signature intact while +// giving tests a way to substitute a deterministic check and a zero sleep. +// +// see #17 -- there is still no context.Context for cancellation; the loop +// runs forever on a permanently-broken destination. +func (dest *Destination) waitForWithCheck(check func() bool, sleep time.Duration) { + for { + if check() { LogDestination(dest, "Connected") return } - - time.Sleep(15 * time.Second) + time.Sleep(sleep) } } diff --git a/statsd.go b/statsd.go index f3da303..2164150 100644 --- a/statsd.go +++ b/statsd.go @@ -8,22 +8,49 @@ import ( "time" ) +// queue is the package-level statsd message buffer that main wires +// StatsdSender to consume. The lowercase helpers (count, timer, gauge, +// increment, statsdSender) take an explicit queue parameter so tests can +// drive the producer/consumer surface with a hermetic channel; see #49 and +// concurrency_test.go. +// +// The capacity-100 default and blocking send are the back-pressure hazard +// described in #11. The seam below does NOT fix that bug — it just makes it +// observable from a test. See #17 for the broader process-lifecycle work +// that needs to land before non-blocking sends and drain-on-shutdown can be +// done correctly. var queue = make(chan string, 100) func Increment(metric string, tags []string) { - Count(metric, 1, tags) + increment(queue, metric, tags) } func Count(metric string, value int, tags []string) { - queue <- fmt.Sprintf("%s:%d|c|#%s", metric, value, formatTags(tags)) + count(queue, metric, value, tags) } func Timer(metric string, took time.Duration, tags []string) { - queue <- fmt.Sprintf("%s:%d|ms|#%s", metric, took/1e6, formatTags(tags)) + timer(queue, metric, took, tags) } func Gauge(metric string, value int, tags []string) { - queue <- fmt.Sprintf("%s:%d|g|#%s", metric, value, formatTags(tags)) + gauge(queue, metric, value, tags) +} + +func increment(q chan<- string, metric string, tags []string) { + count(q, metric, 1, tags) +} + +func count(q chan<- string, metric string, value int, tags []string) { + q <- fmt.Sprintf("%s:%d|c|#%s", metric, value, formatTags(tags)) +} + +func timer(q chan<- string, metric string, took time.Duration, tags []string) { + q <- fmt.Sprintf("%s:%d|ms|#%s", metric, took/1e6, formatTags(tags)) +} + +func gauge(q chan<- string, metric string, value int, tags []string) { + q <- fmt.Sprintf("%s:%d|g|#%s", metric, value, formatTags(tags)) } func formatTags(tags []string) string { @@ -40,7 +67,11 @@ func EscapeTag(s string) string { } func StatsdSender(config *Config) { - for s := range queue { + statsdSender(config, queue) +} + +func statsdSender(config *Config, q <-chan string) { + for s := range q { statsdHostPort := fmt.Sprintf("%s:%d", config.StatsdHost, config.StatsdPort) if conn, err := net.Dial(config.StatsdProtocol, statsdHostPort); err == nil { io.WriteString(conn, s)