Make test synchronization deterministic (#1346)

* test(ci): remove wall-clock synchronization

Replace sleeps, polling deadlines, and elapsed-time assertions with lifecycle signals, exact protocol reads, and expired deadlines.

Add a policy test and contributor guidance to keep the suite deterministic across variable CI runner load.

* test(ci): restore deterministic wait coverage

Add lifecycle barriers and private clock and wait seams so deterministic tests still execute active wait, wake, handoff, and shutdown paths.

Restore project coverage without reintroducing elapsed-time coordination.

Author: quinnj

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