fix(http2): stream abstract bodies incrementally

Write each body chunk as soon as body_read! returns it. Finish body-only responses with an empty END_STREAM DATA frame so trailers keep their existing terminal HEADERS behavior.

This prevents long-lived SSE responses, including MCP subscription streams, from holding the first event until the next event or EOF. Add a real HTTP/2 regression test and keep the HTTP/1 test synchronized without atomics.

Author: quinnj

src/http2_server.jl

@@ -1158,26 +1158,22 @@ function _write_response_body_h2_server!(
     end
     body isa AbstractBody || throw(ProtocolError("unsupported HTTP/2 response body type $(typeof(body))"))
     buf = Vector{UInt8}(undef, 16 * 1024)
-    pending = UInt8[]
-    have_pending = false
     try
         while true
             n = body_read!(body::AbstractBody, buf)
             if n == 0
-                if have_pending
-                    _write_data_frames_h2_server!(conn, write_lock, send_state, stream_id, pending; end_stream=end_stream, write_deadline_ns=write_deadline_ns)
-                elseif end_stream
-                    _write_frame_h2_server_threadsafe!(write_lock, conn, DataFrame(stream_id, true, UInt8[]), write_deadline_ns)
-                end
+                end_stream && _write_frame_h2_server_threadsafe!(write_lock, conn, DataFrame(stream_id, true, UInt8[]), write_deadline_ns)
                 return nothing
             end
-            current = Vector{UInt8}(undef, n)
-            copyto!(current, 1, buf, 1, n)
-            if have_pending
-                _write_data_frames_h2_server!(conn, write_lock, send_state, stream_id, pending; end_stream=false, write_deadline_ns=write_deadline_ns)
-            end
-            pending = current
-            have_pending = true
+            _write_data_frames_h2_server!(
+                conn,
+                write_lock,
+                send_state,
+                stream_id,
+                @view(buf[1:n]);
+                end_stream=false,
+                write_deadline_ns=write_deadline_ns,
+            )
         end
     finally
         @try_ignore begin

test/http2_server_tests.jl

@@ -237,6 +237,38 @@ end
     end
 end
 
+@testset "HTTP/2 server SSE events stream incrementally" begin
+    # The producer only writes the second event after the client receives the
+    # first. Holding one AbstractBody chunk until the next read or EOF makes the
+    # producer emit the sentinel instead and fails without a latency assertion.
+    saw_first = Channel{Nothing}(1)
+    server = HT.serve!("127.0.0.1", 0; listenany = true) do request
+            _ = request
+            return HT.sse_stream(200) do stream
+                write(stream, HT.SSEEvent("one"))
+                delivered = timedwait(() -> isready(saw_first), 5.0; pollint = 0.01)
+                write(stream, HT.SSEEvent(delivered === :ok ? "two" : "first event was not delivered before close"))
+            end
+        end
+    address = HT.server_addr(server)
+    conn = HT.connect_h2!(address; secure = false)
+    try
+        request = HT.Request("GET", "/"; host = address, body = HT.EmptyBody(), content_length = 0, proto_major = 2, proto_minor = 0)
+        response = HT.h2_roundtrip!(conn, request)
+        first_buffer = Vector{UInt8}(undef, 64)
+        first_count = HT.body_read!(response.body, first_buffer)
+        first_event = String(first_buffer[1:first_count])
+        first_event == "data: one\n\n" && put!(saw_first, nothing)
+        remaining = String(_read_all_h2_server(response.body))
+        @test response.status == 200
+        @test first_event * remaining == "data: one\n\ndata: two\n\n"
+    finally
+        close(conn)
+        HT.forceclose(server)
+        _ = timedwait(() -> istaskdone(server.serve_task::Task), 3.0; pollint = 0.001)
+    end
+end
+
 @testset "HTTP/2 server writes unread BytesBody response bytes directly" begin
     payload = collect(codeunits("abcdef"))
     returned_body = Ref{Union{Nothing,HT.BytesBody}}(nothing)

test/http_server_http1_tests.jl

@@ -295,12 +295,12 @@ end
     # callback has observed the first; if the transport buffers until close,
     # the producer times out and writes a sentinel instead, failing the test
     # deterministically without wall-clock assertions.
-    saw_first = Threads.Atomic{Bool}(false)
+    saw_first = Channel{Nothing}(1)
     server = HT.serve!("127.0.0.1", 0; listenany = true) do request
             _ = request
             return HT.sse_stream(200) do stream
                 write(stream, HT.SSEEvent("one"))
-                delivered = timedwait(() -> saw_first[], 5.0; pollint = 0.01)
+                delivered = timedwait(() -> isready(saw_first), 5.0; pollint = 0.01)
                 write(stream, HT.SSEEvent(delivered === :ok ? "two" : "first event was not delivered before close"))
             end
         end
@@ -309,7 +309,7 @@ end
         events = String[]
         response = HT.get("http://$(address)/"; sse_callback = event -> begin
             push!(events, event.data)
-            event.data == "one" && (saw_first[] = true)
+            event.data == "one" && put!(saw_first, nothing)
         end)
         @test response.status == 200
         @test events == ["one", "two"]