tests: test/server start retry takes a fresh port; the shutdown gate test waits for the engine (Refs #1104)

test/server/h2upgrade_matrix_889_linux_test.go

@@ -6,7 +6,7 @@ import (
 	"bufio"
 	"bytes"
 	"context"
-	"fmt"
+	"errors"
 	"io"
 	"net"
 	"net/http"
@@ -176,42 +176,31 @@ func firstReply889(t *testing.T, addr, wire string) []byte {
 // other starters here decide it is ready).
 func startTCPReady889(t *testing.T, cfg celeris.Config) string {
 	t.Helper()
-	var (
-		addr string
-		s    *celeris.Server
-		done chan error
-	)
 	// An io_uring Start can fail with ring-setup ENOMEM under the runner's
-	// 8 MiB memlock while earlier rings are still charged: retry it, as the
-	// sibling starters do (celeris#1104).
-	startRetryingRingENOMEM(t, "server start", func() error {
-		ln, err := net.Listen("tcp", "127.0.0.1:0")
-		if err != nil {
-			t.Fatal(err)
-		}
-		addr = ln.Addr().String()
-		_ = ln.Close()
-		cfg.Addr = addr
-		s = celeris.New(cfg)
+	// 8 MiB memlock while earlier rings are still charged: startPicked1104
+	// retries it, and keeps the port from the pick to the bind (celeris#1104).
+	s, addr, done := startPicked1104(t, cfg, func(s *celeris.Server) {
 		s.GET("/h", func(c *celeris.Context) error { return c.String(http.StatusOK, "h1-answer") })
-		done = make(chan error, 1)
-		go func(s *celeris.Server, done chan error) { done <- s.Start() }(s, done)
+	}, func(s *celeris.Server, addr string, done <-chan error) error {
 		for i := 0; ; i++ {
 			select {
 			case err := <-done:
-				return fmt.Errorf("Start returned before the listener was up: %w", err)
+				if err == nil {
+					err = errors.New("Start returned nil before the listener was up")
+				}
+				return err
 			default:
 			}
-			c, err := net.DialTimeout("tcp", addr, time.Second)
-			if err == nil {
-				_ = c.Close()
-				return nil
+			// The listener is the server's from the start, so a dial alone proves
+			// nothing: wait for the engine to publish its address, then dial.
+			if s.Addr() != nil {
+				if c, err := net.DialTimeout("tcp", addr, time.Second); err == nil {
+					_ = c.Close()
+					return nil
+				}
 			}
 			if i > 300 {
-				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-				_ = s.Shutdown(ctx)
-				cancel()
-				return fmt.Errorf("not listening: %w", err)
+				return errors.New("the engine did not publish an address within 15 s")
 			}
 			time.Sleep(50 * time.Millisecond)
 		}

test/server/ring_enomem_test.go

@@ -37,31 +37,55 @@ func isRingENOMEM(err error) bool {
 	return strings.Contains(msg, "cannot allocate memory") || strings.Contains(msg, "io_uring not available on this system")
 }
 
-// retryRingENOMEM calls attempt until it returns nil or an error that is not a
-// ring ENOMEM, or until window has passed since the first call. It returns the
-// number of calls and the last error. attempt must release whatever a failed
-// try holds (stop the server, close the listener) before it returns.
-func retryRingENOMEM(attempt func() error, window time.Duration) (tries int, err error) {
+// isAddrInUse reports a Start error that is the bind race of a starter that
+// has Start bind a port it picked and closed (the start-shutdown rig's plain
+// Start arm): another test of the process was given the port in between.
+func isAddrInUse(err error) bool {
+	return err != nil && strings.Contains(err.Error(), "address already in use")
+}
+
+// maxAddrInUseTries bounds how often a taken port is retried: a port that stays
+// taken is a real conflict, not the race.
+const maxAddrInUseTries = 10
+
+// retryStart1104 calls attempt until it returns nil or an error it should not
+// retry, or until window has passed since the first call. It retries a ring
+// ENOMEM for the whole window and an address already in use up to
+// maxAddrInUseTries times (attempt picks a fresh port each call); any other
+// error is returned at once. It returns the number of calls and the last error.
+// attempt must release whatever a failed try holds (stop the server, close the
+// listener) before it returns.
+func retryStart1104(attempt func() error, window time.Duration) (tries int, err error) {
 	deadline := time.Now().Add(window)
+	inUse := 0
 	for {
 		tries++
-		if err = attempt(); err == nil || !isRingENOMEM(err) || !time.Now().Before(deadline) {
+		if err = attempt(); err == nil || !time.Now().Before(deadline) {
+			return tries, err
+		}
+		switch {
+		case isRingENOMEM(err):
+		case isAddrInUse(err):
+			if inUse++; inUse >= maxAddrInUseTries {
+				return tries, err
+			}
+		default:
 			return tries, err
 		}
 		time.Sleep(20 * time.Millisecond)
 	}
 }
 
-// startRetryingRingENOMEM runs attempt with retryRingENOMEM and fails t with
+// startRetrying1104 runs attempt with retryStart1104 and fails t with
 // what if it still fails.
-func startRetryingRingENOMEM(t testing.TB, what string, attempt func() error) {
+func startRetrying1104(t testing.TB, what string, attempt func() error) {
 	t.Helper()
-	tries, err := retryRingENOMEM(attempt, ringENOMEMWindow)
+	tries, err := retryStart1104(attempt, ringENOMEMWindow)
 	if err != nil {
 		t.Fatalf("%s: %v (try %d)", what, err, tries)
 	}
 	if tries > 1 {
-		t.Logf("%s: retried on ring ENOMEM: %d tries", what, tries)
+		t.Logf("%s: retried (a ring ENOMEM or a taken port): %d tries", what, tries)
 	}
 }
 
@@ -88,7 +112,7 @@ func TestRetryRingENOMEM1104(t *testing.T) {
 
 	// A ring ENOMEM is retried until the start succeeds.
 	calls := 0
-	tries, err := retryRingENOMEM(func() error {
+	tries, err := retryStart1104(func() error {
 		calls++
 		if calls < 3 {
 			return enomem
@@ -101,15 +125,45 @@ func TestRetryRingENOMEM1104(t *testing.T) {
 
 	// Any other error is returned at once.
 	calls = 0
-	tries, err = retryRingENOMEM(func() error { calls++; return other }, time.Second)
+	tries, err = retryStart1104(func() error { calls++; return other }, time.Second)
 	if !errors.Is(err, other) || tries != 1 || calls != 1 {
 		t.Errorf("a non-ENOMEM error: %d tries, err %v; want 1, %v", tries, err, other)
 	}
 
 	// A persistent ENOMEM is retried for the window, then returned.
 	start := time.Now()
-	tries, err = retryRingENOMEM(func() error { return enomem }, 100*time.Millisecond)
+	tries, err = retryStart1104(func() error { return enomem }, 100*time.Millisecond)
 	if !errors.Is(err, enomem) || tries < 2 || time.Since(start) < 100*time.Millisecond {
 		t.Errorf("a persistent ENOMEM: %d tries after %v, err %v; want >=2 tries over the 100 ms window, then %v", tries, time.Since(start), err, enomem)
 	}
 }
+
+// A start whose port is taken before it binds fails with "address already in
+// use": the start-shutdown rig's plain-Start arm picks a free port, closes it
+// and has Start bind it, and two parallel test cells of one process can pick the
+// same port (celeris#1104: TestStalledSSEEndsWithShutdown855/std/h2c/budget
+// failed once with "startErr=listen: ... bind: address already in use"). The
+// starters that can hand the server its listener do (startPicked1104); where
+// Start must bind itself, the error reaches this retry (waitReadyErr refuses
+// another server's /ping answer), which takes a fresh port, up to
+// maxAddrInUseTries times: a port that stays taken is a real conflict.
+func TestRetryStartRetriesAnAddressInUse1104(t *testing.T) {
+	inUse := errors.New("server did not start: listen tcp 127.0.0.1:38797: bind: address already in use")
+
+	calls := 0
+	tries, err := retryStart1104(func() error {
+		calls++
+		if calls < 3 {
+			return inUse
+		}
+		return nil
+	}, time.Second)
+	if err != nil || tries != 3 {
+		t.Errorf("two taken ports then success: %d tries, err %v; want 3, nil", tries, err)
+	}
+
+	tries, err = retryStart1104(func() error { return inUse }, 5*time.Second)
+	if !errors.Is(err, inUse) || tries != maxAddrInUseTries {
+		t.Errorf("a port that stays taken: %d tries, err %v; want exactly %d tries, then %v", tries, err, maxAddrInUseTries, inUse)
+	}
+}

test/server/server_async_goexit_791_linux_test.go

@@ -113,7 +113,7 @@ func runServerAsyncAbort791(t *testing.T, eng celeris.EngineType, mode string) {
 	// A fresh server per try: an io_uring Start can fail with ring-setup ENOMEM
 	// under the runner's 8 MiB memlock while earlier rings are still charged
 	// (celeris#1104).
-	startRetryingRingENOMEM(t, "server start", func() error {
+	startRetrying1104(t, "server start", func() error {
 		s := celeris.New(celeris.Config{Engine: eng, Workers: 2})
 		s.GET("/boom", func(c *celeris.Context) error {
 			hits.Add(1)

test/server/shutdown_h2_pool_linux_test.go

@@ -1191,6 +1191,10 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) {
 					s.GET("/aping", func(c *celeris.Context) error { return c.String(http.StatusOK, "a") }).Async()
 				}
 			})
+			// What the engine counts before this test sends anything: the readiness
+			// probe of startH2PoolServer759 (its keep-alive conn stays open in the
+			// default transport's pool, and it polled /ping one or more times).
+			base := settledEngineMetrics1104(srv)
 			// The wait is held: GET /big on 24 HTTP/2 conns, no WINDOW_UPDATE.
 			var h2conns []net.Conn
 			for range 24 {
@@ -1270,6 +1274,35 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) {
 			// TCP_DEFER_ACCEPT hands the engine a conn with no byte after about a
 			// second: the silent conns are in its table by then.
 			time.Sleep(1300 * time.Millisecond)
+			// The gate's first pass runs 250 ms after the shutdown began, and a
+			// conn the loop has accepted but not read yet, or a request whose
+			// bytes it has not read, is closed by it as idle (the loop of a runner
+			// busy with 24 HTTP/2 downloads under -race and -cover can be that far
+			// behind: celeris#1104). So begin only once the engine counts, on top
+			// of the baseline taken before the first dial, every conn this test
+			// opened (24 HTTP/2, the keep-alive ones, the silent ones, the download,
+			// the upload on the sync arm, the promoted conn on the async arm) and
+			// every request it sent (24 HTTP/2 GETs, one ping per keep-alive conn,
+			// the download's GET, and on the sync arm the upload, which the loop
+			// counts once it has read its headers and first 64 KiB, or on the
+			// async arm the promoted conn's ping). The conn count is exact: +34 in
+			// every run. The request count is a lower bound: measured, it ends at
+			// the arithmetic or one above it (sync +32 or +33 against 32, async
+			// +32 or +33 against 32), and the extra unit is already there after the
+			// keep-alive step, before the download and the upload are sent, so it is
+			// neither of them; its source was not found, which is why this wait is
+			// `>=` for requests and exact for conns.
+			wantConns := uint64(24 + len(idle) + len(silent) + 1) // + the download
+			wantReqs := uint64(24 + len(idle) + 1)                // + the download's GET
+			if arm == "sync" {
+				wantConns++ // the upload
+				wantReqs++  // the upload's request, counted when its first bytes are read
+			}
+			if promoted != nil {
+				wantConns++
+				wantReqs++
+			}
+			waitEngineBusy1104(t, srv, base, wantConns, wantReqs)
 
 			begin := time.Now()
 			srv.beginShutdown("Shutdown", 20*time.Second)
@@ -1344,3 +1377,38 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) {
 		})
 	}
 }
+
+// settledEngineMetrics1104 returns the engine's metrics once neither the open
+// conns nor the handled requests have changed for 100 ms: the readiness probe
+// of startH2PoolServer759 counts its request after it has its response, so a
+// baseline taken the instant the server is up can miss one.
+func settledEngineMetrics1104(srv *h2PoolServer759) celeris.EngineMetrics {
+	prev := srv.s.EngineInfo().Metrics
+	for i := 0; i < 100; i++ {
+		time.Sleep(100 * time.Millisecond)
+		m := srv.s.EngineInfo().Metrics
+		if m.ActiveConnections == prev.ActiveConnections && m.RequestCount == prev.RequestCount {
+			return m
+		}
+		prev = m
+	}
+	return prev
+}
+
+// waitEngineBusy1104 waits, up to 20 s, until the engine counts at least conns
+// more open connections and reqs more handled requests than base, the metrics
+// taken before the test's first dial, and logs the surplus it ended with.
+func waitEngineBusy1104(t *testing.T, srv *h2PoolServer759, base celeris.EngineMetrics, conns, reqs uint64) {
+	t.Helper()
+	var m celeris.EngineMetrics
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) {
+		m = srv.s.EngineInfo().Metrics
+		if uint64(m.ActiveConnections-base.ActiveConnections) >= conns && m.RequestCount-base.RequestCount >= reqs {
+			t.Logf("engine counted +%d conns (want %d) and +%d requests (want %d) over the baseline of %d conns and %d requests",
+				m.ActiveConnections-base.ActiveConnections, conns, m.RequestCount-base.RequestCount, reqs, base.ActiveConnections, base.RequestCount)
+			return
+		}
+	}
+	t.Fatalf("the engine counted +%d conns and +%d requests over its baseline, want +%d and +%d before the shutdown begins",
+		m.ActiveConnections-base.ActiveConnections, m.RequestCount-base.RequestCount, conns, reqs)
+}

test/server/sse_detach_stalled_855_linux_test.go

@@ -123,29 +123,16 @@ func stallBegin855(t *testing.T, eng celeris.EngineType, wt, shutdown time.Durat
 		cfg.MaxQueueDepth = 4
 		cfg.OnSlowClient = func(*sse.Client, sse.Event) sse.ClientPolicy { return sse.ClientPolicyClose }
 	}
-	var addr string
 	// A ring's memlock is returned when the engine before it has closed, so a
-	// start that fails with ring-setup ENOMEM is retried (celeris#1104).
-	startRetryingRingENOMEM(t, "server start", func() error {
-		ln, err := net.Listen("tcp", "127.0.0.1:0")
-		if err != nil {
-			t.Fatal(err)
-		}
-		addr = ln.Addr().String()
-		_ = ln.Close()
-		s := celeris.New(celeris.Config{Addr: addr, Engine: eng, ReadTimeout: wt, WriteTimeout: wt, ShutdownTimeout: shutdown})
-		s.GET("/ping", func(c *celeris.Context) error { return c.String(http.StatusOK, "ok") })
-		s.GET("/events", sse.New(cfg))
-		st.s, st.startDone = s, make(chan error, 1)
-		go func(done chan error) { done <- s.Start() }(st.startDone)
-		if err := waitReady761(addr, st.startDone); err != nil {
-			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-			_ = s.Shutdown(ctx)
-			cancel()
-			return err
-		}
-		return nil
-	})
+	// start that fails with ring-setup ENOMEM is retried; the port stays with
+	// the server from the pick to the bind (celeris#1104).
+	var addr string
+	st.s, addr, st.startDone = startPicked1104(t, celeris.Config{Engine: eng, ReadTimeout: wt, WriteTimeout: wt, ShutdownTimeout: shutdown},
+		func(s *celeris.Server) {
+			s.GET("/ping", func(c *celeris.Context) error { return c.String(http.StatusOK, "ok") })
+			s.GET("/events", sse.New(cfg))
+		},
+		func(_ *celeris.Server, addr string, done <-chan error) error { return waitReady761(addr, done) })
 	st.addr = addr
 	t.Cleanup(func() {
 		if st.conn != nil {

test/server/start_picked_linux_test.go

@@ -0,0 +1,59 @@
+//go:build linux
+
+package celeris_test
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"testing"
+
+	"github.com/goceleris/celeris"
+)
+
+// listenForStart1104 gives startPicked1104 the listener it picks its port with.
+// A variable so a test can make two starters pick the same port, which two
+// parallel test cells of one process do now and then: net.Listen on ":0" hands a
+// port out again as soon as it is closed.
+var listenForStart1104 = func() (net.Listener, error) { return net.Listen("tcp", "127.0.0.1:0") }
+
+// startPicked1104 builds a server with cfg (Addr is set here) and routes, starts
+// it on a listener it picked, and returns it with its address and Start's result
+// channel once ready reports it up. The listener is handed to the server
+// (StartWithListener), so the port is never free between the pick and the bind:
+// another cell picking the same one gets a different port, and a readiness probe
+// can only be answered by this server. A ring ENOMEM is retried for the window
+// (startRetrying1104; a fresh server and port per try) and a start that fails
+// otherwise fails t. ready is given the server, its address and Start's result
+// channel and must return Start's error if it comes first.
+func startPicked1104(t testing.TB, cfg celeris.Config, routes func(*celeris.Server), ready func(s *celeris.Server, addr string, done <-chan error) error) (*celeris.Server, string, chan error) {
+	t.Helper()
+	var (
+		s    *celeris.Server
+		addr string
+		done chan error
+	)
+	startRetrying1104(t, "server start", func() error {
+		ln, err := listenForStart1104()
+		if err != nil {
+			t.Fatal(err)
+		}
+		addr = ln.Addr().String()
+		cfg.Addr = addr // the adaptive engine wants Addr to be its listener's address
+		s = celeris.New(cfg)
+		routes(s)
+		done = make(chan error, 1)
+		go func(s *celeris.Server, ln net.Listener, done chan error) { done <- s.StartWithListener(ln) }(s, ln, done)
+		if err := ready(s, addr, done); err != nil {
+			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			_ = s.Shutdown(ctx)
+			cancel()
+			_ = ln.Close()
+			return fmt.Errorf("Start returned before the server was ready: %w", err)
+		}
+		return nil
+	})
+	return s, addr, done
+}