From 2eb33a7c733df76a8708d4faeab93a139f311f5e Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 09:13:08 +0100 Subject: [PATCH 1/7] test/server: a test shows the start retry ignoring an address already in use (Refs #1104) --- test/server/ring_enomem_test.go | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/test/server/ring_enomem_test.go b/test/server/ring_enomem_test.go index 71026632d..a9ba86482 100644 --- a/test/server/ring_enomem_test.go +++ b/test/server/ring_enomem_test.go @@ -113,3 +113,32 @@ func TestRetryRingENOMEM1104(t *testing.T) { 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 starter that picks a free port, closes it and has Start bind it can lose +// the race: another test of the process, running in parallel, is given the port +// in between (a client socket of its own takes ephemeral ports from the same +// range). The SSE 855 starter retried any error for three seconds before the +// ENOMEM change narrowed it, and TestStalledSSEEndsWithShutdown855/std/h2c/budget +// failed once in CI with "startErr=listen tcp 127.0.0.1:38797: bind: address +// already in use". The starters that pick a fresh port per try retry that class +// too, a bounded number of 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 := retryRingENOMEM(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 = retryRingENOMEM(func() error { return inUse }, 5*time.Second) + if !errors.Is(err, inUse) || tries < 2 || tries > 20 { + t.Errorf("a port that stays taken: %d tries, err %v; want a bounded number of tries (2 to 20), then %v", tries, err, inUse) + } +} From 195c6514bcbecae01ffeec07a99c134d06517051 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 09:25:49 +0100 Subject: [PATCH 2/7] test/server: the start retry takes a fresh port when its port was rebound (Refs #1104) TestStalledSSEEndsWithShutdown855/std/h2c/budget failed once in CI with "startErr=listen tcp 127.0.0.1:38797: bind: address already in use": the SSE starter picks a free port, closes it and has Start bind it, parallel cells of the test run in the same process, and the test/server ENOMEM change had narrowed the loop that used to retry any error for three seconds to a ring ENOMEM only. retryRingENOMEM now also retries an address already in use, up to ten times (each try picks a fresh port; a port that stays taken is a real conflict), for every starter that goes through it: the SSE 855 starter, startTCPReady889 and the start-shutdown rig. --- test/server/ring_enomem_test.go | 34 ++++++++++++++++++++++++++++----- 1 file changed, 29 insertions(+), 5 deletions(-) diff --git a/test/server/ring_enomem_test.go b/test/server/ring_enomem_test.go index a9ba86482..d25fe462a 100644 --- a/test/server/ring_enomem_test.go +++ b/test/server/ring_enomem_test.go @@ -37,15 +37,39 @@ 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. +// isAddrInUse reports a Start error that is the bind race of a starter that +// picks a free port, closes it and has Start bind it: 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 + +// retryRingENOMEM 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 retryRingENOMEM(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) From 8ea95a9b54464be411a2f3457a4dd1b6a4089e9f Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 09:26:43 +0100 Subject: [PATCH 3/7] test/server: the shutdown gate test begins the shutdown once the engine has the conns and the requests (Refs #1104) TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885 slept a fixed 1.3 s and then began the shutdown, assuming the engine had accepted the silent conns and read the requests of the keep-alive, download and upload conns. On a loaded runner (-race and -cover, 24 HTTP/2 downloads of 16 MiB running) it had not, and the gate's first pass, 250 ms in, closed a conn the loop had not read yet: three resets at 257 ms in CI. It now waits, up to 20 s, until the engine counts every conn open and every whole request handled (waitEngineBusy1104). --- test/server/shutdown_h2_pool_linux_test.go | 32 ++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/test/server/shutdown_h2_pool_linux_test.go b/test/server/shutdown_h2_pool_linux_test.go index 931a621d5..73295118c 100644 --- a/test/server/shutdown_h2_pool_linux_test.go +++ b/test/server/shutdown_h2_pool_linux_test.go @@ -1270,6 +1270,24 @@ 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 lane, see the finding in the PR that added + // this wait). So begin only once the engine has accepted every conn and + // handled every request this test sent whole: 24 HTTP/2 GETs, 6 (7) + // keep-alive pings, and the download's GET. + wantConns := uint64(24 + len(idle) + len(silent) + 1) // + the download + wantReqs := uint64(24 + len(idle) + 1) + if arm == "sync" { + wantConns++ // the upload + } + if promoted != nil { + wantConns++ + wantReqs++ + } + waitEngineBusy1104(t, srv, wantConns, wantReqs) begin := time.Now() srv.beginShutdown("Shutdown", 20*time.Second) @@ -1344,3 +1362,17 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { }) } } + +// waitEngineBusy1104 waits, up to 20 s, until the engine counts at least conns +// open connections and reqs handled requests. +func waitEngineBusy1104(t *testing.T, srv *h2PoolServer759, 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) >= conns && m.RequestCount >= reqs { + return + } + } + t.Fatalf("the engine had %d open conns and %d requests handled, want at least %d and %d before the shutdown begins", m.ActiveConnections, m.RequestCount, conns, reqs) +} From a2e43a3166f8eca966a0eb52ac27875fcbc5d51a Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 09:45:54 +0100 Subject: [PATCH 4/7] test/server: the shutdown gate test waits for exact conns and requests over a baseline (Refs #1104) --- test/server/shutdown_h2_pool_linux_test.go | 51 ++++++++++++++++++---- 1 file changed, 42 insertions(+), 9 deletions(-) diff --git a/test/server/shutdown_h2_pool_linux_test.go b/test/server/shutdown_h2_pool_linux_test.go index 73295118c..bac244363 100644 --- a/test/server/shutdown_h2_pool_linux_test.go +++ b/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 { @@ -1274,10 +1278,18 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { // 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 lane, see the finding in the PR that added - // this wait). So begin only once the engine has accepted every conn and - // handled every request this test sent whole: 24 HTTP/2 GETs, 6 (7) - // keep-alive pings, and the download's GET. + // 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 whole (24 HTTP/2 GETs, one ping per keep-alive + // conn, the download's GET, the promoted conn's ping). The conn count + // is exact (+34 in every run). The request count runs at or one above the + // arithmetic (+32 against 31 on the sync arm, +32 or +33 against + // 32 on the async one, as counted at the instant the wait returns) and the source of that unit is not pinned down; + // it is waited for as a lower bound. The upload's request is not whole + // until its body is, so it is not counted: its headers and first 64 KiB + // may still be unread when the shutdown begins. wantConns := uint64(24 + len(idle) + len(silent) + 1) // + the download wantReqs := uint64(24 + len(idle) + 1) if arm == "sync" { @@ -1287,7 +1299,7 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { wantConns++ wantReqs++ } - waitEngineBusy1104(t, srv, wantConns, wantReqs) + waitEngineBusy1104(t, srv, base, wantConns, wantReqs) begin := time.Now() srv.beginShutdown("Shutdown", 20*time.Second) @@ -1363,16 +1375,37 @@ 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 -// open connections and reqs handled requests. -func waitEngineBusy1104(t *testing.T, srv *h2PoolServer759, conns, reqs uint64) { +// 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) >= conns && m.RequestCount >= reqs { + 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 had %d open conns and %d requests handled, want at least %d and %d before the shutdown begins", m.ActiveConnections, m.RequestCount, conns, reqs) + 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) } From fea8b18798d926d242e758b80364af664e4b03b6 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 09:46:26 +0100 Subject: [PATCH 5/7] test/server: a test shows two starters picking one port both ending on it (Refs #1104) startPicked1104 is the pick, close and Start the SSE 855 and h2upgrade-matrix starters did inline, in one place (behaviour unchanged). The test makes two starters pick the same port at the same moment: one server fails to bind it, and the other's /ping answers the loser's readiness probe, so both starters return the same address (10 of 10 runs fail). --- .../server/h2upgrade_matrix_889_linux_test.go | 31 ++----- .../sse_detach_stalled_855_linux_test.go | 31 ++----- test/server/start_picked_linux_test.go | 56 ++++++++++++ test/server/start_picked_race_linux_test.go | 90 +++++++++++++++++++ 4 files changed, 164 insertions(+), 44 deletions(-) create mode 100644 test/server/start_picked_linux_test.go create mode 100644 test/server/start_picked_race_linux_test.go diff --git a/test/server/h2upgrade_matrix_889_linux_test.go b/test/server/h2upgrade_matrix_889_linux_test.go index d7c72ff0e..3ae550489 100644 --- a/test/server/h2upgrade_matrix_889_linux_test.go +++ b/test/server/h2upgrade_matrix_889_linux_test.go @@ -6,6 +6,7 @@ import ( "bufio" "bytes" "context" + "errors" "fmt" "io" "net" @@ -176,30 +177,19 @@ 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) @@ -208,9 +198,6 @@ func startTCPReady889(t *testing.T, cfg celeris.Config) string { 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) } time.Sleep(50 * time.Millisecond) diff --git a/test/server/sse_detach_stalled_855_linux_test.go b/test/server/sse_detach_stalled_855_linux_test.go index d1f872f90..f7f5413ad 100644 --- a/test/server/sse_detach_stalled_855_linux_test.go +++ b/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 { diff --git a/test/server/start_picked_linux_test.go b/test/server/start_picked_linux_test.go new file mode 100644 index 000000000..b20c0e819 --- /dev/null +++ b/test/server/start_picked_linux_test.go @@ -0,0 +1,56 @@ +//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 port it picked, and returns it with its address and Start's result +// channel once ready reports it up. A ring ENOMEM is retried for the window +// (startRetryingRingENOMEM; a fresh server and port per try) and a start that +// fails otherwise fails t. ready is given the server 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 + ) + startRetryingRingENOMEM(t, "server start", func() error { + ln, err := listenForStart1104() + if err != nil { + t.Fatal(err) + } + addr = ln.Addr().String() + _ = ln.Close() + cfg.Addr = addr + s = celeris.New(cfg) + routes(s) + done = make(chan error, 1) + go func(s *celeris.Server, done chan error) { done <- s.Start() }(s, done) + if err := ready(s, addr, done); err != nil { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + _ = s.Shutdown(ctx) + cancel() + return fmt.Errorf("Start returned before the server was ready: %w", err) + } + return nil + }) + return s, addr, done +} diff --git a/test/server/start_picked_race_linux_test.go b/test/server/start_picked_race_linux_test.go new file mode 100644 index 000000000..fc81e7049 --- /dev/null +++ b/test/server/start_picked_race_linux_test.go @@ -0,0 +1,90 @@ +//go:build linux + +package celeris_test + +import ( + "context" + "fmt" + "io" + "net" + "net/http" + "sync" + "testing" + "time" + + "github.com/goceleris/celeris" +) + +// celeris#1104: two parallel cells of TestStalledSSEEndsWithShutdown855 picked +// the same port 0.5 ms apart (the CI log shows two "celeris starting +// addr=127.0.0.1:38797" lines): one bound it, the other's Start failed with +// "address already in use", and that error surfaced at the end of the cell +// because the loser's readiness probe had been answered by the WINNER's /ping. +// A starter that picks a port, closes it and has Start bind it cannot see that; +// it has to keep the port from the pick to the bind. Here two starters are made +// to pick the very same port at the same moment, again and again, and each must +// end up with a server that answers as itself, on a port of its own. +func TestTwoStartsRacingForOnePortEndOnPortsOfTheirOwn1104(t *testing.T) { + pick, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + port := pick.Addr().String() + _ = pick.Close() + defer func(old func() (net.Listener, error)) { listenForStart1104 = old }(listenForStart1104) + // The first picker is given the port; a picker that finds it taken (held by + // a server) gets a fresh one, as another test's pick would. + listenForStart1104 = func() (net.Listener, error) { + if ln, err := net.Listen("tcp", port); err == nil { + return ln, nil + } + return net.Listen("tcp", "127.0.0.1:0") + } + deadline := time.Now().Add(4 * time.Second) + for round := 0; round < 40 && time.Now().Before(deadline); round++ { + var wg sync.WaitGroup + addrs := make([]string, 2) + answers := make([]string, 2) + srvs := make([]*celeris.Server, 2) + for i := range srvs { + wg.Add(1) + go func() { + defer wg.Done() + token := fmt.Sprintf("server-%d-%d", round, i) + s, addr, _ := startPicked1104(t, celeris.Config{Engine: celeris.Std}, func(s *celeris.Server) { + s.GET("/ping", func(c *celeris.Context) error { return c.String(http.StatusOK, "%s", token) }) + }, func(_ *celeris.Server, addr string, done <-chan error) error { return waitReady761(addr, done) }) + srvs[i], addrs[i] = s, addr + resp, err := http.Get("http://" + addr + "/ping") + if err != nil { + answers[i] = err.Error() + return + } + b, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + answers[i] = string(b) + }() + } + wg.Wait() + for i := range srvs { + if want := fmt.Sprintf("server-%d-%d", round, i); answers[i] != want { + t.Errorf("round %d: server %d at %s answered %q, want %q (its own /ping)", round, i, addrs[i], answers[i], want) + } + } + if addrs[0] == addrs[1] { + t.Errorf("round %d: both starters ended on %s: one of the two servers failed to bind it and the other's /ping answered for it", round, addrs[0]) + } + // Free the port for the next round: a pick that finds it still held gets + // a fresh one and the race would not recur. + for _, s := range srvs { + if s != nil { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + _ = s.Shutdown(ctx) + cancel() + } + } + if t.Failed() { + return + } + } +} From 931f6c5710a2266820d2f333895efbc3ce814cf5 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 09:55:31 +0100 Subject: [PATCH 6/7] test/server: the SSE and h2upgrade starters keep their listener, the shutdown rig tells its own /ping (Refs #1104) startPicked1104 now hands the picked listener to the server (StartWithListener): the port is never free between pick and bind, so a second starter that picks the same port gets another one and a readiness probe can only be answered by the server that owns the port. The h2upgrade readiness waits for the engine to publish its address, since a dial alone proves nothing against a listener the server holds from the start. The start-shutdown rig must keep Start for its plain-Start arm, so its /ping returns a per-server token that waitReadyErr compares; an answer from another server is not ready and Start's bind error is what the retry sees. retryRingENOMEM and startRetryingRingENOMEM are renamed retryStart1104 and startRetrying1104, the log line says what was retried, and the new bound test pins exactly 10 tries. --- .../server/h2upgrade_matrix_889_linux_test.go | 14 ++++++---- test/server/ring_enomem_test.go | 26 ++++++++--------- .../server_async_goexit_791_linux_test.go | 2 +- test/server/start_picked_linux_test.go | 21 ++++++++------ test/server/start_picked_race_linux_test.go | 28 +++++++++++++++++++ .../start_shutdown_return_linux_test.go | 22 ++++++++++----- 6 files changed, 77 insertions(+), 36 deletions(-) diff --git a/test/server/h2upgrade_matrix_889_linux_test.go b/test/server/h2upgrade_matrix_889_linux_test.go index 3ae550489..2aa92a2ff 100644 --- a/test/server/h2upgrade_matrix_889_linux_test.go +++ b/test/server/h2upgrade_matrix_889_linux_test.go @@ -7,7 +7,6 @@ import ( "bytes" "context" "errors" - "fmt" "io" "net" "net/http" @@ -192,13 +191,16 @@ func startTCPReady889(t *testing.T, cfg celeris.Config) string { 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 { - 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) } diff --git a/test/server/ring_enomem_test.go b/test/server/ring_enomem_test.go index d25fe462a..f1a203734 100644 --- a/test/server/ring_enomem_test.go +++ b/test/server/ring_enomem_test.go @@ -48,14 +48,14 @@ func isAddrInUse(err error) bool { // taken is a real conflict, not the race. const maxAddrInUseTries = 10 -// retryRingENOMEM calls attempt until it returns nil or an error it should not +// 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 retryRingENOMEM(attempt func() error, window time.Duration) (tries int, err error) { +func retryStart1104(attempt func() error, window time.Duration) (tries int, err error) { deadline := time.Now().Add(window) inUse := 0 for { @@ -76,16 +76,16 @@ func retryRingENOMEM(attempt func() error, window time.Duration) (tries int, err } } -// 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) } } @@ -112,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 @@ -125,14 +125,14 @@ 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) } @@ -150,7 +150,7 @@ 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 := retryRingENOMEM(func() error { + tries, err := retryStart1104(func() error { calls++ if calls < 3 { return inUse @@ -161,8 +161,8 @@ func TestRetryStartRetriesAnAddressInUse1104(t *testing.T) { t.Errorf("two taken ports then success: %d tries, err %v; want 3, nil", tries, err) } - tries, err = retryRingENOMEM(func() error { return inUse }, 5*time.Second) - if !errors.Is(err, inUse) || tries < 2 || tries > 20 { - t.Errorf("a port that stays taken: %d tries, err %v; want a bounded number of tries (2 to 20), then %v", tries, err, inUse) + 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) } } diff --git a/test/server/server_async_goexit_791_linux_test.go b/test/server/server_async_goexit_791_linux_test.go index d945ffae3..a197e3573 100644 --- a/test/server/server_async_goexit_791_linux_test.go +++ b/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) diff --git a/test/server/start_picked_linux_test.go b/test/server/start_picked_linux_test.go index b20c0e819..ce13e6ae1 100644 --- a/test/server/start_picked_linux_test.go +++ b/test/server/start_picked_linux_test.go @@ -20,11 +20,14 @@ import ( 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 port it picked, and returns it with its address and Start's result -// channel once ready reports it up. A ring ENOMEM is retried for the window -// (startRetryingRingENOMEM; a fresh server and port per try) and a start that -// fails otherwise fails t. ready is given the server and Start's result channel -// and must return Start's error if it comes first. +// 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 ( @@ -32,22 +35,22 @@ func startPicked1104(t testing.TB, cfg celeris.Config, routes func(*celeris.Serv addr string done chan error ) - startRetryingRingENOMEM(t, "server start", func() error { + startRetrying1104(t, "server start", func() error { ln, err := listenForStart1104() if err != nil { t.Fatal(err) } addr = ln.Addr().String() - _ = ln.Close() - cfg.Addr = addr + 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, done chan error) { done <- s.Start() }(s, done) + 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 diff --git a/test/server/start_picked_race_linux_test.go b/test/server/start_picked_race_linux_test.go index fc81e7049..227e54388 100644 --- a/test/server/start_picked_race_linux_test.go +++ b/test/server/start_picked_race_linux_test.go @@ -88,3 +88,31 @@ func TestTwoStartsRacingForOnePortEndOnPortsOfTheirOwn1104(t *testing.T) { } } } + +// A readiness probe answered by another server must not count: with a non-empty +// want, waitReadyErr takes only the body of the server's own /ping, and returns +// Start's error when that arrives instead. +func TestWaitReadyErrDoesNotTakeAnotherServersAnswer1104(t *testing.T) { + other := http.NewServeMux() + other.HandleFunc("/ping", func(w http.ResponseWriter, _ *http.Request) { _, _ = io.WriteString(w, "ok-other") }) + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + srv := &http.Server{Handler: other} + go func() { _ = srv.Serve(ln) }() + defer func() { _ = srv.Close() }() + + bindErr := fmt.Errorf("listen tcp %s: bind: address already in use", ln.Addr()) + done := make(chan error, 1) + go func() { time.Sleep(100 * time.Millisecond); done <- bindErr }() + ok, err := waitReadyErr(&http.Client{Timeout: 300 * time.Millisecond}, "http://"+ln.Addr().String()+"/ping", "ok-mine", done) + if ok || err == nil || !isAddrInUse(err) { + t.Fatalf("waitReadyErr = %v, %v; want not ready and Start's error (the other server's answer is not this one's)", ok, err) + } + // Its own answer is ready. + ok, err = waitReadyErr(&http.Client{Timeout: 300 * time.Millisecond}, "http://"+ln.Addr().String()+"/ping", "ok-other", make(chan error)) + if !ok || err != nil { + t.Fatalf("waitReadyErr with the server's own body = %v, %v; want ready", ok, err) + } +} diff --git a/test/server/start_shutdown_return_linux_test.go b/test/server/start_shutdown_return_linux_test.go index 856006320..3094b5d41 100644 --- a/test/server/start_shutdown_return_linux_test.go +++ b/test/server/start_shutdown_return_linux_test.go @@ -93,7 +93,7 @@ func runStartShutdownCaseCfg(t *testing.T, cfg celeris.Config, withListener bool // A fresh server and listener 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 { ln, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatalf("listen: %v", err) @@ -117,7 +117,12 @@ func runStartShutdownCaseCfg(t *testing.T, cfg celeris.Config, withListener bool cfg.Addr = addr } s = celeris.New(cfg) - s.GET("/ping", func(c *celeris.Context) error { return c.String(http.StatusOK, "ok") }) + // Start picks nothing here when !withListener, but the port was picked and + // closed by this test: a parallel test can take it first, and then the + // OTHER server's /ping would answer this one's readiness probe while this + // Start failed. /ping therefore says whose it is (celeris#1104). + token := fmt.Sprintf("ok-%d", time.Now().UnixNano()) + s.GET("/ping", func(c *celeris.Context) error { return c.String(http.StatusOK, "%s", token) }) s.GET("/slow", func(c *celeris.Context) error { select { case handlerStarted <- struct{}{}: @@ -138,7 +143,7 @@ func runStartShutdownCaseCfg(t *testing.T, cfg celeris.Config, withListener bool // engine is up, the probe listener's backlog still accepts the TCP // connection and nobody answers, so a long timeout would park the poll // loop and hide a Start that already returned an error. - ok, werr := waitReadyErr(&http.Client{Timeout: 300 * time.Millisecond}, "http://"+addr+"/ping", startDone) + ok, werr := waitReadyErr(&http.Client{Timeout: 300 * time.Millisecond}, "http://"+addr+"/ping", token, startDone) if werr != nil { if withListener { _ = ln.Close() @@ -214,7 +219,7 @@ func runStartShutdownCaseCfg(t *testing.T, cfg celeris.Config, withListener bool // a silent skip would look like a pass). func waitReady(t *testing.T, client *http.Client, url string, startDone <-chan error) bool { t.Helper() - ok, err := waitReadyErr(client, url, startDone) + ok, err := waitReadyErr(client, url, "", startDone) if err != nil { t.Fatalf("Start returned before the server was ready: %v", err) } @@ -223,7 +228,10 @@ func waitReady(t *testing.T, client *http.Client, url string, startDone <-chan e // waitReadyErr is waitReady that returns the error Start returned first // instead of failing, so a caller can retry a ring-setup ENOMEM (celeris#1104). -func waitReadyErr(client *http.Client, url string, startDone <-chan error) (bool, error) { +// A non-empty want is the body the server's own /ping returns: an answer with +// another body is another server's (one that holds the port this one failed to +// bind) and does not count as ready, so Start's error is what is returned. +func waitReadyErr(client *http.Client, url, want string, startDone <-chan error) (bool, error) { deadline := time.Now().Add(8 * time.Second) for time.Now().Before(deadline) { select { @@ -233,9 +241,9 @@ func waitReadyErr(client *http.Client, url string, startDone <-chan error) (bool } resp, err := client.Get(url) if err == nil { - _, _ = io.Copy(io.Discard, resp.Body) + b, _ := io.ReadAll(resp.Body) _ = resp.Body.Close() - if resp.StatusCode == http.StatusOK { + if resp.StatusCode == http.StatusOK && (want == "" || string(b) == want) { return true, nil } } From fb7ae80967ae7ce971750424ee0e8acec7896e1f Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 10:07:00 +0100 Subject: [PATCH 7/7] test/server: the gate test waits for the upload's request too; stale retry comments (Refs #1104) On the sync arm the loop counts the upload once it has read its headers and first 64 KiB, so the request count to wait for is 32, which also covers the upload having been read (the case the gate defect cuts). The one extra unit seen in some runs is already present before the download and the upload are sent and its source was not found, so the request wait stays a lower bound while the conn wait is exact. The comments of the retry tests and of isAddrInUse describe the start-shutdown rig's case, not the superseded diagnosis. --- test/server/ring_enomem_test.go | 21 +++++++++++---------- test/server/shutdown_h2_pool_linux_test.go | 21 ++++++++++++--------- 2 files changed, 23 insertions(+), 19 deletions(-) diff --git a/test/server/ring_enomem_test.go b/test/server/ring_enomem_test.go index f1a203734..833e23d39 100644 --- a/test/server/ring_enomem_test.go +++ b/test/server/ring_enomem_test.go @@ -38,8 +38,8 @@ func isRingENOMEM(err error) bool { } // isAddrInUse reports a Start error that is the bind race of a starter that -// picks a free port, closes it and has Start bind it: another test of the -// process was given the port in between. +// 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") } @@ -138,14 +138,15 @@ func TestRetryRingENOMEM1104(t *testing.T) { } } -// A starter that picks a free port, closes it and has Start bind it can lose -// the race: another test of the process, running in parallel, is given the port -// in between (a client socket of its own takes ephemeral ports from the same -// range). The SSE 855 starter retried any error for three seconds before the -// ENOMEM change narrowed it, and TestStalledSSEEndsWithShutdown855/std/h2c/budget -// failed once in CI with "startErr=listen tcp 127.0.0.1:38797: bind: address -// already in use". The starters that pick a fresh port per try retry that class -// too, a bounded number of times: a port that stays taken is a real conflict. +// 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") diff --git a/test/server/shutdown_h2_pool_linux_test.go b/test/server/shutdown_h2_pool_linux_test.go index bac244363..8410939b0 100644 --- a/test/server/shutdown_h2_pool_linux_test.go +++ b/test/server/shutdown_h2_pool_linux_test.go @@ -1282,18 +1282,21 @@ func TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885(t *testing.T) { // 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 whole (24 HTTP/2 GETs, one ping per keep-alive - // conn, the download's GET, the promoted conn's ping). The conn count - // is exact (+34 in every run). The request count runs at or one above the - // arithmetic (+32 against 31 on the sync arm, +32 or +33 against - // 32 on the async one, as counted at the instant the wait returns) and the source of that unit is not pinned down; - // it is waited for as a lower bound. The upload's request is not whole - // until its body is, so it is not counted: its headers and first 64 KiB - // may still be unread when the shutdown begins. + // 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) + 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++