From b98d88e728505b9d11f9cb7e1d64a66e4f147ab9 Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Mon, 29 Jun 2026 10:59:05 -0600 Subject: [PATCH] Harden flaky transport tests --- test/http_client_transport_tests.jl | 30 +++++++++++----- test/http_server_http1_tests.jl | 53 +++++++++++++++++++++-------- 2 files changed, 61 insertions(+), 22 deletions(-) diff --git a/test/http_client_transport_tests.jl b/test/http_client_transport_tests.jl index d7ae332c3..a41f782ca 100644 --- a/test/http_client_transport_tests.jl +++ b/test/http_client_transport_tests.jl @@ -802,6 +802,8 @@ end address = ND.join_host_port("127.0.0.1", Int(laddr.port)) close_count = Ref(0) stage = Ref(1) + second_chunk_started = Channel{Nothing}(1) + release_second_chunk = Channel{Nothing}(1) callback_body = HT.CallbackBody( dst -> begin if stage[] == 1 @@ -811,7 +813,8 @@ end return length(bytes) end if stage[] == 2 - sleep(1.0) + isready(second_chunk_started) || put!(second_chunk_started, nothing) + take!(release_second_chunk) bytes = collect(codeunits("world")) copyto!(dst, 1, bytes, 1, length(bytes)) stage[] = 3 @@ -837,15 +840,26 @@ end transport = HT.Transport(max_idle_per_host = 4, max_idle_total = 4) try req = HT.Request("POST", "/early"; host = address, body = callback_body, content_length = 10) - started = time() - res = HT.roundtrip!(transport, address, req) - elapsed = time() - started - @test res.status == 200 - @test String(_read_all_transport_body_bytes(res.body)) == "early" - @test elapsed < 0.75 - @test timedwait(() -> close_count[] == 1, 2.0; pollint = 0.001) != :timed_out + res_task = errormonitor(Threads.@spawn HT.roundtrip!(transport, address, req)) + ready = timedwait(() -> istaskdone(res_task) || isready(second_chunk_started), 2.0; pollint = 0.001) + @test ready == :ok + if ready == :ok && isready(second_chunk_started) && !istaskdone(res_task) + response_ready = timedwait(() -> istaskdone(res_task), 2.0; pollint = 0.001) + @test response_ready == :ok + end + @test istaskdone(res_task) + isready(release_second_chunk) || put!(release_second_chunk, nothing) + if istaskdone(res_task) + res = fetch(res_task) + @test res.status == 200 + @test String(_read_all_transport_body_bytes(res.body)) == "early" + @test timedwait(() -> close_count[] == 1, 2.0; pollint = 0.001) != :timed_out + else + _wait_task!(res_task) + end _wait_task!(server_task) finally + isready(release_second_chunk) || put!(release_second_chunk, nothing) close(transport) HTTP.@try_ignore NC.close(listener) end diff --git a/test/http_server_http1_tests.jl b/test/http_server_http1_tests.jl index e0eafbb7e..763427aa0 100644 --- a/test/http_server_http1_tests.jl +++ b/test/http_server_http1_tests.jl @@ -99,7 +99,7 @@ function _raw_http_request( 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 + @warn "raw probe got no first byte; re-dialing after known Windows IOCP wake strand (Reseau#107)" attempt port finally NC.close(sock) end @@ -108,15 +108,28 @@ function _raw_http_request( end end -function _raw_http_request_until_close(port::Integer, request::AbstractString; timeout_s::Float64 = 3.0)::Tuple{String, Bool} - return _run_with_timeout(; timeout_s = timeout_s + 4.0, label = "raw http request until close (port $(port))") do - sock = ND.connect("tcp", "127.0.0.1:$(Int(port))") - try - write(sock, Vector{UInt8}(codeunits(String(request)))) - return _read_until_close(sock; timeout_s) - finally - NC.close(sock) +function _raw_http_request_until_close( + port::Integer, + request::AbstractString; + timeout_s::Float64 = 3.0, + wait_for_first_byte::Bool = true, + attempts::Int = Sys.iswindows() ? 3 : 1, +)::Tuple{String, Bool} + return _run_with_timeout(; timeout_s = (timeout_s + 4.0) * attempts, label = "raw http request until close (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)))) + return _read_until_close(sock; timeout_s, wait_for_first_byte) + catch err + _is_first_byte_timeout(err) || rethrow(err) + attempt < attempts || error("timed out waiting for first response byte before close ($(attempts) attempts)") + @warn "raw probe got no first byte before close; re-dialing after known Windows IOCP wake strand (Reseau#107)" attempt port + finally + NC.close(sock) + end end + error("unreachable: raw probe retry loop exited") end end @@ -199,23 +212,34 @@ function _read_until_quiet( end end -function _read_until_close(conn::NC.Conn; timeout_s::Float64 = 1.0)::Tuple{String, Bool} +function _read_until_close(conn::NC.Conn; timeout_s::Float64 = 1.0, wait_for_first_byte::Bool = false)::Tuple{String, Bool} buf = Vector{UInt8}(undef, 1024) out = UInt8[] deadline_ns = Int64(time_ns()) + round(Int64, timeout_s * 1.0e9) + saw_bytes = false while true remaining_ns = deadline_ns - Int64(time_ns()) - remaining_ns <= 0 && return String(out), false + if remaining_ns <= 0 + wait_for_first_byte && !saw_bytes && throw(_FirstByteTimeout()) + return String(out), false + end NC.set_read_deadline!(conn, Int64(time_ns()) + remaining_ns) try chunk = readavailable(conn) n = length(chunk) - n == 0 && return String(out), true + if n == 0 + wait_for_first_byte && !saw_bytes && throw(_FirstByteTimeout()) + return String(out), true + end n > length(buf) && resize!(buf, n) copyto!(buf, 1, chunk, 1, n) append!(out, @view(buf[1:n])) + saw_bytes = true catch err - err isa IOP.DeadlineExceededError && return String(out), false + if err isa IOP.DeadlineExceededError + wait_for_first_byte && !saw_bytes && throw(_FirstByteTimeout()) + return String(out), false + end (err isa EOFError || HT._is_peer_close_error(err::Exception)) && return String(out), true rethrow(err) end @@ -1037,7 +1061,8 @@ end sock = ND.connect("tcp", "127.0.0.1:$(HT.port(server))") try write(sock, Vector{UInt8}(codeunits("GET /one HTTP/1.1\r\nHost: $(address)\r\n\r\n"))) - first = _read_until_quiet(sock; timeout_s = 2.0, quiet_timeout_s = 0.1) + first_timeout_s = Sys.iswindows() ? 5.0 : 2.0 + first = _read_until_quiet(sock; timeout_s = first_timeout_s, quiet_timeout_s = 0.1, wait_for_first_byte = true) @test occursin("HTTP/1.1 200 OK", first) sleep(1.0) closed_after_idle = false