|
52 | 52 | # indefinitely. Wrap the whole exchange in `_run_with_timeout`, a task-level |
53 | 53 | # watchdog that does not depend on socket deadlines, so a stuck read fails the |
54 | 54 | # test in seconds instead of hanging the job until CI's wall-clock limit. |
| 55 | +# Raised when the raw probe saw no response bytes at all within its budget. |
| 56 | +# On Windows this is the known IOCP wake strand (JuliaServices/Reseau.jl#107): |
| 57 | +# the parked reader is intermittently never woken even though bytes arrive. |
| 58 | +struct _FirstByteTimeout <: Exception end |
| 59 | + |
| 60 | +# `_read_until_quiet` runs under its own `_run_with_timeout` watchdog, so the |
| 61 | +# timeout surfaces here wrapped in (possibly nested) TaskFailedException. |
| 62 | +function _is_first_byte_timeout(err)::Bool |
| 63 | + err isa _FirstByteTimeout && return true |
| 64 | + err isa TaskFailedException || return false |
| 65 | + inner = err.task.exception |
| 66 | + return inner !== nothing && _is_first_byte_timeout(inner) |
| 67 | +end |
| 68 | + |
55 | 69 | function _raw_http_request( |
56 | 70 | port::Integer, |
57 | 71 | request::AbstractString; |
58 | 72 | settle_s::Float64 = 0.5, |
59 | 73 | close_write::Bool = true, |
60 | 74 | wait_for_first_byte::Bool = true, |
| 75 | + # Re-dial fresh and retry when no first byte arrives. Windows-only by |
| 76 | + # default so other platforms stay strict; safe for these probes because no |
| 77 | + # caller asserts on server-side invocation counts. |
| 78 | + attempts::Int = Sys.iswindows() ? 3 : 1, |
61 | 79 | )::String |
62 | | - return _run_with_timeout(; timeout_s = max(8.0, settle_s + 4.0), label = "raw http request (port $(port))") do |
63 | | - sock = ND.connect("tcp", "127.0.0.1:$(Int(port))") |
64 | | - try |
65 | | - write(sock, Vector{UInt8}(codeunits(String(request)))) |
66 | | - if close_write |
67 | | - HT.@try_ignore begin |
68 | | - NC.closewrite(sock) |
| 80 | + return _run_with_timeout(; timeout_s = max(8.0, settle_s + 4.0) * attempts, label = "raw http request (port $(port))") do |
| 81 | + for attempt in 1:attempts |
| 82 | + sock = ND.connect("tcp", "127.0.0.1:$(Int(port))") |
| 83 | + try |
| 84 | + write(sock, Vector{UInt8}(codeunits(String(request)))) |
| 85 | + if close_write |
| 86 | + HT.@try_ignore begin |
| 87 | + NC.closewrite(sock) |
| 88 | + end |
69 | 89 | end |
| 90 | + # master hardening kept: longer first-byte budget on Windows, |
| 91 | + # combined with the re-dial retry below |
| 92 | + first_byte_timeout_s = max(Sys.iswindows() ? 5.0 : 2.0, settle_s + 1.0) |
| 93 | + return _read_until_quiet( |
| 94 | + sock; |
| 95 | + timeout_s = first_byte_timeout_s, |
| 96 | + quiet_timeout_s = min(0.25, max(0.05, settle_s)), |
| 97 | + wait_for_first_byte = wait_for_first_byte, |
| 98 | + ) |
| 99 | + catch err |
| 100 | + _is_first_byte_timeout(err) || rethrow(err) |
| 101 | + attempt < attempts || error("timed out waiting for first response byte ($(attempts) attempts)") |
| 102 | + @warn "raw probe got no first byte; re-dialing — known Windows IOCP wake strand (Reseau#107)" attempt port |
| 103 | + finally |
| 104 | + NC.close(sock) |
70 | 105 | end |
71 | | - first_byte_timeout_s = max(Sys.iswindows() ? 5.0 : 2.0, settle_s + 1.0) |
72 | | - return _read_until_quiet( |
73 | | - sock; |
74 | | - timeout_s = first_byte_timeout_s, |
75 | | - quiet_timeout_s = min(0.25, max(0.05, settle_s)), |
76 | | - wait_for_first_byte = wait_for_first_byte, |
77 | | - ) |
78 | | - finally |
79 | | - NC.close(sock) |
80 | 106 | end |
| 107 | + error("unreachable: raw probe retry loop exited") |
81 | 108 | end |
82 | 109 | end |
83 | 110 |
|
@@ -166,7 +193,7 @@ function _read_until_quiet( |
166 | 193 | end |
167 | 194 | end |
168 | 195 | if wait_for_first_byte && !saw_bytes |
169 | | - error("timed out waiting for first response byte") |
| 196 | + throw(_FirstByteTimeout()) |
170 | 197 | end |
171 | 198 | return String(out) |
172 | 199 | end |
|
949 | 976 | end |
950 | 977 | end |
951 | 978 |
|
| 979 | +@testset "HTTP stream writes accept codeunits without Base ambiguity (#1302)" begin |
| 980 | + # Base defines write(::IO, ::Base.CodeUnits); the Stream StridedVector |
| 981 | + # methods used to be ambiguous with it. |
| 982 | + server = HT.listen!("127.0.0.1", 0; listenany = true) do stream |
| 983 | + _ = HT.startread(stream) |
| 984 | + HT.setstatus(stream, 200) |
| 985 | + HT.setheader(stream, "Content-Length" => "2") |
| 986 | + HT.startwrite(stream) |
| 987 | + write(stream, codeunits("ok")) |
| 988 | + return nothing |
| 989 | + end |
| 990 | + address = HT.server_addr(server) |
| 991 | + try |
| 992 | + resp = HT.get("http://$(address)/") |
| 993 | + @test resp.status == 200 |
| 994 | + @test String(resp.body) == "ok" |
| 995 | + finally |
| 996 | + _run_with_timeout(() -> HT.forceclose(server); label = "server forceclose") |
| 997 | + _run_with_timeout(() -> wait(server); label = "server task completion") |
| 998 | + end |
| 999 | +end |
| 1000 | + |
| 1001 | +@testset "HTTP stream handler error after deferred startwrite returns 500 (#1303)" begin |
| 1002 | + # h1 FIXED mode defers the response head at startwrite; a handler throw |
| 1003 | + # before anything reaches the wire must produce a raw 500, not a silent |
| 1004 | + # connection drop the client sees as "unexpected EOF". |
| 1005 | + server = HT.listen!("127.0.0.1", 0; listenany = true) do stream |
| 1006 | + _ = HT.startread(stream) |
| 1007 | + HT.setstatus(stream, 200) |
| 1008 | + HT.setheader(stream, "Content-Length" => "2") |
| 1009 | + HT.startwrite(stream) # FIXED: response_started, head NOT committed |
| 1010 | + error("boom") |
| 1011 | + end |
| 1012 | + address = HT.server_addr(server) |
| 1013 | + try |
| 1014 | + resp = HT.request("GET", "http://$(address)/"; status_exception = false, retry = false) |
| 1015 | + @test resp.status == 500 |
| 1016 | + finally |
| 1017 | + _run_with_timeout(() -> HT.forceclose(server); label = "server forceclose") |
| 1018 | + _run_with_timeout(() -> wait(server); label = "server task completion") |
| 1019 | + end |
| 1020 | + |
| 1021 | + # control: once the head is committed (chunked writes it at startwrite), a |
| 1022 | + # handler throw must NOT emit a spurious 500 after the real head |
| 1023 | + server2 = HT.listen!("127.0.0.1", 0; listenany = true) do stream |
| 1024 | + _ = HT.startread(stream) |
| 1025 | + HT.setstatus(stream, 200) |
| 1026 | + HT.setheader(stream, "Transfer-Encoding" => "chunked") |
| 1027 | + HT.startwrite(stream) # chunked: head committed to the wire here |
| 1028 | + error("boom") |
| 1029 | + end |
| 1030 | + address2 = HT.server_addr(server2) |
| 1031 | + try |
| 1032 | + outcome = try |
| 1033 | + resp2 = HT.request("GET", "http://$(address2)/"; status_exception = false, retry = false) |
| 1034 | + resp2.status |
| 1035 | + catch err |
| 1036 | + err |
| 1037 | + end |
| 1038 | + # truncated 200 or a client-side error are both acceptable; a 500 means |
| 1039 | + # the server wrote a second head after the committed one |
| 1040 | + @test outcome != 500 |
| 1041 | + finally |
| 1042 | + _run_with_timeout(() -> HT.forceclose(server2); label = "server forceclose") |
| 1043 | + _run_with_timeout(() -> wait(server2); label = "server task completion") |
| 1044 | + end |
| 1045 | +end |
| 1046 | + |
952 | 1047 | @testset "HTTP stream byte I/O supports generic AbstractStrings and Chars" begin |
953 | 1048 | # Strings are written via unsafe_write; generic AbstractStrings and Chars |
954 | 1049 | # fall back to Base's per-byte path, which requires write(io, ::UInt8). |
|
0 commit comments