middleware/sse/sse.go (1)

194-220: Optional: harden stopHeartbeat against double-invocation.

The returned closure performs close(stop), which panics if called twice. Today it's only invoked from a single deferred call site, so this is fine — but a sync.Once (or already-closed sentinel) would make the helper resilient against future refactors that might call it from multiple places.

♻️ Proposed refactor

 func (s *Stream) startHeartbeat(interval time.Duration) func() {
     if interval <= 0 {
         return nil
     }

     stop := make(chan struct{})
     go func() {
         ticker := time.NewTicker(interval)
         defer ticker.Stop()
         for {
             select {
             case <-ticker.C:
                 if err := s.Comment(""); err != nil {
                     return
                 }
             case <-stop:
                 return
             case <-s.Done():
                 return
             }
         }
     }()

-    return func() {
-        close(stop)
-    }
+    var once sync.Once
+    return func() {
+        once.Do(func() { close(stop) })
+    }
 }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@middleware/sse/sse.go` around lines 194 - 220, The stop closure returned by
startHeartbeat currently calls close(stop) which will panic if invoked more than
once; modify startHeartbeat to make the stop operation idempotent (e.g., wrap
the close(stop) call with a sync.Once or check an already-closed sentinel) so
calling the returned function multiple times is safe; update the returned
closure in startHeartbeat to use that sync.Once (or atomic/boolean guard) to
ensure close(stop) executes only once while leaving the existing ticker stop and
goroutine exit behavior unchanged.

middleware/sse/sse_test.go (2)

126-147: Potentially flaky timing on slow CI.

HeartbeatInterval: 10 * time.Millisecond with a 30ms handler window leaves only ~2 expected ticks. On a slow/contended CI runner (or a test binary started under -race), the first tick may arrive after the handler returns, leaving the response body without :\n\n and failing the assertion. Consider widening the window (e.g., 5–10× interval) or polling for the heartbeat marker.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@middleware/sse/sse_test.go` around lines 126 - 147, The test
Test_SSE_NewWritesHeartbeat is flaky because HeartbeatInterval is 10ms but the
Handler only waits 30ms, giving very few ticks on slow CI; update the test to
either increase the handler wait to at least 5–10× HeartbeatInterval (e.g., set
the sleep inside Handler to 50–100ms) or implement a small polling loop that
reads resp.Body until the heartbeat marker ":\n\n" appears (retrying for a short
timeout) to reliably detect the heartbeat; reference the
Test_SSE_NewWritesHeartbeat function, New(Config{HeartbeatInterval: ... ,
Handler: ...}) and stream.Done()/stream.Err() when making the change.

---

268-272: Minor: add a timeout around the <-closed receive to avoid hanging the suite on regression.

If the middleware ever stops invoking OnClose, this test will deadlock indefinitely instead of failing fast. A select with a deadline is consistent with the timeout patterns already used elsewhere in this file (e.g., Test_SSE_StreamContextCanceledOnClose, Test_SSE_StreamWriteError).

♻️ Proposed change

 	resp, err := app.Test(httptest.NewRequest(fiber.MethodGet, "/events", http.NoBody))
 	require.NoError(t, err)
 	require.Equal(t, fiber.StatusOK, resp.StatusCode)
-	require.ErrorIs(t, <-closed, handlerErr)
+	select {
+	case err := <-closed:
+		require.ErrorIs(t, err, handlerErr)
+	case <-time.After(time.Second):
+		t.Fatal("OnClose was not invoked")
+	}
 }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@middleware/sse/sse_test.go` around lines 268 - 272, Test can hang waiting on
the closed channel; replace the direct receive "require.ErrorIs(t, <-closed,
handlerErr)" with a select that waits for either a value from the "closed"
channel (asserting ErrorIs against "handlerErr") or a time.After timeout
(failing the test fast). Update imports to include "time" and use the same
timeout duration pattern already used in other tests (e.g.,
Test_SSE_StreamContextCanceledOnClose) so the test fails instead of deadlocking.

docs/middleware/sse.md (1)

117-117: Nit: "Context() are closed" — contexts are canceled, not closed.

Suggest tweaking the wording for accuracy with Go's context vocabulary.

📝 Proposed wording

-Every write is flushed. A failed flush closes `Done`, stores the error returned by `Err`, and lets the handler stop without relying on `fasthttp.RequestCtx.Done`, which is not a per-client disconnect signal. After a normal handler return, `Done` and `Context()` are closed while `Err()` remains `nil`; writes after that return `sse: stream closed`.
+Every write is flushed. A failed flush closes `Done`, stores the error returned by `Err`, and lets the handler stop without relying on `fasthttp.RequestCtx.Done`, which is not a per-client disconnect signal. After a normal handler return, `Done` is closed and `Context()` is canceled while `Err()` remains `nil`; writes after that return `sse: stream closed`.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@docs/middleware/sse.md` at line 117, Update the sentence that currently says
"After a normal handler return, `Done` and `Context()` are closed while `Err()`
remains `nil`;" to use correct Go context terminology: indicate that `Done` is
closed but `Context()` is canceled (not closed) while `Err()` remains `nil`.
Reference the symbols `Done`, `Context()`, and `Err()` so readers see the exact
change.

middleware/sse/event.go (1)

89-106: Minor: untyped nil Data produces a data: \n line, unlike typed nil.

When Event.Data is untyped nil, eventData returns "", and writeData still emits a single data: \n line (because strings.SplitSeq("", "\n") yields one empty element). For a typed-nil pointer (the case covered by Test_SSE_EventJSONEncodesTypedNilStringer) it falls through to json.Marshal and emits data: null. Consider documenting this asymmetry or treating nil as "no data line" for parity. Not a correctness issue, just surprising for callers who pass Data: nil expecting either data: null or no data line at all.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@middleware/sse/event.go` around lines 89 - 106, eventData currently returns
an empty string for an untyped nil which causes writeData to emit a spurious
empty "data: \n" line; change eventData to signal "no data" (e.g. change
signature to return (string, bool, error) where the bool indicates whether there
is data) and update writeData to skip emitting any data lines when that bool is
false; keep handling for typed nils (they should still be marshaled to "null")
and ensure Event.Data is the referenced symbol to locate callers.

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

middleware/sse/sse.go (1)

198-227: Heartbeat goroutine is not joined on stop — residual late-error window remains.

stopHeartbeat only closes the stop channel; it does not wait for the goroutine to exit. With defer order stopHeartbeat → closeStream → OnClose, the heartbeat goroutine can still be mid-s.Comment("") (or about to enter it) while OnClose reads stream.Err() on line 53. If the final tick's write fails and latches s.err after that read, the disconnect error is still missed by OnClose. The mutex serializes writes, so there's no data corruption — only a narrow observability gap.

Joining the goroutine before OnClose runs would close the gap deterministically:

♻️ Optional: join the heartbeat goroutine on stop

 func (s *Stream) startHeartbeat(interval time.Duration) func() {
 	if interval <= 0 {
 		return nil
 	}

 	stop := make(chan struct{})
 	var stopOnce sync.Once
+	done := make(chan struct{})
 	go func() {
+		defer close(done)
 		ticker := time.NewTicker(interval)
 		defer ticker.Stop()
 		for {
 			select {
 			case <-ticker.C:
 				if err := s.Comment(""); err != nil {
 					return
 				}
 			case <-stop:
 				return
 			case <-s.Done():
 				return
 			}
 		}
 	}()

 	return func() {
 		stopOnce.Do(func() {
 			close(stop)
 		})
+		<-done
 	}
 }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@middleware/sse/sse.go` around lines 198 - 227, The heartbeat goroutine
started by Stream.startHeartbeat can still be executing s.Comment("") after the
returned stop function (stopHeartbeat) closes stop, causing a race where OnClose
may read s.Err() before the goroutine finishes and sets the final error; fix
this by making the stop function block until the goroutine has fully exited:
inside startHeartbeat create a done signal (or sync.WaitGroup) that the
goroutine closes/signals on exit and change the returned func() (stopHeartbeat)
to close stop once (using stopOnce) and then wait on that done signal before
returning; keep the existing select cases (ticker.C, stop, s.Done()) and ensure
the done signal is triggered in every goroutine exit path so OnClose sees the
final s.Err() deterministically.

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes