Track streaming responses separately to fix SSE blocking flush

[arkanovicz]

HTTP/2 multiplexes multiple streams on a single connection. When an SSE endpoint is active, it keeps activeRequests > 0 forever, which prevents flushIfNeeded() from flushing responses for other requests on the same connection.

This fix:

• Tracks streaming responses (SSE) separately with a dedicated counter
• Changes flush condition to allow flushing when only streaming requests remain
• Properly decrements the streaming counter when SSE connections end

[coderabbitai]

Walkthrough

Adds per-call isStreamingResponse flag and a global atomic streamingResponses counter; response pipeline now detects text/event-stream, updates the counter on start/finish of streaming responses, and modifies flush/completion logic to account for active streaming responses.

Changes

Cohort / File(s)	Summary
State additions ktor-server/ktor-server-netty/jvm/src/io/ktor/server/netty/NettyApplicationCall.kt, ktor-server/ktor-server-netty/jvm/src/io/ktor/server/netty/NettyHttpHandlerState.kt	Added internal var isStreamingResponse: Boolean = false to NettyApplicationCall and internal val streamingResponses: AtomicLong = atomic(0L) to NettyHttpHandlerState.
Response pipeline updates ktor-server/ktor-server-netty/jvm/src/io/ktor/server/netty/cio/NettyHttpResponsePipeline.kt	flushIfNeeded() now compares activeRequests against streamingResponses; pipeline detects text/event-stream, marks calls as streaming, increments/decrements streamingResponses around streaming responses, and decrements on last-response handling.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~20 minutes

Suggested reviewers

• osipxd
• marychatte

🚥 Pre-merge checks | ✅ 3

✅ Passed checks (3 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title accurately describes the main change: introducing separate tracking for streaming responses to resolve SSE blocking flush issues.
Description check	✅ Passed	The description covers the problem (SSE blocking flush in HTTP/2), solution (separate counter for streaming responses), and key implementation details, aligning well with the template structure.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing touches

• 📝 Generate docstrings

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 1

🤖 Fix all issues with AI agents

In
`@ktor-server/ktor-server-netty/jvm/src/io/ktor/server/netty/cio/NettyHttpResponsePipeline.kt`:
- Around line 183-187: The streaming detection currently uses a case-sensitive
contains check and increments httpHandlerState.streamingResponses without
guaranteeing a decrement on failures; update the Content-Type check to be
case-insensitive (e.g., compare a lower-cased header or use a case-insensitive
match) when setting call.isStreamingResponse, and ensure the streamingResponses
counter is decremented on all failure/exception paths—for example, decrement in
the respondWithFailure path or wrap the response handling around a try/finally
that decrements httpHandlerState.streamingResponses and clears
call.isStreamingResponse when an exception occurs; refer to
call.isStreamingResponse, httpHandlerState.streamingResponses.incrementAndGet(),
and respondWithFailure to locate where to change the logic.

🧹 Nitpick comments (1)

ktor-server/ktor-server-netty/jvm/src/io/ktor/server/netty/NettyApplicationCall.kt (1)

36-37: Consider marking the new internal flag with @InternalAPI.

This is an internal API surface on a public class; the guidelines call for @InternalAPI on internal APIs to make the intent explicit. As per coding guidelines, ...

💡 Proposed change

-    internal var isStreamingResponse = false
+    `@InternalAPI`
+    internal var isStreamingResponse = false

[coderabbitai]

Actionable comments posted: 1

🤖 Fix all issues with AI agents

In
`@ktor-server/ktor-server-netty/jvm/src/io/ktor/server/netty/cio/NettyHttpResponsePipeline.kt`:
- Around line 183-192: The when-expression that computes contentType from
responseMessage (checking HttpResponse and Http2HeadersFrame via
HttpHeaders.ContentType) is left unclosed and the resulting contentType is
unused, breaking compilation and missing header sources; close the when block,
assign its result to contentType, then use contentType (falling back to
response.headers[HttpHeaders.ContentType] if needed) to detect
"text/event-stream" and, when matched, set call.isStreamingResponse = true and
call httpHandlerState.streamingResponses.incrementAndGet(); ensure you reference
responseMessage, HttpResponse, Http2HeadersFrame, HttpHeaders.ContentType,
call.isStreamingResponse, and
httpHandlerState.streamingResponses.incrementAndGet() in the corrected logic.

♻️ Duplicate comments (1)

ktor-server/ktor-server-netty/jvm/src/io/ktor/server/netty/cio/NettyHttpResponsePipeline.kt (1)

183-192: Ensure streaming counter is decremented on failure paths.
If an exception is thrown after incrementing, streamingResponses never decrements, which can reintroduce flush blocking.

🔧 Suggested change (failure path)

     private fun respondWithFailure(call: NettyApplicationCall, actualException: Throwable) {
         val t = when {
             actualException is IOException && actualException !is ChannelIOException ->
                 ChannelWriteException(exception = actualException)
             else -> actualException
         }
+
+        if (call.isStreamingResponse) {
+            httpHandlerState.streamingResponses.decrementAndGet()
+            call.isStreamingResponse = false
+        }