From c3f8f4289c5212e224a913be48a1b90a66699333 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 20:42:40 +0100 Subject: [PATCH 1/3] tests: TestNonReadingClientEndsStream888 says whether a full queue ended the stream before WriteTimeout (diagnostic; Refs #888, Refs #1104) --- .../sse/sse_nonreading_888_linux_test.go | 21 ++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/middleware/sse/sse_nonreading_888_linux_test.go b/middleware/sse/sse_nonreading_888_linux_test.go index c37e3f699..462ecaae5 100644 --- a/middleware/sse/sse_nonreading_888_linux_test.go +++ b/middleware/sse/sse_nonreading_888_linux_test.go @@ -3,6 +3,7 @@ package sse_test import ( + "fmt" "net" "os" "strconv" @@ -39,6 +40,11 @@ const nonReadWriteTimeout888 = time.Second type disconnect888 struct { at time.Time early int64 + // slowClose is when OnSlowClient answered ClientPolicyClose for the first + // time (zero: it never did), and started when the handler began: a stream + // ended by a full queue is not one ended by the stall bound. + slowClose time.Time + started time.Time } // nonReadCell888 is one engine configuration of the matrix. @@ -149,7 +155,7 @@ func burstFor888(tb testing.TB) int { func serve888(tb testing.TB, cell nonReadCell888, p producer888) (addr string, disconnected <-chan disconnect888, shutdown func()) { tb.Helper() done := make(chan disconnect888, 1) - var started, early atomic.Int64 // handler start (unix nanos), events dropped by dropEarly + var started, early, slowClose atomic.Int64 // handler start (unix nanos), events dropped by dropEarly, first ClientPolicyClose (unix nanos) wt := p.wt() var burst string if p.fastFor == 0 { @@ -164,11 +170,16 @@ func serve888(tb testing.TB, cell nonReadCell888, p producer888) (addr string, d early.Add(1) return sse.ClientPolicyDrop } + slowClose.CompareAndSwap(0, time.Now().UnixNano()) return sse.ClientPolicyClose }, OnDisconnect: func(*celeris.Context, *sse.Client) { + d := disconnect888{at: time.Now(), early: early.Load(), started: time.Unix(0, started.Load())} + if n := slowClose.Load(); n != 0 { + d.slowClose = time.Unix(0, n) + } select { - case done <- disconnect888{time.Now(), early.Load()}: + case done <- d: default: } }, @@ -312,7 +323,11 @@ func TestNonReadingClientEndsStream888(t *testing.T) { // queue behind it is full (ClientPolicyClose), earlier than // WriteTimeout: the lower bound is for the native engines. if cell.engine != celeris.Std && el < wt { - t.Errorf("OnDisconnect ran %v after connect, before WriteTimeout %v: the stream was ended by something other than the stall bound", el, wt) + why := "OnSlowClient never answered ClientPolicyClose" + if !d.slowClose.IsZero() { + why = fmt.Sprintf("OnSlowClient answered ClientPolicyClose %v after the handler began (a full queue)", d.slowClose.Sub(d.started).Round(time.Millisecond)) + } + t.Errorf("OnDisconnect ran %v after connect, before WriteTimeout %v: the stream was ended by something other than the stall bound (%s)", el, wt, why) } case <-time.After(wt + slack): t.Fatalf("OnDisconnect never ran %v after connect (WriteTimeout %v): a client that reads nothing holds its stream for ever (celeris#888 item 38)", From d51b748a1c77ba313cb18abc324ddec656e93262 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 21:27:58 +0100 Subject: [PATCH 2/3] tests: TestNonReadingClientEndsStream888 gets a producer whose queue overflows before WriteTimeout (red on main: the stream ends at 0.8 s; Refs #888, Refs #1104) --- middleware/sse/sse_nonreading_888_linux_test.go | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/middleware/sse/sse_nonreading_888_linux_test.go b/middleware/sse/sse_nonreading_888_linux_test.go index 462ecaae5..3f9a8bc8f 100644 --- a/middleware/sse/sse_nonreading_888_linux_test.go +++ b/middleware/sse/sse_nonreading_888_linux_test.go @@ -101,6 +101,11 @@ type producer888 struct { fastFor time.Duration trickleBytes int trickleEvery time.Duration + // flood is how many trickle events are sent back to back when the settle + // ends, before the trickle: the 4-slot queue cannot take them all, which is + // what a drain goroutine that was not scheduled for a few milliseconds + // (a starved runner under -race and coverage) does to the producer's pace. + flood int // writeTimeout is the server's WriteTimeout (nonReadWriteTimeout888 when // zero). noHeaderTimer sets ReadHeaderTimeout -1, which takes away the @@ -208,6 +213,11 @@ func serve888(tb testing.TB, cell nonReadCell888, p producer888) (addr string, d return case <-time.After(p.settle): } + for range p.flood { + if err := client.Send(sse.Event{Event: "tick", Data: trickle}); err != nil { + return + } + } tick := time.NewTicker(p.trickleEvery) defer tick.Stop() for { @@ -292,6 +302,11 @@ func TestNonReadingClientEndsStream888(t *testing.T) { // The burst fills the socket and queues the rest in the engine; then // a trickle of 1 KB every 5 ms. {"steady", producer888{burstBytes: burstFor888(t), settle: 700 * time.Millisecond, trickleBytes: 1000, trickleEvery: 5 * time.Millisecond}}, + // The same, with 50 events sent back to back where the trickle starts: a + // full queue before WriteTimeout (a drain goroutine that was not + // scheduled for a few ms does that to the 5 ms trickle too: celeris#1104) + // must not end the stream. + {"steady-flood", producer888{burstBytes: burstFor888(t), settle: 700 * time.Millisecond, trickleBytes: 1000, trickleEvery: 5 * time.Millisecond, flood: 50}}, // The same, then one event every 300 ms: a sweep must still reach the // conn when almost nothing wakes the worker. {"sparse", producer888{burstBytes: burstFor888(t), settle: 700 * time.Millisecond, trickleBytes: 1000, trickleEvery: 300 * time.Millisecond}}, From 1fe81184cc91b85c935601d341bdc09e55d0b4a4 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Wed, 28 Oct 2026 21:29:20 +0100 Subject: [PATCH 3/3] tests: the non-reading SSE producers drop on a full queue for the first WriteTimeout, so the lower bound tests the stall bound only (Refs #888, Refs #884, Refs #1104) --- middleware/sse/sse_nonreading_888_linux_test.go | 12 ++++++++---- middleware/sse/sse_nonreading_h2c_884_linux_test.go | 4 ++-- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/middleware/sse/sse_nonreading_888_linux_test.go b/middleware/sse/sse_nonreading_888_linux_test.go index 3f9a8bc8f..05f71eb7d 100644 --- a/middleware/sse/sse_nonreading_888_linux_test.go +++ b/middleware/sse/sse_nonreading_888_linux_test.go @@ -115,7 +115,11 @@ type producer888 struct { noHeaderTimer bool // dropEarly makes a full queue drop the event, not close the stream, for - // the first WriteTimeout of the handler's run. The fast phase sends 2000 + // the first WriteTimeout of the handler's run. Every producer sets it: the + // test's lower bound (the stream does not end before WriteTimeout) is for + // the stall bound, and a close by a full queue is a different one (CI saw + // the trickle of "steady" cut at 0.82 s, and so did a run of this test with + // two processes on one CPU: celeris#1104). The fast phase sends 2000 // events a second; a drain goroutine that falls 4 events behind (the loop // holds the connection's write lock for a partial send, or the runner is // starved under -race and coverage) is a full queue, and @@ -301,15 +305,15 @@ func TestNonReadingClientEndsStream888(t *testing.T) { }{ // The burst fills the socket and queues the rest in the engine; then // a trickle of 1 KB every 5 ms. - {"steady", producer888{burstBytes: burstFor888(t), settle: 700 * time.Millisecond, trickleBytes: 1000, trickleEvery: 5 * time.Millisecond}}, + {"steady", producer888{burstBytes: burstFor888(t), settle: 700 * time.Millisecond, trickleBytes: 1000, trickleEvery: 5 * time.Millisecond, dropEarly: true}}, // The same, with 50 events sent back to back where the trickle starts: a // full queue before WriteTimeout (a drain goroutine that was not // scheduled for a few ms does that to the 5 ms trickle too: celeris#1104) // must not end the stream. - {"steady-flood", producer888{burstBytes: burstFor888(t), settle: 700 * time.Millisecond, trickleBytes: 1000, trickleEvery: 5 * time.Millisecond, flood: 50}}, + {"steady-flood", producer888{burstBytes: burstFor888(t), settle: 700 * time.Millisecond, trickleBytes: 1000, trickleEvery: 5 * time.Millisecond, flood: 50, dropEarly: true}}, // The same, then one event every 300 ms: a sweep must still reach the // conn when almost nothing wakes the worker. - {"sparse", producer888{burstBytes: burstFor888(t), settle: 700 * time.Millisecond, trickleBytes: 1000, trickleEvery: 300 * time.Millisecond}}, + {"sparse", producer888{burstBytes: burstFor888(t), settle: 700 * time.Millisecond, trickleBytes: 1000, trickleEvery: 300 * time.Millisecond, dropEarly: true}}, // No burst: 4 KB events every 0.5 ms for 1.2 s fill the socket with // small sends (an io_uring SEND of under 64 KiB is not one the worker // is told to expect to wait for), then one event every 300 ms. diff --git a/middleware/sse/sse_nonreading_h2c_884_linux_test.go b/middleware/sse/sse_nonreading_h2c_884_linux_test.go index ae47201eb..09f2780a1 100644 --- a/middleware/sse/sse_nonreading_h2c_884_linux_test.go +++ b/middleware/sse/sse_nonreading_h2c_884_linux_test.go @@ -118,8 +118,8 @@ func TestNonReadingH2CClientEndsStream884(t *testing.T) { window uint32 p producer888 }{ - {"default-window", 65535, producer888{burstBytes: 1000, settle: 100 * time.Millisecond, trickleBytes: 1000, trickleEvery: 5 * time.Millisecond}}, - {"huge-window", 1 << 30, producer888{burstBytes: burstFor888(t), settle: 700 * time.Millisecond, trickleBytes: 1000, trickleEvery: 5 * time.Millisecond}}, + {"default-window", 65535, producer888{burstBytes: 1000, settle: 100 * time.Millisecond, trickleBytes: 1000, trickleEvery: 5 * time.Millisecond, dropEarly: true}}, + {"huge-window", 1 << 30, producer888{burstBytes: burstFor888(t), settle: 700 * time.Millisecond, trickleBytes: 1000, trickleEvery: 5 * time.Millisecond, dropEarly: true}}, } for _, cell := range cells { for _, w := range wins {