From 26146e79cad564c50c169b1b8857166616b0991e Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 01:16:19 +0100 Subject: [PATCH 1/3] adaptive tests: the remaining fixed bind polls use the Listen-aware wait (Refs #1104) 791, driver_provider, 974 and prebound waited 3 to 10 s for the address with a poll that cannot tell a starved poller from a slow Listen; the epoll 844 test waited 5 s. They use awaitBind890 (30 s guard, stops at once with Listen's own error when it returns unbound) through awaitBoundOrFail890, a helper for tests that start Listen themselves; the epoll test does the same inline. The 25 s wait of listenFloor682, which already follows Listen, stays. --- .../adaptive/async_abort_791_linux_test.go | 10 +++------- .../adaptive/detect_bound_974_linux_test.go | 10 +++------- internal/adaptive/driver_provider_test.go | 13 +++---------- internal/adaptive/flap_conns_per_ring_test.go | 16 ++++++++++++++++ internal/adaptive/prebound_listener_test.go | 14 +++----------- .../epoll/adaptive_abort_844_linux_test.go | 18 +++++++++++++----- 6 files changed, 41 insertions(+), 40 deletions(-) diff --git a/internal/adaptive/async_abort_791_linux_test.go b/internal/adaptive/async_abort_791_linux_test.go index 5343b8347..b9bc3d4f3 100644 --- a/internal/adaptive/async_abort_791_linux_test.go +++ b/internal/adaptive/async_abort_791_linux_test.go @@ -157,13 +157,9 @@ func abortAdaptive791(t *testing.T, adopted bool, inject func(*Engine)) { ctx, cancel := context.WithCancel(context.Background()) t.Cleanup(cancel) done := make(chan error, 1) - go func() { done <- e.Listen(ctx) }() - for dl := time.Now().Add(3 * time.Second); e.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if e.Addr() == nil { - t.Fatal("adaptive engine never bound") - } + ended := make(chan struct{}) + go func() { done <- e.Listen(ctx); close(ended) }() + awaitBoundOrFail890(t, e, cancel, done, ended) if inject != nil { inject(e) } diff --git a/internal/adaptive/detect_bound_974_linux_test.go b/internal/adaptive/detect_bound_974_linux_test.go index 5be80f0b0..327fedb9a 100644 --- a/internal/adaptive/detect_bound_974_linux_test.go +++ b/internal/adaptive/detect_bound_974_linux_test.go @@ -156,7 +156,8 @@ func startOnIOUring974(t *testing.T, readHeaderTimeout time.Duration) string { } ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) - go func() { done <- e.Listen(ctx) }() + ended := make(chan struct{}) + go func() { done <- e.Listen(ctx); close(ended) }() t.Cleanup(func() { cancel() select { @@ -165,12 +166,7 @@ func startOnIOUring974(t *testing.T, readHeaderTimeout time.Duration) string { t.Error("adaptive engine did not stop within 10s") } }) - for dl := time.Now().Add(10 * time.Second); e.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if e.Addr() == nil { - t.Fatal("adaptive engine never bound") - } + awaitBoundOrFail890(t, e, cancel, done, ended) if got := getOnce(t, addr); got != 200 { t.Fatalf("GET = %d, want 200", got) } diff --git a/internal/adaptive/driver_provider_test.go b/internal/adaptive/driver_provider_test.go index c3cd526c0..ea9eceab7 100644 --- a/internal/adaptive/driver_provider_test.go +++ b/internal/adaptive/driver_provider_test.go @@ -51,18 +51,11 @@ func newBoundAdaptive(t *testing.T) (*Engine, func()) { ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) - go func() { done <- e.Listen(ctx) }() + ended := make(chan struct{}) + go func() { done <- e.Listen(ctx); close(ended) }() // Wait for bound. - deadline := time.Now().Add(3 * time.Second) - for e.Addr() == nil && time.Now().Before(deadline) { - time.Sleep(10 * time.Millisecond) - } - if e.Addr() == nil { - cancel() - <-done - t.Fatal("adaptive engine never bound") - } + awaitBoundOrFail890(t, e, cancel, done, ended) stop := func() { cancel() diff --git a/internal/adaptive/flap_conns_per_ring_test.go b/internal/adaptive/flap_conns_per_ring_test.go index e82cc5f63..1038dc60a 100644 --- a/internal/adaptive/flap_conns_per_ring_test.go +++ b/internal/adaptive/flap_conns_per_ring_test.go @@ -296,6 +296,22 @@ func awaitBind890(addr func() net.Addr, ended <-chan struct{}, limit time.Durati } } +// awaitBoundOrFail890 is awaitBind890 for a test that started Listen itself +// (it keeps ctx's cancel and the done channel, to assert on the stop). The +// caller's Listen goroutine sends its result on done and then closes ended. On +// failure it cancels Listen, joins it and fails t with Listen's own error. +func awaitBoundOrFail890(t *testing.T, e *Engine, cancel context.CancelFunc, done <-chan error, ended <-chan struct{}) net.Addr { + t.Helper() + a, took, err := awaitBind890(e.Addr, ended, bindLimit890) + if err != nil { + cancel() + lerr := <-done + t.Fatalf("adaptive engine never bound: %v (waited %v; Listen: %v)", err, took, lerr) + } + t.Logf("adaptive engine bound %v after Listen started", took.Round(time.Millisecond)) + return a +} + // listenAndAwaitBind890 starts e.Listen, waits for the bound address (see // awaitBind890), logs how long the bind took, and returns the address with a // stop func that cancels Listen and joins it. It fails the test with Listen's diff --git a/internal/adaptive/prebound_listener_test.go b/internal/adaptive/prebound_listener_test.go index cb1818440..720ef8d69 100644 --- a/internal/adaptive/prebound_listener_test.go +++ b/internal/adaptive/prebound_listener_test.go @@ -52,18 +52,10 @@ func preBoundAdaptive(t *testing.T, workers int) (*Engine, string, func()) { ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) - go func() { done <- e.Listen(ctx) }() + ended := make(chan struct{}) + go func() { done <- e.Listen(ctx); close(ended) }() - for dl := time.Now().Add(10 * time.Second); e.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if e.Addr() == nil { - cancel() - if lerr := <-done; lerr != nil { - t.Fatalf("adaptive engine never bound; Listen: %v", lerr) - } - t.Fatal("adaptive engine never bound") - } + awaitBoundOrFail890(t, e, cancel, done, ended) return e, want, func() { cancel() <-done diff --git a/internal/engine/epoll/adaptive_abort_844_linux_test.go b/internal/engine/epoll/adaptive_abort_844_linux_test.go index c02648554..b03c69bac 100644 --- a/internal/engine/epoll/adaptive_abort_844_linux_test.go +++ b/internal/engine/epoll/adaptive_abort_844_linux_test.go @@ -263,11 +263,19 @@ func start844(t *testing.T, h *a844, gate bool) *rig844 { t.Cleanup(cancel) r.done = make(chan error, 1) go func() { r.done <- ae.Listen(ctx) }() - for dl := time.Now().Add(5 * time.Second); ae.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if ae.Addr() == nil { - t.Fatal("adaptive engine never bound") + // A guard against a hang, not an expectation: Listen gives up on its own after + // its bind deadline (20 s) and returns its error, which this reports. The + // 5 s it replaces could be missed by a starved poll or a slow Listen alone + // (celeris#890 item 98, internal/adaptive's awaitBind890). + for dl := time.Now().Add(30 * time.Second); ae.Addr() == nil; time.Sleep(10 * time.Millisecond) { + select { + case lerr := <-r.done: + t.Fatalf("adaptive engine never bound: Listen returned %v", lerr) + default: + } + if time.Now().After(dl) { + t.Fatal("adaptive engine never bound within 30s") + } } r.addr = ae.Addr().String() if pkg844 == "iouring" { From 0852da1fdd48bc9734492f217c5fa91b0afdda23 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 01:41:57 +0100 Subject: [PATCH 2/3] engine tests: the iouring adaptive-abort twin and the 848 test wait for Listen, as the epoll one does (Refs #1104) internal/engine/iouring's adaptive_abort_844 is the same test as the epoll one and polled the adaptive engine for 5 s; adaptive_switch_848 did the same. Both wait up to 30 s and stop at once with Listen's own error when it returns unbound (awaitAdaptiveBound844, a copy of the epoll file's: another package's test cannot import internal/adaptive's awaitBind890). The epoll copy of the inline loop becomes the same helper. --- .../epoll/adaptive_abort_844_linux_test.go | 37 +++++++++++++------ .../iouring/adaptive_abort_844_linux_test.go | 30 ++++++++++++--- .../iouring/adaptive_switch_848_linux_test.go | 7 +--- 3 files changed, 52 insertions(+), 22 deletions(-) diff --git a/internal/engine/epoll/adaptive_abort_844_linux_test.go b/internal/engine/epoll/adaptive_abort_844_linux_test.go index b03c69bac..e4e86326f 100644 --- a/internal/engine/epoll/adaptive_abort_844_linux_test.go +++ b/internal/engine/epoll/adaptive_abort_844_linux_test.go @@ -264,18 +264,11 @@ func start844(t *testing.T, h *a844, gate bool) *rig844 { r.done = make(chan error, 1) go func() { r.done <- ae.Listen(ctx) }() // A guard against a hang, not an expectation: Listen gives up on its own after - // its bind deadline (20 s) and returns its error, which this reports. The - // 5 s it replaces could be missed by a starved poll or a slow Listen alone - // (celeris#890 item 98, internal/adaptive's awaitBind890). - for dl := time.Now().Add(30 * time.Second); ae.Addr() == nil; time.Sleep(10 * time.Millisecond) { - select { - case lerr := <-r.done: - t.Fatalf("adaptive engine never bound: Listen returned %v", lerr) - default: - } - if time.Now().After(dl) { - t.Fatal("adaptive engine never bound within 30s") - } + // its bind deadline (20 s) and returns its error, which awaitAdaptiveBound844 reports. + // The 5 s it replaces could be missed by a starved poll or a slow Listen + // alone (celeris#890 item 98, internal/adaptive's awaitBind890). + if err := awaitAdaptiveBound844(ae, r.done, 30*time.Second); err != nil { + t.Fatalf("adaptive engine never bound: %v", err) } r.addr = ae.Addr().String() if pkg844 == "iouring" { @@ -554,3 +547,23 @@ func TestAdaptiveAbortLogsAPanicValueThatSharesStreamMemory844(t *testing.T) { t.Errorf("celeris#844 item 3: the slog handler formatted the panic value after the conn was torn down and got %q", text) } } + +// awaitAdaptiveBound844 waits up to d for the adaptive engine ae to publish its +// address. If Listen returns first (done receives its result) it does not wait +// out d: it fails with that error, and puts the value back on done for the +// teardown. The same wait as internal/adaptive's awaitBind890 (this is another +// package's test, so it cannot import it). +func awaitAdaptiveBound844(ae interface{ Addr() net.Addr }, done chan error, d time.Duration) error { + for dl := time.Now().Add(d); ae.Addr() == nil; { + select { + case err := <-done: + done <- err + return fmt.Errorf("Listen returned before the engine bound: %v", err) + case <-time.After(10 * time.Millisecond): + } + if time.Now().After(dl) { + return fmt.Errorf("no address after %v", d) + } + } + return nil +} diff --git a/internal/engine/iouring/adaptive_abort_844_linux_test.go b/internal/engine/iouring/adaptive_abort_844_linux_test.go index 059a0dd24..69ee5071c 100644 --- a/internal/engine/iouring/adaptive_abort_844_linux_test.go +++ b/internal/engine/iouring/adaptive_abort_844_linux_test.go @@ -263,11 +263,11 @@ func start844(t *testing.T, h *a844, gate bool) *rig844 { t.Cleanup(cancel) r.done = make(chan error, 1) go func() { r.done <- ae.Listen(ctx) }() - for dl := time.Now().Add(5 * time.Second); ae.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if ae.Addr() == nil { - t.Fatal("adaptive engine never bound") + // 30 s is a guard against a hang (Listen gives up by itself after 20 s and + // awaitAdaptiveBound844 reports its error); the 5 s it replaces could be missed by a + // starved poll or a slow Listen alone (celeris#890 item 98). + if err := awaitAdaptiveBound844(ae, r.done, 30*time.Second); err != nil { + t.Fatalf("adaptive engine never bound: %v", err) } r.addr = ae.Addr().String() if pkg844 == "iouring" { @@ -546,3 +546,23 @@ func TestAdaptiveAbortLogsAPanicValueThatSharesStreamMemory844(t *testing.T) { t.Errorf("celeris#844 item 3: the slog handler formatted the panic value after the conn was torn down and got %q", text) } } + +// awaitAdaptiveBound844 waits up to d for the adaptive engine ae to publish its +// address. If Listen returns first (done receives its result) it does not wait +// out d: it fails with that error, and puts the value back on done for the +// teardown. The same wait as internal/adaptive's awaitBind890 (this is another +// package's test, so it cannot import it). +func awaitAdaptiveBound844(ae interface{ Addr() net.Addr }, done chan error, d time.Duration) error { + for dl := time.Now().Add(d); ae.Addr() == nil; { + select { + case err := <-done: + done <- err + return fmt.Errorf("Listen returned before the engine bound: %v", err) + case <-time.After(10 * time.Millisecond): + } + if time.Now().After(dl) { + return fmt.Errorf("no address after %v", d) + } + } + return nil +} diff --git a/internal/engine/iouring/adaptive_switch_848_linux_test.go b/internal/engine/iouring/adaptive_switch_848_linux_test.go index a05d97382..73ab6d524 100644 --- a/internal/engine/iouring/adaptive_switch_848_linux_test.go +++ b/internal/engine/iouring/adaptive_switch_848_linux_test.go @@ -88,11 +88,8 @@ func TestAdaptiveSwitchIsNotALoopExit848(t *testing.T) { done := make(chan error, 1) go func() { done <- ae.Listen(ctx) }() t.Cleanup(func() { cancel(); <-done }) - for dl := time.Now().Add(5 * time.Second); ae.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if ae.Addr() == nil { - t.Fatal("adaptive engine never bound") + if err := awaitAdaptiveBound844(ae, done, 30*time.Second); err != nil { + t.Fatalf("adaptive engine never bound: %v", err) } addr := ae.Addr().String() keep, err := net.DialTimeout("tcp", addr, 2*time.Second) // idle keep-alive across the switches From 0cba5ba4067cc25f2f0e4b04a9258849a0b54c78 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 01:41:57 +0100 Subject: [PATCH 3/3] adaptive tests: switch_idle_conns, pinned_threads_905 and epoll_halves_e9 wait for Listen; 974's cleanup waits on ended (Refs #1104) Three more tests that drive a real adaptive engine used 5, 10 and 15 s polls that cannot tell a starved poller from a slow Listen; they use awaitBoundOrFail890. The teardowns of switch_idle_conns, epoll_halves_e9 and detect_bound_974 waited on the done channel, which the helper drains on a failure, so a failing run printed a false 'did not stop' after 10 to 15 s per test; they wait on ended, which is closed after done is sent on. --- internal/adaptive/detect_bound_974_linux_test.go | 2 +- internal/adaptive/epoll_halves_e9_linux_test.go | 14 +++++--------- internal/adaptive/pinned_threads_905_linux_test.go | 11 +++-------- internal/adaptive/switch_idle_conns_linux_test.go | 12 ++++-------- 4 files changed, 13 insertions(+), 26 deletions(-) diff --git a/internal/adaptive/detect_bound_974_linux_test.go b/internal/adaptive/detect_bound_974_linux_test.go index 327fedb9a..09bbd4afb 100644 --- a/internal/adaptive/detect_bound_974_linux_test.go +++ b/internal/adaptive/detect_bound_974_linux_test.go @@ -161,7 +161,7 @@ func startOnIOUring974(t *testing.T, readHeaderTimeout time.Duration) string { t.Cleanup(func() { cancel() select { - case <-done: + case <-ended: // closed after done is sent on, whoever has drained it case <-time.After(10 * time.Second): t.Error("adaptive engine did not stop within 10s") } diff --git a/internal/adaptive/epoll_halves_e9_linux_test.go b/internal/adaptive/epoll_halves_e9_linux_test.go index 97c643cda..546533ee2 100644 --- a/internal/adaptive/epoll_halves_e9_linux_test.go +++ b/internal/adaptive/epoll_halves_e9_linux_test.go @@ -57,21 +57,17 @@ func startEpollAdaptive(t *testing.T, cfg resource.Config, h stream.Handler) (*E } ctx, cancel := context.WithCancel(context.Background()) errCh := make(chan error, 1) - go func() { errCh <- e.Listen(ctx) }() + ended := make(chan struct{}) + go func() { errCh <- e.Listen(ctx); close(ended) }() t.Cleanup(func() { cancel() select { - case <-errCh: - case <-time.After(5 * time.Second): // 5 s: a test that waited for Listen has drained errCh already + case <-ended: // closed after errCh is sent on, whoever has drained it + case <-time.After(5 * time.Second): } _ = e.Shutdown(context.Background()) }) - for dl := time.Now().Add(15 * time.Second); e.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if e.Addr() == nil { - t.Fatal("adaptive engine did not bind") - } + awaitBoundOrFail890(t, e, cancel, errCh, ended) if got := e.ActiveEngine().Type(); got != engine.Epoll { t.Fatalf("e9 PREMISE: the active engine is %v, want epoll: the epoll halves of #874 and #876 are not what this test would run", got) } diff --git a/internal/adaptive/pinned_threads_905_linux_test.go b/internal/adaptive/pinned_threads_905_linux_test.go index c1b855ea5..f6172d1e4 100644 --- a/internal/adaptive/pinned_threads_905_linux_test.go +++ b/internal/adaptive/pinned_threads_905_linux_test.go @@ -60,7 +60,8 @@ func startPromoted905(t *testing.T, base string) (int, func()) { e.FreezeSwitching() ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) - go func() { done <- e.Listen(ctx) }() + ended := make(chan struct{}) + go func() { done <- e.Listen(ctx); close(ended) }() stop := func() { cancel() select { @@ -72,13 +73,7 @@ func startPromoted905(t *testing.T, base string) (int, func()) { t.Fatal("the adaptive engine did not stop within 15s") } } - for dl := time.Now().Add(10 * time.Second); e.Addr() == nil && time.Now().Before(dl); { - time.Sleep(5 * time.Millisecond) - } - if e.Addr() == nil { - stop() - t.Fatal("the adaptive engine did not bind within 10s") - } + awaitBoundOrFail890(t, e, cancel, done, ended) if got := e.ActiveEngine().Type(); got != engine.Epoll { stop() t.Fatalf("this test promotes from epoll, but the engine started on %v", got) diff --git a/internal/adaptive/switch_idle_conns_linux_test.go b/internal/adaptive/switch_idle_conns_linux_test.go index 3d051ebfd..21984a20a 100644 --- a/internal/adaptive/switch_idle_conns_linux_test.go +++ b/internal/adaptive/switch_idle_conns_linux_test.go @@ -255,21 +255,17 @@ func startAdaptive662(t *testing.T, cfg resource.Config) (*Engine, int) { e.FreezeSwitching() ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) - go func() { done <- e.Listen(ctx) }() + ended := make(chan struct{}) + go func() { done <- e.Listen(ctx); close(ended) }() t.Cleanup(func() { cancel() select { - case <-done: + case <-ended: case <-time.After(15 * time.Second): t.Error("adaptive engine did not stop within 15s") } }) - for dl := time.Now().Add(5 * time.Second); e.Addr() == nil && time.Now().Before(dl); { - time.Sleep(10 * time.Millisecond) - } - if e.Addr() == nil { - t.Fatal("adaptive engine never bound") - } + awaitBoundOrFail890(t, e, cancel, done, ended) if got := e.ActiveEngine().Type(); got != engine.Epoll { t.Fatalf("these tests switch between the two sub-engines starting from epoll, started on %v", got) }