Fix CodeUnits write ambiguity + 500 fallback after deferred startwrite (#1302, #1303)

src/http_server.jl

@@ -177,6 +177,11 @@ mutable struct Stream{ISCLIENT,Req<:Request} <: IO
     @atomic write_closed::Bool
     @atomic read_closed::Bool
     @atomic response_started::Bool
+    # `response_started` flips at `startwrite` even when the response head is
+    # deferred (h1 FIXED mode); `head_committed` flips only when head bytes have
+    # actually been written to the transport, so error paths can tell whether a
+    # raw error response is still possible (#1303).
+    @atomic head_committed::Bool
     @atomic continue_sent::Bool
     ignore_writes::Bool
     write_mode::_ServerStreamWriteMode.T
@@ -242,6 +247,7 @@ function Stream(server::Server, tracked::_ServerConn, request::Req) where {Req<:
         false,
         false,
         false,
+        false,
         _ServerStreamWriteMode.UNDECIDED,
         Int64(0),
     )
@@ -290,6 +296,7 @@ function Stream(request::Req) where {Req<:Request}
         false,
         false,
         false,
+        false,
         _ServerStreamWriteMode.UNDECIDED,
         Int64(0),
     )
@@ -346,6 +353,7 @@ function Stream(
         false,
         false,
         false,
+        false,
         _ServerStreamWriteMode.UNDECIDED,
         Int64(0),
     )
@@ -1107,6 +1115,13 @@ function _serve_h1_conn!(server::Server, tracked::_ServerConn, reader_source)::N
                             startwrite(stream)
                             closewrite(stream)
                         end
+                    elseif !(@atomic :acquire stream.head_committed)
+                        # startwrite ran but the response head was deferred (h1
+                        # FIXED mode) and never reached the wire: answer with a
+                        # raw error response instead of silently dropping the
+                        # connection (#1303).
+                        _try_write_server_error!(tracked.conn, request, status === nothing ? 500 : status::Int)
+                        return nothing
                     end
                     @try_ignore begin
                         stream.response.close = true

src/http_server_streams.jl

@@ -86,6 +86,7 @@ function _write_server_stream_head!(stream::Stream)::Nothing
             end_stream,
             _server_write_deadline_ns(stream.server::Server),
         )
+        @atomic :release stream.head_committed = true
         @atomic :release stream.response_started = true
         return nothing
     end
@@ -109,6 +110,7 @@ function _write_server_stream_head!(stream::Stream)::Nothing
     _write_headers!(io, headers)
     write(io, "\r\n")
     _write_server_stream_bytes!(stream, take!(io), false)
+    @atomic :release stream.head_committed = true
     @atomic :release stream.response_started = true
     return nothing
 end

src/http_stream.jl

@@ -89,6 +89,7 @@ function Stream(
         false,
         false,
         false,
+        false,
         _ServerStreamWriteMode.UNDECIDED,
         Int64(0),
     )
@@ -246,6 +247,16 @@ function write(stream::Stream{true}, data::StridedVector{UInt8})::Int
 end
 write(stream::Stream{false}, data::StridedVector{UInt8}) = _server_write(stream, data)
 
+# Base defines write(::IO, ::Base.CodeUnits); without these methods, writing
+# codeunits to a Stream is ambiguous (CodeUnits <: DenseVector{UInt8}
+# intersects the StridedVector methods above) (#1302).
+function write(stream::Stream{true}, data::Base.CodeUnits{UInt8})::Int
+    (@atomic :acquire stream.started) && throw(ArgumentError("cannot write request body after response reading has started"))
+    (@atomic :acquire stream.write_closed) && throw(ArgumentError("request body writes are closed"))
+    return write(stream.request_buffer, data)
+end
+write(stream::Stream{false}, data::Base.CodeUnits{UInt8}) = _server_write(stream, data)
+
 function write(stream::Stream{true}, data::AbstractVector{UInt8})::Int
     (@atomic :acquire stream.started) && throw(ArgumentError("cannot write request body after response reading has started"))
     (@atomic :acquire stream.write_closed) && throw(ArgumentError("request body writes are closed"))

src/http_transport.jl

@@ -78,6 +78,8 @@ end
 Base.write(io::_RequestDeadlineWriteIO, data::Vector{UInt8})::Int = _request_deadline_write_bytes!(io, data)
 Base.write(io::_RequestDeadlineWriteIO, data::StridedVector{UInt8})::Int = _request_deadline_write_bytes!(io, data)
 Base.write(io::_RequestDeadlineWriteIO, data::AbstractVector{UInt8})::Int = _request_deadline_write_bytes!(io, data)
+# disambiguate vs Base's write(::IO, ::Base.CodeUnits) (#1302)
+Base.write(io::_RequestDeadlineWriteIO, data::Base.CodeUnits{UInt8})::Int = _request_deadline_write_bytes!(io, data)
 
 # Hook unsafe_write instead of write(::IO, ::Union{String, SubString{String}}):
 # Base's generic string write funnels through unsafe_write, so strings still

test/http_client_tests.jl

@@ -2385,3 +2385,25 @@ end
         HT.forceclose(server)
     end
 end
+
+@testset "HTTP client codeunits bodies dispatch without Base ambiguity (#1302)" begin
+    server = HTTP.serve!("127.0.0.1", 0) do req
+        body = req.body === nothing ? UInt8[] : req.body
+        return HT.Response(200, String(copy(body)))
+    end
+    try
+        url = "http://127.0.0.1:$(HTTP.port(server))/"
+        # codeunits request body with a write deadline: routes through the
+        # deadline write IO, which was ambiguous with Base's CodeUnits write
+        resp = HT.post(url; body = codeunits("cu-body"), write_idle_timeout = 5)
+        @test resp.status == 200
+        @test String(resp.body) == "cu-body"
+        # client stream write of codeunits (Stream{true} disambiguation)
+        resp_open = HT.open("POST", url) do io
+            write(io, codeunits("cu-open"))
+        end
+        @test resp_open.status == 200
+    finally
+        close(server)
+    end
+end

test/http_server_http1_tests.jl

@@ -52,32 +52,59 @@ end
 # indefinitely. Wrap the whole exchange in `_run_with_timeout`, a task-level
 # watchdog that does not depend on socket deadlines, so a stuck read fails the
 # test in seconds instead of hanging the job until CI's wall-clock limit.
+# Raised when the raw probe saw no response bytes at all within its budget.
+# On Windows this is the known IOCP wake strand (JuliaServices/Reseau.jl#107):
+# the parked reader is intermittently never woken even though bytes arrive.
+struct _FirstByteTimeout <: Exception end
+
+# `_read_until_quiet` runs under its own `_run_with_timeout` watchdog, so the
+# timeout surfaces here wrapped in (possibly nested) TaskFailedException.
+function _is_first_byte_timeout(err)::Bool
+    err isa _FirstByteTimeout && return true
+    err isa TaskFailedException || return false
+    inner = err.task.exception
+    return inner !== nothing && _is_first_byte_timeout(inner)
+end
+
 function _raw_http_request(
     port::Integer,
     request::AbstractString;
     settle_s::Float64 = 0.5,
     close_write::Bool = true,
     wait_for_first_byte::Bool = true,
+    # Re-dial fresh and retry when no first byte arrives. Windows-only by
+    # default so other platforms stay strict; safe for these probes because no
+    # caller asserts on server-side invocation counts.
+    attempts::Int = Sys.iswindows() ? 3 : 1,
 )::String
-    return _run_with_timeout(; timeout_s = max(8.0, settle_s + 4.0), label = "raw http request (port $(port))") do
-        sock = ND.connect("tcp", "127.0.0.1:$(Int(port))")
-        try
-            write(sock, Vector{UInt8}(codeunits(String(request))))
-            if close_write
-                HT.@try_ignore begin
-                    NC.closewrite(sock)
+    return _run_with_timeout(; timeout_s = max(8.0, settle_s + 4.0) * attempts, label = "raw http request (port $(port))") do
+        for attempt in 1:attempts
+            sock = ND.connect("tcp", "127.0.0.1:$(Int(port))")
+            try
+                write(sock, Vector{UInt8}(codeunits(String(request))))
+                if close_write
+                    HT.@try_ignore begin
+                        NC.closewrite(sock)
+                    end
                 end
+                # master hardening kept: longer first-byte budget on Windows,
+                # combined with the re-dial retry below
+                first_byte_timeout_s = max(Sys.iswindows() ? 5.0 : 2.0, settle_s + 1.0)
+                return _read_until_quiet(
+                    sock;
+                    timeout_s = first_byte_timeout_s,
+                    quiet_timeout_s = min(0.25, max(0.05, settle_s)),
+                    wait_for_first_byte = wait_for_first_byte,
+                )
+            catch err
+                _is_first_byte_timeout(err) || rethrow(err)
+                attempt < attempts || error("timed out waiting for first response byte ($(attempts) attempts)")
+                @warn "raw probe got no first byte; re-dialing — known Windows IOCP wake strand (Reseau#107)" attempt port
+            finally
+                NC.close(sock)
             end
-            first_byte_timeout_s = max(Sys.iswindows() ? 5.0 : 2.0, settle_s + 1.0)
-            return _read_until_quiet(
-                sock;
-                timeout_s = first_byte_timeout_s,
-                quiet_timeout_s = min(0.25, max(0.05, settle_s)),
-                wait_for_first_byte = wait_for_first_byte,
-            )
-        finally
-            NC.close(sock)
         end
+        error("unreachable: raw probe retry loop exited")
     end
 end
 
@@ -166,7 +193,7 @@ function _read_until_quiet(
             end
         end
         if wait_for_first_byte && !saw_bytes
-            error("timed out waiting for first response byte")
+            throw(_FirstByteTimeout())
         end
         return String(out)
     end
@@ -949,6 +976,74 @@ end
     end
 end
 
+@testset "HTTP stream writes accept codeunits without Base ambiguity (#1302)" begin
+    # Base defines write(::IO, ::Base.CodeUnits); the Stream StridedVector
+    # methods used to be ambiguous with it.
+    server = HT.listen!("127.0.0.1", 0; listenany = true) do stream
+        _ = HT.startread(stream)
+        HT.setstatus(stream, 200)
+        HT.setheader(stream, "Content-Length" => "2")
+        HT.startwrite(stream)
+        write(stream, codeunits("ok"))
+        return nothing
+    end
+    address = HT.server_addr(server)
+    try
+        resp = HT.get("http://$(address)/")
+        @test resp.status == 200
+        @test String(resp.body) == "ok"
+    finally
+        _run_with_timeout(() -> HT.forceclose(server); label = "server forceclose")
+        _run_with_timeout(() -> wait(server); label = "server task completion")
+    end
+end
+
+@testset "HTTP stream handler error after deferred startwrite returns 500 (#1303)" begin
+    # h1 FIXED mode defers the response head at startwrite; a handler throw
+    # before anything reaches the wire must produce a raw 500, not a silent
+    # connection drop the client sees as "unexpected EOF".
+    server = HT.listen!("127.0.0.1", 0; listenany = true) do stream
+        _ = HT.startread(stream)
+        HT.setstatus(stream, 200)
+        HT.setheader(stream, "Content-Length" => "2")
+        HT.startwrite(stream)      # FIXED: response_started, head NOT committed
+        error("boom")
+    end
+    address = HT.server_addr(server)
+    try
+        resp = HT.request("GET", "http://$(address)/"; status_exception = false, retry = false)
+        @test resp.status == 500
+    finally
+        _run_with_timeout(() -> HT.forceclose(server); label = "server forceclose")
+        _run_with_timeout(() -> wait(server); label = "server task completion")
+    end
+
+    # control: once the head is committed (chunked writes it at startwrite), a
+    # handler throw must NOT emit a spurious 500 after the real head
+    server2 = HT.listen!("127.0.0.1", 0; listenany = true) do stream
+        _ = HT.startread(stream)
+        HT.setstatus(stream, 200)
+        HT.setheader(stream, "Transfer-Encoding" => "chunked")
+        HT.startwrite(stream)      # chunked: head committed to the wire here
+        error("boom")
+    end
+    address2 = HT.server_addr(server2)
+    try
+        outcome = try
+            resp2 = HT.request("GET", "http://$(address2)/"; status_exception = false, retry = false)
+            resp2.status
+        catch err
+            err
+        end
+        # truncated 200 or a client-side error are both acceptable; a 500 means
+        # the server wrote a second head after the committed one
+        @test outcome != 500
+    finally
+        _run_with_timeout(() -> HT.forceclose(server2); label = "server forceclose")
+        _run_with_timeout(() -> wait(server2); label = "server task completion")
+    end
+end
+
 @testset "HTTP stream byte I/O supports generic AbstractStrings and Chars" begin
     # Strings are written via unsafe_write; generic AbstractStrings and Chars
     # fall back to Base's per-byte path, which requires write(io, ::UInt8).