Make test synchronization deterministic

src/http2_client.jl

@@ -310,16 +310,22 @@ end
     end
 end
 
-function _wait_h2_send_window_locked!(conn::H2Connection, stream_id::UInt32, deadline_ns::Int64)::Nothing
+function _wait_h2_send_window_locked!(
+    conn::H2Connection,
+    stream_id::UInt32,
+    deadline_ns::Int64;
+    clock_ns::Function=time_ns,
+    wait_for::Function=IOPoll.timedwait,
+)::Nothing
     if deadline_ns == 0
         wait(conn.window_condition)
         return nothing
     end
-    remaining_ns = deadline_ns - Int64(time_ns())
+    remaining_ns = deadline_ns - Int64(clock_ns())
     remaining_ns <= 0 && throw(IOPoll.DeadlineExceededError())
     unlock(conn.state_lock)
     try
-        status = IOPoll.timedwait(() -> begin
+        status = wait_for(() -> begin
             lock(conn.state_lock)
             try
                 return _h2_send_window_ready_locked(conn, stream_id)

src/http2_server.jl

@@ -310,16 +310,21 @@ function _apply_h2_window_update!(send_state::_H2SendWindowState, frame::WindowU
     return nothing
 end
 
-function _wait_h2_send_window_locked!(send_state::_H2SendWindowState, deadline_ns::Int64)::Nothing
+function _wait_h2_send_window_locked!(
+    send_state::_H2SendWindowState,
+    deadline_ns::Int64;
+    clock_ns::Function=time_ns,
+    wait_ns::Function=IOPoll.sleep_ns,
+)::Nothing
     if deadline_ns == 0
         wait(send_state.window_condition)
         return nothing
     end
-    remaining_ns = deadline_ns - Int64(time_ns())
+    remaining_ns = deadline_ns - Int64(clock_ns())
     remaining_ns <= 0 && throw(IOPoll.DeadlineExceededError())
     unlock(send_state.state_lock)
     try
-        IOPoll.sleep_ns(min(remaining_ns, Int64(1_000_000)))
+        wait_ns(min(remaining_ns, Int64(1_000_000)))
     finally
         lock(send_state.state_lock)
     end

src/http_client_timeouts.jl

@@ -77,10 +77,11 @@ function _apply_request_timeout_settings!(
     ctx::RequestContext,
     request_timeout_ns::Int64,
     config::Union{Nothing,_RequestTimeoutConfig},
+    ;
+    now_ns::Int64=Int64(time_ns()),
 )::RequestContext
     request_timeout_ns < 0 && throw(ArgumentError("request_timeout_ns must be >= 0"))
     if request_timeout_ns > 0
-        now_ns = Int64(time_ns())
         deadline_ns = now_ns > typemax(Int64) - request_timeout_ns ? typemax(Int64) : now_ns + request_timeout_ns
         set_deadline!(ctx, deadline_ns)
     end

src/http_retry.jl

@@ -174,7 +174,10 @@ function _retry_after_delay_ns(headers::Headers)::Union{Nothing,Int64}
     return _parse_retry_after_delay_ns(value::String)
 end
 
-function _parse_retry_after_delay_ns(value::AbstractString)::Union{Nothing,Int64}
+function _parse_retry_after_delay_ns(
+    value::AbstractString;
+    now::Dates.DateTime=Dates.now(Dates.UTC),
+)::Union{Nothing,Int64}
     stripped = strip(String(value))
     isempty(stripped) && return nothing
     parsed_secs = try
@@ -190,7 +193,7 @@ function _parse_retry_after_delay_ns(value::AbstractString)::Union{Nothing,Int64
     end
     parsed_dt = Cookies._parse_http_gmt_datetime(stripped)
     parsed_dt === nothing && return nothing
-    delta = parsed_dt::Dates.DateTime - Dates.now(Dates.UTC)
+    delta = parsed_dt::Dates.DateTime - now
     millis = Dates.value(delta)
     millis <= 0 && return Int64(0)
     millis > typemax(Int64) ÷ 1_000_000 && return typemax(Int64)

src/http_transport.jl

@@ -217,6 +217,7 @@ mutable struct Transport
     max_conns_per_host::Int
     idle_timeout_ns::Int64
     lock::ReentrantLock
+    waiter_condition::Threads.Condition
     idle::Dict{String,Vector{Conn}}
     waiters::Dict{String,Vector{_ConnWaiter}}
     conns_per_host::Dict{String,Int}
@@ -280,6 +281,7 @@ function Transport(;
     max_conns_per_host >= 0 || throw(ArgumentError("max_conns_per_host must be >= 0"))
     idle_timeout_ns >= 0 || throw(ArgumentError("idle_timeout_ns must be >= 0"))
     host_resolver = HostResolvers.HostResolver(local_addr=_normalize_local_addr(local_addr))
+    lock = ReentrantLock()
     return Transport(
         host_resolver,
         tls_config,
@@ -289,7 +291,8 @@ function Transport(;
         Int(max_idle_total),
         Int(max_conns_per_host),
         Int64(idle_timeout_ns),
-        ReentrantLock(),
+        lock,
+        Threads.Condition(lock),
         Dict{String,Vector{Conn}}(),
         Dict{String,Vector{_ConnWaiter}}(),
         Dict{String,Int}(),
@@ -541,6 +544,7 @@ function _enqueue_waiter_locked!(transport::Transport, waiter::_ConnWaiter)
     queue = get(() -> _ConnWaiter[], transport.waiters, waiter.key)
     push!(queue, waiter)
     transport.waiters[waiter.key] = queue
+    notify(transport.waiter_condition; all=true)
     return waiter
 end
 
@@ -643,7 +647,13 @@ function _deliver_waiter_error_locked!(waiter::_ConnWaiter, err::Exception)::Boo
     return true
 end
 
-function _wait_for_conn!(transport::Transport, waiter::_ConnWaiter, deadline_ns::Int64)
+function _wait_for_conn!(
+    transport::Transport,
+    waiter::_ConnWaiter,
+    deadline_ns::Int64;
+    clock_ns::Function=time_ns,
+    wait_for::Function=IOPoll.timedwait,
+)
     while true
         state = @atomic :acquire waiter.state
         if state == _CONN_WAITER_CONN
@@ -659,7 +669,7 @@ function _wait_for_conn!(transport::Transport, waiter::_ConnWaiter, deadline_ns:
             wait(waiter.signal)
             continue
         end
-        now_ns = Int64(time_ns())
+        now_ns = Int64(clock_ns())
         if now_ns >= deadline_ns
             lock(transport.lock)
             try
@@ -674,7 +684,7 @@ function _wait_for_conn!(transport::Transport, waiter::_ConnWaiter, deadline_ns:
             continue
         end
         timeout_s = min((deadline_ns - now_ns) / 1.0e9, 0.05)
-        IOPoll.timedwait(() -> (@atomic :acquire waiter.state) != _CONN_WAITER_WAITING, timeout_s; pollint=0.001)
+        wait_for(() -> (@atomic :acquire waiter.state) != _CONN_WAITER_WAITING, timeout_s; pollint=0.001)
     end
 end
 

test/README.md

@@ -0,0 +1,26 @@
+# Deterministic test synchronization
+
+HTTP.jl tests must not depend on scheduler speed or elapsed wall-clock time.
+GitHub Actions runners can pause a task for an unknown period. A delay that is
+safe on one runner can fail on another runner without a product defect.
+
+Use observable state transitions instead:
+
+- Use a `Channel`, `Base.Event`, or `Threads.Condition` for task handshakes.
+- Read exact byte counts, complete protocol frames, markers, or EOF.
+- Use `fetch(task)` or `wait(task)` for task completion. Wrap unexpected
+  `Threads.@spawn` failures with `errormonitor`.
+- Inject a fixed clock value into pure deadline calculations.
+- Use an already-expired absolute deadline when a test must enter a product
+  timeout branch. Do not wait for a future deadline to expire.
+- Mutate private lifecycle state only when the test directly covers that state,
+  such as an idle-pool eviction test.
+
+Do not use `sleep`, `timedwait`, `time`, `time_ns`, `Timer`, elapsed-time
+assertions, polling intervals, or helper-level timeout arguments in test code.
+Do not use a short delay to prove that an event has not occurred. Build a
+barrier that makes the event impossible until the test releases it.
+
+Product timeout configuration remains valid test input. It tests parsing,
+propagation, and expired-deadline behavior. It must not act as the test harness.
+The GitHub Actions job timeout remains the final guard for a true deadlock.

test/http1_wire_tests.jl

@@ -267,7 +267,7 @@ end
             # _ConnReader pulls it into one buffer fill -> later lines are served
             # from the buffered fast path.
             write(client, bytes)
-            Reseau.IOPoll.timedwait(() -> server_conn[] !== nothing, 5.0; pollint = 0.001)
+            fetch(t)
             return HT._ConnReader(server_conn[]::Reseau.TCP.Conn), client, listener
         catch
             HT.@try_ignore close(listener)
@@ -326,7 +326,7 @@ end
             t = Task(() -> (server_conn[] = Reseau.TCP.accept(listener)))
             schedule(t)
             client = Reseau.TCP.connect(Reseau.TCP.loopback_addr(Int(addr.port)))
-            Reseau.IOPoll.timedwait(() -> server_conn[] !== nothing, 5.0; pollint = 0.001)
+            fetch(t)
             for c in chunks
                 write(client, c)
             end