tests: the adaptive engine's fixed bind polls wait on Listen (the direct-engine polls are a follow-up)

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)
 	}

internal/adaptive/detect_bound_974_linux_test.go

@@ -156,21 +156,17 @@ 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 {
-		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")
 		}
 	})
-	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)
 	}

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()

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)
 	}

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

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)

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

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)
 	}

internal/engine/epoll/adaptive_abort_844_linux_test.go

@@ -263,11 +263,12 @@ 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 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" {
@@ -546,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
+}

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
+}

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