diff --git a/CHANGELOG.md b/CHANGELOG.md index 51e4a6551..5fbce7d2a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,12 +10,51 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - Added `HTTP2Settings` to configure HTTP/2 receive flow-control windows (per-stream `initial_window_size` and connection-level `connection_window_size`). Pass it via the `http2_settings` keyword on `Client`, `Server`, `listen!`, `serve!`, `serve`, and `connect_h2!`. Defaults preserve the protocol-default 65535-byte windows, and the per-stream receive buffer cap is derived from the window. Raising the windows improves single-stream throughput on links with non-trivial latency. - Added `HTTP.peeraddr(::HTTP.Stream)`, returning the remote (client) `SocketAddr` of a server stream for both plain-TCP and TLS connections and both HTTP/1 and HTTP/2. This is the supported way to obtain the client IP (for rate limiting, audit logging, and per-client policy) without reaching into transport internals, and restores the capability `Sockets.getpeername(::HTTP.Stream)` provided in HTTP.jl 1.x. +- Added `HTTP.RetrySkippedEvent`, a request trace event emitted when the retry + policy wanted to retry an attempt but the retry was not armed — because the + transport's `RetryBucket` denied capacity (`reason = :retry_bucket`) or the + request deadline preempted the backoff (`reason = :deadline`). Previously a + denied retry was indistinguishable from a non-retryable failure. ([#1353]) +- Added `HTTP.TLSTransportError`, raised when TLS-level I/O fails on an + established connection during a request. Previously such failures were + mislabeled `TLSHandshakeError`; that type is now reserved for actual + connection-setup failures. `HTTP.isrecoverable` classifies both wrappers by + their underlying cause. ([#1353]) ### Fixed - Restored HTTP and WebSocket server task scheduling to Julia's `:interactive` thread pool so default-pool compute work cannot starve server and health-check tasks when an interactive thread is configured. ([#1342]) - Percent-decode `userinfo` before building the `Basic` auth header (RFC 3986); fixes wrong credentials for request URLs and proxies containing percent-encoded characters. +- Fixed the client retry budget (`RetryBucket`) treating a successful retried + attempt as a full-cost failure. The per-host budget drained by 10 of 500 + units on every retry — even one that recovered with a 2xx — and never + refilled, so after ~50 retries against a host every subsequent retry was + silently denied for the transport's lifetime and transient errors surfaced + raw despite `retry=true`. A retry reservation is now settled by the + effective retry decision for the response it produced: refunded in full when + the built-in policy (or a custom `retry_if`) no longer wants a retry, and + keeping the partial cost when the response is still classified as a failure. + On the final attempt the built-in classification applies without invoking + `retry_if`, and a retry that `retry_if` explicitly requested conservatively + keeps cost on a non-2xx/3xx outcome. Each successful non-retried request + restores one unit of previously consumed budget, retry reservations and + response connections are released even when a trace or retry-policy callback + throws, and the request deadline is rechecked after the backoff sleep. + ([#1353]) +- The HTTP/1 transport now retries a replayable idempotent request for as long + as failures land on *reused* pooled connections. It tries at most + `max_idle_per_host` reused connections, then forces a fresh dial instead of + accepting another concurrent pool return. PUT and DELETE receive the same + stale-connection recovery as the other idempotent methods. + Pooled connections can be discarded by the peer in correlated batches, in + which case the single retry would draw the next equally-dead pooled + connection and fail. ([#1353]) +- Dead reused TLS connections that fail with `Reseau.TLS.TLSError` (for + example an RST surfacing as a wrapped `SystemError` or a truncated TLS record + reported as `unexpected EOF`) are now classified by their public error shape + in the transport's reused-connection retry. HTTP/2 read-loop wrappers also + preserve this classification. ([#1353]) ## [v2.0.0] - 2026-04-27 HTTP.jl 2.0 is a major rewrite of the package internals and public API. The @@ -826,3 +865,4 @@ See changes for 0.9.15: this release is equivalent to 0.9.15 with [#752] reverte [#1126]: https://github.com/JuliaWeb/HTTP.jl/issues/1126 [#1127]: https://github.com/JuliaWeb/HTTP.jl/issues/1127 [#1342]: https://github.com/JuliaWeb/HTTP.jl/issues/1342 +[#1353]: https://github.com/JuliaWeb/HTTP.jl/issues/1353 diff --git a/docs/src/api/client.md b/docs/src/api/client.md index a30e13e5c..0c063eb22 100644 --- a/docs/src/api/client.md +++ b/docs/src/api/client.md @@ -43,6 +43,7 @@ HTTP.isaborted HTTP.RequestEvent HTTP.ResponseHeadEvent HTTP.RetryEvent +HTTP.RetrySkippedEvent HTTP.RedirectEvent HTTP.DoneEvent ``` diff --git a/docs/src/api/core.md b/docs/src/api/core.md index 5ac9dfc65..4900d6ad2 100644 --- a/docs/src/api/core.md +++ b/docs/src/api/core.md @@ -27,6 +27,7 @@ HTTP.TooManyRedirectsError HTTP.ConnectError HTTP.DNSError HTTP.TLSHandshakeError +HTTP.TLSTransportError HTTP.AddressInUseError ``` diff --git a/docs/src/guides/client.md b/docs/src/guides/client.md index bcf4b68f1..74d49750a 100644 --- a/docs/src/guides/client.md +++ b/docs/src/guides/client.md @@ -511,6 +511,7 @@ logger — pass a `trace` callback. The callback receives subtypes of - [`HTTP.RequestEvent`](@ref) — request being sent - [`HTTP.ResponseHeadEvent`](@ref) — response headers received - [`HTTP.RetryEvent`](@ref) — retry scheduled +- [`HTTP.RetrySkippedEvent`](@ref) — retry denied by the budget or deadline - [`HTTP.RedirectEvent`](@ref) — redirect followed - [`HTTP.DoneEvent`](@ref) — request finished (with response or error) diff --git a/src/HTTP.jl b/src/HTTP.jl index f2aebf7cc..efc4596ef 100644 --- a/src/HTTP.jl +++ b/src/HTTP.jl @@ -78,7 +78,8 @@ include("http_websockets.jl") :Handlers, :Headers, :NoProxy, :ParseError, :ProtocolError, :ProxyConfig, :ProxyFromEnvironment, :ProxyURL, :RedirectEvent, :Request, :RequestContext, :RequestEvent, :RequestRetryError, :Response, :ResponseHeadEvent, :RetryBucket, - :RetryEvent, :SSEEvent, :SSEStream, :Server, :StatusError, :Stream, :TLSHandshakeError, + :RetryEvent, :RetrySkippedEvent, :SSEEvent, :SSEStream, :Server, :StatusError, :Stream, + :TLSHandshakeError, :TLSTransportError, :TimeoutError, :TooManyRedirectsError, :Transport, :addtrailer, :appendheader, :body_close!, :body_closed, :body_read!, :cancel!, :canceled, :canonical_header_key, :close_idle_connections!, :defaultheader!, :delete, :do!, :expired, :fileserver, diff --git a/src/http2_client.jl b/src/http2_client.jl index dd725156f..2e65cba64 100644 --- a/src/http2_client.jl +++ b/src/http2_client.jl @@ -471,8 +471,14 @@ end @inline function _throw_stream_error(conn::H2Connection, state::H2StreamState)::Nothing err = state.stream_error - err === nothing || throw(err::Exception) - state.conn_errored && throw(_stream_conn_error(conn)) + if err !== nothing + wrapped = _wrap_tls_transport_error(err::Exception) + throw(wrapped) + end + if state.conn_errored + wrapped = _wrap_tls_transport_error(_stream_conn_error(conn)) + throw(wrapped) + end return nothing end @@ -1213,10 +1219,17 @@ function _connect_h2_from_tcp!( stream_reader = nothing connect_deadline_ns == 0 || TCP.set_deadline!(tcp, connect_deadline_ns) if secure - cfg = _make_tls_config_for_h2(tls_config, address) - tls_conn = TLS.client(tcp, cfg) - connect_deadline_ns == 0 || TLS.set_deadline!(tls_conn, connect_deadline_ns) - TLS.handshake!(tls_conn) + try + cfg = _make_tls_config_for_h2(tls_config, address) + tls_conn = TLS.client(tcp, cfg) + connect_deadline_ns == 0 || TLS.set_deadline!(tls_conn, connect_deadline_ns) + TLS.handshake!(tls_conn) + catch err + # TLS.client can fail while it initializes client state, before + # handshake! starts. Both operations are connection setup. + err isa TLS.TLSError && throw(TLSHandshakeError(err::TLS.TLSError)) + rethrow() + end stream_reader = _ConnReader(tls_conn::TLS.Conn) else stream_reader = _ConnReader(tcp) @@ -1653,7 +1666,7 @@ function body_read!(body::H2Body, dst::Vector{UInt8})::Int @atomic :release body.closed = true _clear_h2_cancel_callback!(body) _unregister_stream!(body.conn, body.stream_id) - throw(terminal_error::Exception) + throw(_wrap_tls_transport_error(terminal_error::Exception)) end if too_many @atomic :release body.closed = true @@ -1664,7 +1677,12 @@ function body_read!(body::H2Body, dst::Vector{UInt8})::Int end if nread > 0 body.bytes_read += Int64(nread) - _send_window_updates!(body.conn, body.stream_id, nread) + try + _send_window_updates!(body.conn, body.stream_id, nread) + catch err + wrapped = err isa Exception ? _wrap_tls_transport_error(err::Exception) : err + wrapped === err ? rethrow() : throw(wrapped) + end return nread end if done @@ -1884,5 +1902,10 @@ Send `request` over an existing `H2Connection` and return the streaming `Response`. """ function h2_roundtrip!(conn::H2Connection, request::Request)::Response - return _streaming_response(_h2_roundtrip_incoming!(conn, request)) + try + return _streaming_response(_h2_roundtrip_incoming!(conn, request)) + catch err + wrapped = err isa Exception ? _wrap_tls_transport_error(err::Exception) : err + wrapped === err ? rethrow() : throw(wrapped) + end end diff --git a/src/http_client.jl b/src/http_client.jl index 794b2b77c..5b94f6e63 100644 --- a/src/http_client.jl +++ b/src/http_client.jl @@ -144,6 +144,35 @@ struct RetryEvent <: ClientEvent err::Union{Nothing,Exception} end +""" + RetrySkippedEvent + +Trace event emitted when the retry policy wanted to retry an attempt but the +retry was not armed, so the attempt's failure becomes the request's outcome. +Without this event a denied retry is indistinguishable from a non-retryable +failure. + +Fields: +- `request`: request metadata for the attempt that will not be retried +- `url`: absolute request URL for the attempt +- `attempt`: current 1-based attempt number +- `redirect_count`: redirects already followed when the retry decision was made +- `reason`: `:retry_bucket` when the transport's [`RetryBucket`](@ref) denied + retry capacity; `:deadline` when the request deadline preempted the retry + backoff +- `response`: retry-triggering response, or `nothing` for request-path failures +- `err`: retry-triggering exception, or `nothing` for response-based retries +""" +struct RetrySkippedEvent <: ClientEvent + request::Request + url::String + attempt::Int + redirect_count::Int + reason::Symbol + response::Union{Nothing,Response} + err::Union{Nothing,Exception} +end + """ RedirectEvent @@ -308,6 +337,18 @@ function (trace::_VerboseTrace)(event::RetryEvent)::Nothing return nothing end +function (trace::_VerboseTrace)(event::RetrySkippedEvent)::Nothing + detail = if event.err !== nothing + sprint(showerror, event.err::Exception) + elseif event.response !== nothing + string("status ", (event.response::Response).status) + else + "failure" + end + _verbose_line!(string("retry of attempt ", event.attempt, " skipped (", event.reason, ") after ", detail)) + return nothing +end + function (trace::_VerboseTrace)(event::RedirectEvent)::Nothing _verbose_line!(string("redirect ", event.response.status, " ", event.from_url, " -> ", event.to_url)) return nothing @@ -845,18 +886,29 @@ function _do_incoming!( previous_response = nothing retry_attempt = 1 retry_token = nothing + retry_custom_wanted = false for redirect_count in 0:redirect_policy.max_redirects while true - send_request = _copy_request_for_send(current_request, retry_attempt == 1) - request_url = _request_url(current_secure, current_address, current_request.target) - proxy_plan = _proxy_plan(proxy_config, current_secure, current_address) - use_h2 = proxy_plan.mode != _ProxyPlanMode.HTTP_FORWARD && - _use_h2(client, proxy_plan, current_secure, protocol) - _emit_trace(trace, RequestEvent(send_request, request_url, retry_attempt, redirect_count, use_h2 ? :h2 : :h1)) - host, path = _host_path_from_request(current_address, current_request) - manual_cookies = cookies === false ? Cookie[] : Cookies.readcookies(send_request.headers, "") - cookie_value = _cookie_header(cookiejar, cookies, current_secure, host, path, manual_cookies) - cookie_value === nothing || setheader(send_request.headers, "Cookie", cookie_value) + send_request, request_url, proxy_plan, use_h2, host, path = try + next_request = _copy_request_for_send(current_request, retry_attempt == 1) + next_url = _request_url(current_secure, current_address, current_request.target) + next_proxy_plan = _proxy_plan(proxy_config, current_secure, current_address) + next_use_h2 = next_proxy_plan.mode != _ProxyPlanMode.HTTP_FORWARD && + _use_h2(client, next_proxy_plan, current_secure, protocol) + _emit_trace(trace, RequestEvent(next_request, next_url, retry_attempt, redirect_count, next_use_h2 ? :h2 : :h1)) + next_host, next_path = _host_path_from_request(current_address, current_request) + manual_cookies = cookies === false ? Cookie[] : Cookies.readcookies(next_request.headers, "") + cookie_value = _cookie_header(cookiejar, cookies, current_secure, next_host, next_path, manual_cookies) + cookie_value === nothing || setheader(next_request.headers, "Cookie", cookie_value) + (next_request, next_url, next_proxy_plan, next_use_h2, next_host, next_path) + catch + if retry_token !== nothing + @try_ignore _settle_request_retry_token!(retry_token::RetryToken, 0) + retry_token = nothing + retry_custom_wanted = false + end + rethrow() + end response = try if use_h2 conn = nothing @@ -898,59 +950,107 @@ function _do_incoming!( ) end catch err - if retry_controller !== nothing && retry_controller.bucket !== nothing && retry_token !== nothing - release(retry_controller.bucket::RetryBucket, retry_token::RetryToken, _RETRY_BUCKET_ACQUIRE_COST) + if retry_token !== nothing + _settle_request_retry_token!(retry_token::RetryToken, _RETRY_BUCKET_ACQUIRE_COST) end retry_token = nothing + retry_custom_wanted = false if retry_controller !== nothing - if _should_retry_request_attempt(retry_controller, retry_attempt, current_request, RequestRetryError(err::Exception), nothing) - scheduled, next_token, delay_ns = _arm_request_retry!(retry_controller, current_address, current_request, retry_attempt, nothing) + can_retry = retry_controller.enabled && retry_controller.remaining > 0 + retry_wanted, custom_wanted = can_retry ? + _retry_policy_decision(retry_controller, retry_attempt, current_request, RequestRetryError(err::Exception), nothing) : + (false, false) + if retry_wanted + scheduled, next_token, delay_ns, skip_reason = _arm_request_retry!(retry_controller, current_address, current_request, retry_attempt, nothing) if scheduled - _emit_trace(trace, RetryEvent(current_request, request_url, retry_attempt, retry_attempt + 1, redirect_count, delay_ns, nothing, err::Exception)) - get_request_context(current_request)[:retryattempt] = retry_attempt + try + _emit_trace(trace, RetryEvent(current_request, request_url, retry_attempt, retry_attempt + 1, redirect_count, delay_ns, nothing, err::Exception)) + get_request_context(current_request)[:retryattempt] = retry_attempt + catch + next_token === nothing || @try_ignore _settle_request_retry_token!(next_token::RetryToken, 0) + rethrow() + end retry_attempt += 1 retry_token = next_token + retry_custom_wanted = custom_wanted continue end + skip_reason === nothing || _emit_trace(trace, RetrySkippedEvent(current_request, request_url, retry_attempt, redirect_count, skip_reason::Symbol, nothing, err::Exception)) end end rethrow(err) end - response = _annotate_incoming_response( - response, - request_url, - previous_response, - redirect_count, - ) - _store_set_cookies!(cookiejar, cookies, current_secure, host, path, response.head.headers) - status_response = _retry_policy_response(response, current_request) - _emit_trace(trace, ResponseHeadEvent(status_response, request_url, retry_attempt, redirect_count)) - if retry_controller !== nothing && retry_controller.bucket !== nothing && retry_token !== nothing - release(retry_controller.bucket::RetryBucket, retry_token::RetryToken, _retry_bucket_failure_cost(status_response.status)) - end - retry_token = nothing - if retry_controller !== nothing - should_retry = try - _should_retry_request_attempt(retry_controller, retry_attempt, current_request, nothing, status_response) - catch - @try_ignore begin - body_close!(response.rawbody) + response = try + response = _annotate_incoming_response( + response, + request_url, + previous_response, + redirect_count, + ) + _store_set_cookies!(cookiejar, cookies, current_secure, host, path, response.head.headers) + status_response = _retry_policy_response(response, current_request) + _emit_trace(trace, ResponseHeadEvent(status_response, request_url, retry_attempt, redirect_count)) + + can_retry = retry_controller !== nothing && retry_controller.enabled && retry_controller.remaining > 0 + retry_wanted, custom_wanted = can_retry ? + _retry_policy_decision(retry_controller, retry_attempt, current_request, nothing, status_response) : + (false, false) + if retry_controller !== nothing && retry_controller.bucket !== nothing + response_bucket = retry_controller.bucket::RetryBucket + if retry_token !== nothing + terminal_builtin_failure = !can_retry && + _retry_builtin_wants_retry(retry_controller, current_request, nothing, status_response) + terminal_custom_failure = !can_retry && retry_custom_wanted && + !_successful_response_status(status_response.status) + failure_cost = _retry_bucket_response_cost( + retry_wanted || terminal_builtin_failure || terminal_custom_failure, + ) + _settle_request_retry_token!(retry_token::RetryToken, failure_cost) + elseif !retry_wanted && _successful_response_status(status_response.status) + # Healthy non-retried traffic slowly restores retry + # capacity consumed by an earlier failure burst. + _retry_bucket_replenish!(response_bucket, _retry_partition_for_address(current_address)) end - rethrow() end - if should_retry - scheduled, next_token, delay_ns = _arm_request_retry!(retry_controller, current_address, current_request, retry_attempt, status_response) + retry_token = nothing + retry_custom_wanted = false + + if retry_wanted + scheduled, next_token, delay_ns, skip_reason = _arm_request_retry!(retry_controller, current_address, current_request, retry_attempt, status_response) if scheduled - _emit_trace(trace, RetryEvent(current_request, request_url, retry_attempt, retry_attempt + 1, redirect_count, delay_ns, status_response, nothing)) - get_request_context(current_request)[:retryattempt] = retry_attempt + try + _emit_trace(trace, RetryEvent(current_request, request_url, retry_attempt, retry_attempt + 1, redirect_count, delay_ns, status_response, nothing)) + get_request_context(current_request)[:retryattempt] = retry_attempt + catch + next_token === nothing || @try_ignore _settle_request_retry_token!(next_token::RetryToken, 0) + rethrow() + end retry_attempt += 1 retry_token = next_token + retry_custom_wanted = custom_wanted @try_ignore begin body_close!(response.rawbody) end continue end + skip_reason === nothing || _emit_trace(trace, RetrySkippedEvent(current_request, request_url, retry_attempt, redirect_count, skip_reason::Symbol, status_response, nothing)) + end + + response + catch + if retry_token !== nothing + # A response arrived, but a trace or policy callback failed + # before the custom decision could settle the reservation. + fallback_cost = _retry_bucket_failure_cost(response.head.status) + if retry_custom_wanted && !_successful_response_status(response.head.status) + fallback_cost = max(fallback_cost, _RETRY_BUCKET_RETRYABLE_RESPONSE_COST) + end + @try_ignore _settle_request_retry_token!(retry_token::RetryToken, fallback_cost) + retry_token = nothing + retry_custom_wanted = false end + @try_ignore body_close!(response.rawbody) + rethrow() end if !_is_redirect_status(response.head.status) return response @@ -1119,7 +1219,7 @@ function do!( end end elapsed_ns = Int64(time_ns()) - start_ns - wrapped = _wrap_client_transport_error(err, "request", timeout_ns, elapsed_ns) + wrapped = err isa Exception ? _wrap_client_transport_error(err::Exception, "request", timeout_ns, elapsed_ns) : err wrapped === err ? rethrow() : throw(wrapped) end end @@ -2140,7 +2240,7 @@ function request( throw(wrapped) end elapsed_ns = Int64(time_ns()) - request_start_ns - wrapped = _wrap_client_transport_error(err, "request", request_timeout_ns, elapsed_ns) + wrapped = err isa Exception ? _wrap_client_transport_error(err::Exception, "request", request_timeout_ns, elapsed_ns) : err final_error = wrapped::Exception wrapped === err ? rethrow() : throw(wrapped) finally @@ -2156,7 +2256,8 @@ High-level one-shot HTTP request API. When `trace` is provided, it must be callable on any emitted client event. Current events are [`RequestEvent`](@ref), [`ResponseHeadEvent`](@ref), -[`RetryEvent`](@ref), [`RedirectEvent`](@ref), and [`DoneEvent`](@ref). +[`RetryEvent`](@ref), [`RetrySkippedEvent`](@ref), [`RedirectEvent`](@ref), and +[`DoneEvent`](@ref). Keyword arguments: - `basicauth`: optional basic-auth credentials supplied as diff --git a/src/http_client_retry.jl b/src/http_client_retry.jl index 11c9c2645..b6df3dd24 100644 --- a/src/http_client_retry.jl +++ b/src/http_client_retry.jl @@ -24,10 +24,6 @@ end return status == 408 || status == 429 || status == 500 || status == 502 || status == 503 || status == 504 end -@inline function _retryable_request_method(method::String)::Bool - return method == "GET" || method == "HEAD" || method == "OPTIONS" || method == "TRACE" || method == "PUT" || method == "DELETE" || method == "QUERY" -end - @inline function _retryable_request_headers(request::Request)::Bool key = header(request.headers, "Idempotency-Key", nothing) key !== nothing && !isempty(key::String) && return true @@ -35,10 +31,6 @@ end return legacy !== nothing && !isempty(legacy::String) end -@inline function _retryable_request_body(request::Request)::Bool - return request.content_length == 0 || request.body isa EmptyBody || request.body isa BytesBody -end - @inline function _retryable_policy_request(request::Request, retry_non_idempotent::Bool)::Bool _retryable_request_body(request) || return false retry_non_idempotent && return true @@ -60,11 +52,26 @@ function _retryable_request_error(err::Exception)::Bool current isa IOPoll.NotPollableError && return true current isa IOPoll.DeadlineExceededError && return false current isa TLS.TLSHandshakeTimeoutError && return true + if current isa ProtocolError + cause = (current::ProtocolError).err + cause === nothing && return false + current = cause::Exception + continue + end if current isa HostResolvers.OpError current = (current::HostResolvers.OpError).err continue end + if current isa TLSHandshakeError + current = (current::TLSHandshakeError).cause + continue + end + if current isa TLSTransportError + current = (current::TLSTransportError).cause + continue + end if current isa TLS.TLSError + (current::TLS.TLSError).message == _TLS_TRUNCATED_STREAM_MESSAGE && return true cause = (current::TLS.TLSError).cause cause === nothing && return false current = cause::Exception @@ -85,7 +92,9 @@ applies to request-path exceptions. Recoverable cases include connection resets and EOFs (`EOFError`, `IOPoll.NetClosingError`), socket errors (`SystemError`), malformed responses (`ParseError`), and dial/handshake timeouts (`HostResolvers.DialTimeoutError`, `TLS.TLSHandshakeTimeoutError`), including the -underlying causes of wrapped `HostResolvers.OpError`/`TLS.TLSError` exceptions. +underlying causes of wrapped `HostResolvers.OpError`/`TLS.TLSError` exceptions, +HTTP/2 `ProtocolError` connection wrappers, and the public +[`TLSHandshakeError`](@ref)/[`TLSTransportError`](@ref) wrappers. A request *deadline* being exceeded (`IOPoll.DeadlineExceededError`) is treated as non-recoverable, as is anything else. @@ -120,20 +129,44 @@ function _retry_hook_decision(controller::_RetryController, attempt::Int, err, r return decision end -function _should_retry_request_attempt(controller::_RetryController, attempt::Int, req::Request, err, resp)::Bool - controller.enabled || return false - controller.remaining > 0 || return false +function _retry_builtin_wants_retry(controller::_RetryController, req::Request, err, resp)::Bool _retryable_request_body(req) || return false - built_in = false if err !== nothing policy_ok = _retryable_policy_request(req, controller.retry_non_idempotent) || _h2_guaranteed_unprocessed(err) - built_in = policy_ok && _retryable_request_error(err) + return policy_ok && _retryable_request_error(err) elseif resp !== nothing - built_in = _retryable_policy_request(req, controller.retry_non_idempotent) && _retryable_status((resp::Response).status) + return _retryable_policy_request(req, controller.retry_non_idempotent) && _retryable_status((resp::Response).status) end + return false +end + +function _retry_policy_decision(controller::_RetryController, attempt::Int, req::Request, err, resp)::Tuple{Bool,Bool} + _retryable_request_body(req) || return (false, false) + built_in = _retry_builtin_wants_retry(controller, req, err, resp) decision = _retry_hook_decision(controller, attempt, err, req, resp) - decision === nothing && return built_in - return decision::Bool + decision === nothing && return (built_in, false) + wants_retry = decision::Bool + return (wants_retry, wants_retry) +end + +function _retry_policy_wants_retry(controller::_RetryController, attempt::Int, req::Request, err, resp)::Bool + wants_retry, _ = _retry_policy_decision(controller, attempt, req, err, resp) + return wants_retry +end + +function _should_retry_request_attempt(controller::_RetryController, attempt::Int, req::Request, err, resp)::Bool + controller.enabled || return false + controller.remaining > 0 || return false + return _retry_policy_wants_retry(controller, attempt, req, err, resp) +end + +@inline function _settle_request_retry_token!(token::RetryToken, failure_cost::Int)::Nothing + release(token.bucket, token, failure_cost) + return nothing +end + +@inline function _successful_response_status(status::Int)::Bool + return 200 <= status < 400 end @inline function _retry_bucket_for_request(client::Client, retry_bucket::Union{Bool,RetryBucket}) @@ -169,20 +202,31 @@ end return nothing end -function _sleep_retry_delay!(request::Request, delay_ns::Int64)::Bool +function _sleep_retry_delay!( + request::Request, + delay_ns::Int64; + clock_ns::Function=time_ns, + sleep_ns::Function=IOPoll.sleep_ns, +)::Bool delay_ns < 0 && return false deadline_ns = _request_deadline_ns(request) if deadline_ns != 0 - now_ns = Int64(time_ns()) + now_ns = Int64(clock_ns()) now_ns >= deadline_ns && return false now_ns > typemax(Int64) - delay_ns && return false now_ns + delay_ns <= deadline_ns || return false end delay_ns == 0 && return true - IOPoll.sleep_ns(delay_ns) + sleep_ns(delay_ns) + deadline_ns != 0 && Int64(clock_ns()) >= deadline_ns && return false return true end +# Arm one retry attempt: reserve bucket capacity, sleep the backoff, and +# consume one of the controller's remaining retries. Returns +# `(armed, token, delay_ns, skip_reason)` where `skip_reason` is `nothing` when +# the retry was armed, `:retry_bucket` when the bucket denied capacity, or +# `:deadline` when the request deadline preempted the backoff. function _arm_request_retry!( controller::_RetryController, address::AbstractString, @@ -198,18 +242,18 @@ function _arm_request_retry!( token = Base.acquire(bucket::RetryBucket, _retry_partition_for_address(address)) catch err err isa RetryDeniedError || rethrow(err) - return false, nothing, delay_ns + return false, nothing, delay_ns, :retry_bucket end end ok = false try - _sleep_retry_delay!(request, delay_ns) || return false, nothing, delay_ns + _sleep_retry_delay!(request, delay_ns) || return false, nothing, delay_ns, :deadline ok = true finally ok || (bucket !== nothing && token !== nothing && release(bucket::RetryBucket, token, _retry_bucket_failure_cost(nothing))) end controller.remaining -= 1 - return true, token, delay_ns + return true, token, delay_ns, nothing end function _retry_controller( diff --git a/src/http_core.jl b/src/http_core.jl index 14ab2bb91..769565e7d 100644 --- a/src/http_core.jl +++ b/src/http_core.jl @@ -119,12 +119,27 @@ DNSError(hostname::AbstractString, cause::Exception) = DNSError(String(hostname) TLSHandshakeError(cause) Raised when a TLS handshake fails (other than a handshake timeout, which -surfaces as [`TimeoutError`](@ref) with `operation = "tls_handshake"`). +surfaces as [`TimeoutError`](@ref) with `operation = "tls_handshake"`). TLS +failures after connection setup surface as [`TLSTransportError`](@ref) instead. """ struct TLSHandshakeError <: HTTPError cause::Exception end +""" + TLSTransportError(cause) + +Raised when TLS-level I/O fails on an established connection during a request +— for example a connection reset or a truncated TLS stream while writing the +request or reading the response. `cause` is the underlying +`Reseau.TLS.TLSError`. TLS failures during connection setup surface as +[`TLSHandshakeError`](@ref) (or [`TimeoutError`](@ref) for handshake timeouts) +instead. +""" +struct TLSTransportError <: HTTPError + cause::Exception +end + """ AddressInUseError(address) @@ -190,6 +205,12 @@ function Base.showerror(io::IO, err::TLSHandshakeError) return nothing end +function Base.showerror(io::IO, err::TLSTransportError) + print(io, "http tls transport error: ") + showerror(io, err.cause) + return nothing +end + function Base.showerror(io::IO, err::AddressInUseError) print(io, "http address already in use: ", err.address) return nothing @@ -277,9 +298,16 @@ end Wrap Reseau-internal transport errors that escape the client request path so callers see HTTP-typed exceptions only. Returns either a wrapped [`TimeoutError`](@ref), [`ConnectError`](@ref), [`DNSError`](@ref), -[`TLSHandshakeError`](@ref), or `err` unchanged when no wrapping rule applies. +[`TLSHandshakeError`](@ref), [`TLSTransportError`](@ref), or `err` unchanged +when no wrapping rule applies. + +A bare `TLS.TLSError` reaching this boundary arose on an established +connection (handshake failures are typed `TLSHandshakeError` at the dial +sites), so it wraps as [`TLSTransportError`](@ref). """ -function _wrap_client_transport_error(err, operation::AbstractString="request", timeout_ns::Integer=Int64(0), elapsed_ns::Integer=Int64(0)) +_wrap_client_transport_error(err, operation::AbstractString="request", timeout_ns::Integer=Int64(0), elapsed_ns::Integer=Int64(0)) = err + +@inline function _wrap_client_transport_error(err::Exception, operation::AbstractString="request", timeout_ns::Integer=Int64(0), elapsed_ns::Integer=Int64(0)) if err isa TLS.TLSHandshakeTimeoutError return TimeoutError(String("tls_handshake"), Int64(err.timeout_ns), Int64(elapsed_ns)) end @@ -295,12 +323,32 @@ function _wrap_client_transport_error(err, operation::AbstractString="request", if err isa HostResolvers.LookupError return DNSError(err.name, err) end - if err isa TLS.TLSError - return TLSHandshakeError(err) - end + return _wrap_tls_transport_error(err) +end + +_wrap_tls_transport_error(err) = err + +@inline function _wrap_tls_transport_error(err::Exception) + err isa TLSTransportError && return err + err isa TLS.TLSError && return TLSTransportError(err) + nested_tls = _find_nested_tls_transport_error(err) + nested_tls === nothing || return TLSTransportError(nested_tls::TLS.TLSError) return err end +function _find_nested_tls_transport_error(err)::Union{Nothing,TLS.TLSError} + current = err + while current isa ProtocolError + cause = (current::ProtocolError).err + cause === nothing && return nothing + current = cause::Exception + end + current isa TLS.TLSError && return current::TLS.TLSError + current isa TLSTransportError || return nothing + cause = (current::TLSTransportError).cause + return cause isa TLS.TLSError ? cause::TLS.TLSError : nothing +end + """ _wrap_server_listen_error(err, address) diff --git a/src/http_retry.jl b/src/http_retry.jl index 28c36b10a..70c4f21a3 100644 --- a/src/http_retry.jl +++ b/src/http_retry.jl @@ -12,6 +12,23 @@ const _RETRY_BUCKET_RETRYABLE_RESPONSE_COST = 5 const _RETRY_BUCKET_DEFAULT_BACKOFF_SCALE_FACTOR_NS = Int64(_RETRY_BUCKET_DEFAULT_BACKOFF_SCALE_FACTOR_MS) * Int64(1_000_000) const _RETRY_BUCKET_DEFAULT_MAX_BACKOFF_NS = Int64(_RETRY_BUCKET_DEFAULT_MAX_BACKOFF_SECS) * Int64(1_000_000_000) +# Reseau's TLS layer reports a stream cut mid-record or before close_notify as +# a `TLSError` carrying exactly this message. Its cause is a Reseau-private +# type, so the retry classifiers match the public message instead; the +# end-to-end truncation tests (transport and HTTP/2) pin this coupling so a +# Reseau wording change fails loudly rather than silently dropping the +# classification. +const _TLS_TRUNCATED_STREAM_MESSAGE = "unexpected EOF" + +@inline function _retryable_request_method(method::String)::Bool + return method == "GET" || method == "HEAD" || method == "OPTIONS" || method == "TRACE" || + method == "PUT" || method == "DELETE" || method == "QUERY" +end + +@inline function _retryable_request_body(request::Request)::Bool + return request.content_length == 0 || request.body isa EmptyBody || request.body isa BytesBody +end + """Retry capacity tracked independently for one retry partition key.""" mutable struct _RetryPartition capacity::Int @@ -27,6 +44,16 @@ reserves retry capacity for one retry attempt, and `release(bucket, token, failure_cost)` returns all or part of that reserved capacity. Use `failure_cost = 0` for a full refund, or a positive cost to keep some or all of the reserved retry capacity consumed. + +The built-in client retry flow refunds a reservation in full when the retried +attempt reaches a non-retryable response (the retry did its job), keeps part of +the cost when the built-in or custom policy still classifies the response as a +failure, keeps a conservative partial cost for a non-success terminal response +after a retry explicitly requested by `retry_if`, keeps the full cost when the +retried attempt fails with an exception, +and slowly restores consumed capacity by crediting one unit per successful +non-retried response — so a burst of real failures can drain a partition, but +healthy traffic always heals it. """ mutable struct RetryBucket backoff_scale_factor_ms::Int @@ -34,6 +61,10 @@ mutable struct RetryBucket capacity::Int partitions::Dict{String,_RetryPartition} lock::ReentrantLock + # Copy-on-write snapshot of partition keys below full capacity. Published + # snapshots are treated as immutable. Writers publish under `lock`; readers + # use the snapshot to avoid locking for healthy traffic to unrelated keys. + @atomic depleted_partitions::Set{String} end """Handle returned by `acquire` and consumed by `release` to refund retry budget.""" @@ -74,9 +105,71 @@ function RetryBucket(; Int(capacity), Dict{String,_RetryPartition}(), ReentrantLock(), + Set{String}(), + ) +end + +# Preserve the positional constructor that the five-field RetryBucket exposed +# before the depleted-partition fast path was added. +function RetryBucket( + backoff_scale_factor_ms::Int, + max_backoff_secs::Int, + capacity::Int, + partitions::Dict{String,_RetryPartition}, + lock::ReentrantLock, +) + depleted_partitions = Set( + key for (key, state) in partitions if state.capacity < capacity + ) + return RetryBucket( + backoff_scale_factor_ms, + max_backoff_secs, + capacity, + partitions, + lock, + depleted_partitions, ) end +function RetryBucket( + backoff_scale_factor_ms, + max_backoff_secs, + capacity, + partitions, + lock, +) + return RetryBucket( + convert(Int, backoff_scale_factor_ms), + convert(Int, max_backoff_secs), + convert(Int, capacity), + convert(Dict{String,_RetryPartition}, partitions), + convert(ReentrantLock, lock), + ) +end + +# Set a partition's capacity while keeping the published depleted-key snapshot +# in sync. Must be called with `bucket.lock` held. +@inline function _retry_partition_set_capacity!( + bucket::RetryBucket, + partition_key::String, + state::_RetryPartition, + new_capacity::Int, +)::Nothing + was_full = state.capacity >= bucket.capacity + now_full = new_capacity >= bucket.capacity + state.capacity = new_capacity + if was_full && !now_full + depleted = copy(@atomic :acquire bucket.depleted_partitions) + push!(depleted, partition_key) + @atomic :release bucket.depleted_partitions = depleted + elseif !was_full && now_full + depleted = copy(@atomic :acquire bucket.depleted_partitions) + delete!(depleted, partition_key) + @atomic :release bucket.depleted_partitions = depleted + end + return nothing +end + @inline function _retry_bucket_partition_key(partition)::String partition === nothing && throw(ArgumentError("retry bucket partition is required")) key = lowercase(String(partition)) @@ -97,7 +190,7 @@ function acquire(bucket::RetryBucket, partition) if state.capacity < _RETRY_BUCKET_ACQUIRE_COST throw(RetryDeniedError(partition_key)) end - state.capacity -= _RETRY_BUCKET_ACQUIRE_COST + _retry_partition_set_capacity!(bucket, partition_key, state, state.capacity - _RETRY_BUCKET_ACQUIRE_COST) return RetryToken(bucket, partition_key, _RETRY_BUCKET_ACQUIRE_COST, false) end end @@ -106,12 +199,18 @@ end return min(token.reserved_capacity, _RETRY_BUCKET_ACQUIRE_COST) end +# Cost to keep from a retry reservation given the retried attempt's response +# status. A non-retryable status means the retry did its job (the request +# completed, whatever the outcome), so the reservation is refunded in full; +# charging successes would make the budget a strictly-decreasing resource that +# eventually denies every retry for the transport's lifetime (#1353). @inline function _retry_bucket_failure_cost(status::Union{Nothing,Int})::Int status === nothing && return 0 - if status == 429 || (500 <= status < 600) - return _RETRY_BUCKET_RETRYABLE_RESPONSE_COST - end - return _RETRY_BUCKET_ACQUIRE_COST + return _retryable_status(status) ? _RETRY_BUCKET_RETRYABLE_RESPONSE_COST : 0 +end + +@inline function _retry_bucket_response_cost(retry_wanted::Bool)::Int + return retry_wanted ? _RETRY_BUCKET_RETRYABLE_RESPONSE_COST : 0 end @inline function release(bucket::RetryBucket, token::RetryToken, failure_cost::Int)::Nothing @@ -123,7 +222,7 @@ end reserved = _retry_bucket_reserved_cost(token) consumed = min(reserved, max(0, failure_cost)) refund = reserved - consumed - state.capacity = min(bucket.capacity, state.capacity + refund) + _retry_partition_set_capacity!(bucket, token.partition, state, min(bucket.capacity, state.capacity + refund)) token.released = true return nothing finally @@ -131,6 +230,35 @@ end end end +""" + _retry_bucket_replenish!(bucket, partition) + +Credit one unit of retry capacity back to `partition` after a successful +non-retried request, capped at the bucket's full capacity. This is the slow +recovery path that lets a partition legitimately drained by a burst of real +failures regain retry budget from healthy traffic instead of staying empty for +the transport's lifetime. Partitions that have never spent capacity are left +untouched. The published depleted-key snapshot also keeps this lock-free for +healthy traffic to other partitions. +""" +function _retry_bucket_replenish!(bucket::RetryBucket, partition)::Nothing + depleted = @atomic :acquire bucket.depleted_partitions + isempty(depleted) && return nothing + partition_key = _retry_bucket_partition_key(partition) + partition_key in depleted || return nothing + lock(bucket.lock) + try + state = get(() -> nothing, bucket.partitions, partition_key) + state === nothing && return nothing + partition_state = state::_RetryPartition + partition_state.capacity >= bucket.capacity && return nothing + _retry_partition_set_capacity!(bucket, partition_key, partition_state, partition_state.capacity + 1) + return nothing + finally + unlock(bucket.lock) + end +end + @inline function _retry_bucket_max_backoff_ns(bucket::RetryBucket)::Int64 max_secs = max(0, bucket.max_backoff_secs) max_secs > typemax(Int64) ÷ 1_000_000_000 && return typemax(Int64) diff --git a/src/http_stream.jl b/src/http_stream.jl index 6b2dd434a..23b002cae 100644 --- a/src/http_stream.jl +++ b/src/http_stream.jl @@ -179,20 +179,25 @@ function _client_start_stream_read!(stream::Stream{true})::Response meta.close, get_request_context(meta), ) - incoming = _do_incoming!( - nothing, - stream.client::Client, - (stream.parsed::_URLParts).address, - req, - (stream.parsed::_URLParts).secure, - nothing, - stream.protocol, - stream.redirect ? (stream.redirect_policy::_RedirectPolicy) : _redirect_policy(stream.client::Client, 0), - stream.retry_controller, - stream.proxy_config, - stream.cookies, - stream.cookiejar, - ) + incoming = try + _do_incoming!( + nothing, + stream.client::Client, + (stream.parsed::_URLParts).address, + req, + (stream.parsed::_URLParts).secure, + nothing, + stream.protocol, + stream.redirect ? (stream.redirect_policy::_RedirectPolicy) : _redirect_policy(stream.client::Client, 0), + stream.retry_controller, + stream.proxy_config, + stream.cookies, + stream.cookiejar, + ) + catch err + wrapped = err isa Exception ? _wrap_tls_transport_error(err::Exception) : err + wrapped === err ? rethrow() : throw(wrapped) + end resolved_request = incoming.head.request === nothing ? req : incoming.head.request::Request stream.response = _finalize_request_response( incoming, @@ -562,7 +567,7 @@ function open( msg = (ctx_kw::RequestContext).cancel_message === nothing ? "request canceled" : (ctx_kw::RequestContext).cancel_message::String throw(CanceledError(msg)) end - wrapped = _wrap_client_transport_error(err, "request", Int64(0), elapsed_ns) + wrapped = err isa Exception ? _wrap_client_transport_error(err::Exception, "request", Int64(0), elapsed_ns) : err wrapped === err ? rethrow() : throw(wrapped) end try @@ -575,7 +580,7 @@ function open( msg = (ctx_kw::RequestContext).cancel_message === nothing ? "request canceled" : (ctx_kw::RequestContext).cancel_message::String throw(CanceledError(msg)) end - wrapped = _wrap_client_transport_error(err, "request", Int64(0), elapsed_ns) + wrapped = err isa Exception ? _wrap_client_transport_error(err::Exception, "request", Int64(0), elapsed_ns) : err wrapped === err ? rethrow() : throw(wrapped) finally @try_ignore closewrite(stream) @@ -588,7 +593,7 @@ function open( msg = (ctx_kw::RequestContext).cancel_message === nothing ? "request canceled" : (ctx_kw::RequestContext).cancel_message::String throw(CanceledError(msg)) end - wrapped = _wrap_client_transport_error(err, "request", Int64(0), elapsed_ns) + wrapped = err isa Exception ? _wrap_client_transport_error(err::Exception, "request", Int64(0), elapsed_ns) : err wrapped === err ? rethrow() : throw(wrapped) end if status_exception && _status_throws(response) diff --git a/src/http_transport.jl b/src/http_transport.jl index 6fbca02ce..cabd84662 100644 --- a/src/http_transport.jl +++ b/src/http_transport.jl @@ -154,6 +154,7 @@ mutable struct Conn request_buf::IOBuffer reused::Bool @atomic closed::Bool + @atomic slot_released::Bool last_used_ns::Int64 end @@ -612,10 +613,12 @@ function _release_conn_slot_locked!(transport::Transport, key::String)::Union{No end function _close_owned_conn!(transport::Transport, conn::Conn) - _close_conn!(conn) || return nothing + _close_conn!(conn) waiter = nothing lock(transport.lock) try + (@atomic :acquire conn.slot_released) && return nothing + @atomic :release conn.slot_released = true waiter = _release_conn_slot_locked!(transport, conn.key) finally unlock(transport.lock) @@ -1002,7 +1005,7 @@ function _new_conn_tcp!( connect_deadline_ns::Int64=Int64(0), )::Conn tcp = _new_tcp_conn!(plan, address, host_resolver, connect_deadline_ns) - return Conn(plan.pool_key, plan.first_hop_address, false, tcp, nothing, _ConnReader(tcp), IOBuffer(), false, false, time_ns()) + return Conn(plan.pool_key, plan.first_hop_address, false, tcp, nothing, _ConnReader(tcp), IOBuffer(), false, false, false, time_ns()) end function _new_conn_tls!( @@ -1020,9 +1023,14 @@ function _new_conn_tls!( tls = TLS.client(tcp, cfg) connect_deadline_ns == 0 || TLS.set_deadline!(tls, connect_deadline_ns) TLS.handshake!(tls) - return Conn(plan.pool_key, plan.first_hop_address, true, tcp, tls, _ConnReader(tls), IOBuffer(), false, false, time_ns()) - catch + return Conn(plan.pool_key, plan.first_hop_address, true, tcp, tls, _ConnReader(tls), IOBuffer(), false, false, false, time_ns()) + catch err @try_ignore TCP.close(tcp) + # Type TLS failures at the site where the phase is known: a TLSError + # here arose during connection setup, so surface it as a handshake + # failure. TLS failures on established connections reach the public + # boundary as `TLSTransportError` instead. + err isa TLS.TLSError && throw(TLSHandshakeError(err::TLS.TLSError)) rethrow() end end @@ -1049,6 +1057,40 @@ function _evict_expired_idle_locked!(transport::Transport, key::String, now_ns:: return stale end +function _dial_reserved_conn!( + transport::Transport, + plan::_ProxyPlan, + address::String, + secure::Bool, + server_name::Union{Nothing,String}, + host_resolver::_TransportHostResolver, + connect_deadline_ns::Int64, + tls_handshake_timeout_ns::Int64, +)::Conn + try + return _new_conn!( + transport, + plan, + address, + secure, + server_name, + host_resolver, + connect_deadline_ns, + tls_handshake_timeout_ns, + ) + catch + waiter_to_notify = nothing + lock(transport.lock) + try + waiter_to_notify = _release_conn_slot_locked!(transport, plan.pool_key) + finally + unlock(transport.lock) + end + waiter_to_notify === nothing || _notify_waiter!(waiter_to_notify) + rethrow() + end +end + function _acquire_conn!( transport::Transport, plan::_ProxyPlan, @@ -1059,6 +1101,8 @@ function _acquire_conn!( host_resolver::_TransportHostResolver=transport.host_resolver, connect_deadline_ns::Int64=Int64(0), tls_handshake_timeout_ns::Int64=Int64(0), + ; + force_fresh::Bool=false, )::Conn _transport_closed(transport) && throw(ProtocolError("transport is closed")) waiter = nothing @@ -1095,13 +1139,16 @@ function _acquire_conn!( finally unlock(transport.lock) end - isempty(stale) || (_close_owned_conns!(transport, stale); continue) - if conn !== nothing - return conn::Conn + if !isempty(stale) + _close_owned_conns!(transport, stale) + conn === nothing && continue end - if should_dial - try - return _new_conn!( + if conn !== nothing + if force_fresh + # Transfer this connection's already-counted pool slot to the + # replacement dial. This keeps max_conns_per_host exact. + _close_conn!(conn::Conn) + return _dial_reserved_conn!( transport, plan, address, @@ -1111,45 +1158,51 @@ function _acquire_conn!( connect_deadline_ns, tls_handshake_timeout_ns, ) - catch err - waiter_to_notify = nothing - lock(transport.lock) - try - waiter_to_notify = _release_conn_slot_locked!(transport, plan.pool_key) - finally - unlock(transport.lock) - end - waiter_to_notify === nothing || _notify_waiter!(waiter_to_notify) - rethrow(err) end + return conn::Conn + end + if should_dial + return _dial_reserved_conn!( + transport, + plan, + address, + secure, + server_name, + host_resolver, + connect_deadline_ns, + tls_handshake_timeout_ns, + ) end result = _wait_for_conn!(transport, waiter::_ConnWaiter, acquire_deadline_ns) if result === :dial - try - return _new_conn!( - transport, - plan, - address, - secure, - server_name, - host_resolver, - connect_deadline_ns, - tls_handshake_timeout_ns, - ) - catch err - waiter_to_notify = nothing - lock(transport.lock) - try - waiter_to_notify = _release_conn_slot_locked!(transport, plan.pool_key) - finally - unlock(transport.lock) - end - waiter_to_notify === nothing || _notify_waiter!(waiter_to_notify) - rethrow(err) - end + return _dial_reserved_conn!( + transport, + plan, + address, + secure, + server_name, + host_resolver, + connect_deadline_ns, + tls_handshake_timeout_ns, + ) end conn = result::Conn conn.reused = true + if force_fresh + # A capped acquire can receive a direct handoff. Replace it under + # the same reserved slot instead of returning a reused connection. + _close_conn!(conn) + return _dial_reserved_conn!( + transport, + plan, + address, + secure, + server_name, + host_resolver, + connect_deadline_ns, + tls_handshake_timeout_ns, + ) + end return conn end end @@ -1473,26 +1526,38 @@ end return true end -@inline function _retryable_method(method::String)::Bool - return method == "GET" || method == "HEAD" || method == "OPTIONS" || method == "TRACE" || method == "QUERY" -end - @inline function _retryable_request(request::Request)::Bool - _retryable_method(request.method) || return false - request.content_length == 0 && return true - request.body isa EmptyBody && return true - request.body isa BytesBody && return true - return false + return _retryable_request_method(request.method) && _retryable_request_body(request) end +const _retryable_method = _retryable_request_method + @inline function _retryable_reused_conn_error(err)::Bool - err isa EOFError && return true - err isa SystemError && return true - err isa ParseError && return true - err isa IOPoll.NetClosingError && return true - err isa IOPoll.NotPollableError && return true - err isa IOPoll.DeadlineExceededError && return false - return false + # Iterative cause-unwrapping (not recursion) keeps this resolvable for + # trimmed static compilation. + current = err + while true + current isa EOFError && return true + current isa SystemError && return true + current isa ParseError && return true + current isa IOPoll.NetClosingError && return true + current isa IOPoll.NotPollableError && return true + current isa IOPoll.DeadlineExceededError && return false + if current isa TLS.TLSError + # A reused TLS connection whose peer vanished surfaces reads and + # writes as TLSError wrapping the underlying transport failure + # (e.g. an RST as `SystemError`, or a truncated stream carrying + # only the shared truncation message). Classify by the cause so + # dead reused connections are retried here instead of consuming + # the caller's retry budget (#1353). + (current::TLS.TLSError).message == _TLS_TRUNCATED_STREAM_MESSAGE && return true + cause = (current::TLS.TLSError).cause + cause === nothing && return false + current = cause::Exception + continue + end + return false + end end @inline function _request_upload_abort_error(err)::Bool @@ -1630,12 +1695,13 @@ function body_read!(body::H1Body, dst::Vector{UInt8})::Int return n end error("unexpected H1 body kind") - catch + catch err body.reusable = false body.done = true @atomic :release body.closed = true _release_h1_body!(body) - rethrow() + wrapped = err isa Exception ? _wrap_tls_transport_error(err::Exception) : err + wrapped === err ? rethrow() : throw(wrapped) end end @@ -1730,6 +1796,15 @@ Execute one HTTP/1 request/response exchange through `transport`. This is the low-level HTTP/1 path used by the higher-level client APIs. It returns an `_IncomingResponse` before the public `Response` conversion step. +When a replayable idempotent request fails on a *reused* pooled connection with +an error that marks the connection dead, the exchange is retried on another +connection. Pooled connections can die in correlated batches (a peer or +middlebox discarding every connection parked during the same idle window), so +the retry repeats while failures keep landing on reused connections. After at +most `max_idle_per_host` reused acquisitions, the last attempt replaces any +idle or handed-off connection with a fresh dial (#1353). A failure on that +fresh connection propagates. + Throws parser, protocol, transport, TLS, and timeout exceptions depending on where the exchange fails. """ @@ -1741,161 +1816,163 @@ function _roundtrip_incoming!( server_name::Union{Nothing,AbstractString}=nothing, proxy_config::ProxyConfig=transport.proxy, attempt::Int=1, + retry_template::Union{Nothing,Request}=nothing, ) - request_deadline = _request_deadline_ns(request) - retry_template = attempt == 1 && _retryable_request(request) ? _copy_request(request) : nothing - plan = _proxy_plan(proxy_config, secure, String(address)) - connect_host_resolver = _request_connect_host_resolver(transport.host_resolver, request) - connect_deadline_ns = _request_connect_phase_deadline_ns(transport.host_resolver, request) - tls_handshake_timeout_ns = _request_connect_phase_timeout_ns(transport.host_resolver, request) - request_ctx = get_request_context(request) - canceled(request_ctx) && throw(CanceledError(request_ctx.cancel_message === nothing ? "request canceled" : request_ctx.cancel_message::String)) - conn = _acquire_conn!( - transport, - plan, - String(address), - secure, - server_name === nothing ? nothing : String(server_name), - request_deadline, - connect_host_resolver, - connect_deadline_ns, - tls_handshake_timeout_ns, - ) - was_reused = conn.reused - cancel_cb = let conn = conn - () -> begin - try - _set_conn_read_deadline!(conn, Int64(1)) - _set_conn_write_deadline!(conn, Int64(1)) - catch - end - try - _close_conn!(conn) - catch - end - end + if retry_template === nothing && attempt == 1 && _retryable_request(request) + retry_template = _copy_request(request) end - _on_cancel!(request_ctx, cancel_cb) - try + while true + attempt_request = request + request_deadline = _request_deadline_ns(attempt_request) + plan = _proxy_plan(proxy_config, secure, String(address)) + connect_host_resolver = _request_connect_host_resolver(transport.host_resolver, attempt_request) + connect_deadline_ns = _request_connect_phase_deadline_ns(transport.host_resolver, attempt_request) + tls_handshake_timeout_ns = _request_connect_phase_timeout_ns(transport.host_resolver, attempt_request) + request_ctx = get_request_context(attempt_request) canceled(request_ctx) && throw(CanceledError(request_ctx.cancel_message === nothing ? "request canceled" : request_ctx.cancel_message::String)) - _apply_conn_deadline!(conn, request_deadline) - request_io = _reset_request_buffer!(conn) - stream = _conn_stream(conn) - deadline_stream = _RequestDeadlineWriteIO(stream, conn, request) - has_request_body = _request_has_body(request) - write_state = has_request_body ? _RequestWriteState(_request_expects_continue(request)) : nothing - writer_err = Base.RefValue{Union{Nothing,Exception}}(nothing) - writer_task = nothing - if has_request_body - writer_task = Threads.@spawn begin + conn = _acquire_conn!( + transport, + plan, + String(address), + secure, + server_name === nothing ? nothing : String(server_name), + request_deadline, + connect_host_resolver, + connect_deadline_ns, + tls_handshake_timeout_ns; + force_fresh = attempt > transport.max_idle_per_host, + ) + was_reused = conn.reused + cancel_cb = let conn = conn + () -> begin try - _write_request_streaming!( - request_io, - deadline_stream, - request, - plan, - write_state, - request_deadline, - ) - catch err - writer_err[] = err isa Exception ? err : ProtocolError("request upload failed") - _request_write_allows_close(write_state) || return nothing - @try_ignore begin - _close_conn!(conn) + _set_conn_read_deadline!(conn, Int64(1)) + _set_conn_write_deadline!(conn, Int64(1)) + catch + end + try + _close_conn!(conn) + catch + end + end + end + _on_cancel!(request_ctx, cancel_cb) + try + canceled(request_ctx) && throw(CanceledError(request_ctx.cancel_message === nothing ? "request canceled" : request_ctx.cancel_message::String)) + _apply_conn_deadline!(conn, request_deadline) + request_io = _reset_request_buffer!(conn) + stream = _conn_stream(conn) + deadline_stream = _RequestDeadlineWriteIO(stream, conn, attempt_request) + has_request_body = _request_has_body(attempt_request) + write_state = has_request_body ? _RequestWriteState(_request_expects_continue(attempt_request)) : nothing + writer_err = Base.RefValue{Union{Nothing,Exception}}(nothing) + writer_task = nothing + if has_request_body + writer_task = Threads.@spawn let send_request = attempt_request + try + _write_request_streaming!( + request_io, + deadline_stream, + send_request, + plan, + write_state, + request_deadline, + ) + catch err + writer_err[] = err isa Exception ? err : ProtocolError("request upload failed") + _request_write_allows_close(write_state) || return nothing + @try_ignore begin + _close_conn!(conn) + end + finally + @try_ignore begin + body_close!(send_request.body) + end + _request_write_mark_done!(write_state) end + return nothing + end + if request_deadline == 0 + while !_request_write_head_written_or_done(write_state) + IOPoll.timedwait(() -> _request_write_head_written_or_done(write_state), 0.05; pollint=0.001) + end + else + status = IOPoll.timedwait(() -> _request_write_head_written_or_done(write_state), max((request_deadline - Int64(time_ns())) / 1.0e9, 0.0); pollint=0.001) + status == :timed_out && throw(IOPoll.DeadlineExceededError()) + end + if _request_write_done(write_state) + wait(writer_task::Task) + err = writer_err[] + err === nothing || throw(err::Exception) + end + else + try + _write_request_streaming!(request_io, deadline_stream, attempt_request, plan) finally @try_ignore begin - body_close!(request.body) + body_close!(attempt_request.body) end - _request_write_mark_done!(write_state) end - return nothing end - if request_deadline == 0 - while !_request_write_head_written_or_done(write_state) - IOPoll.timedwait(() -> _request_write_head_written_or_done(write_state), 0.05; pollint=0.001) + reader = conn.reader + _set_conn_read_deadline!(conn, _request_response_header_deadline_ns(attempt_request)) + raw_response = _read_transport_incoming_response(reader, transport, conn, attempt_request) + # HTTP/1 informational responses are consumed internally so callers + # observe the final non-1xx response. + while (raw_response.head.status >= 100 && raw_response.head.status < 200) && raw_response.head.status != 101 + if _request_write_should_wait_for_continue(write_state) && raw_response.head.status == 100 + _request_write_mark_continue_allowed!(write_state::_RequestWriteState) end - else - status = IOPoll.timedwait(() -> _request_write_head_written_or_done(write_state), max((request_deadline - Int64(time_ns())) / 1.0e9, 0.0); pollint=0.001) - status == :timed_out && throw(IOPoll.DeadlineExceededError()) - end - if _request_write_done(write_state) - wait(writer_task::Task) - err = writer_err[] - err === nothing || throw(err::Exception) - end - else - try - _write_request_streaming!(request_io, deadline_stream, request, plan) - finally @try_ignore begin - body_close!(request.body) + body_close!(raw_response.rawbody) end + raw_response = _read_transport_incoming_response(reader, transport, conn, attempt_request) end - end - reader = conn.reader - _set_conn_read_deadline!(conn, _request_response_header_deadline_ns(request)) - raw_response = _read_transport_incoming_response(reader, transport, conn, request) - # HTTP/1 informational responses are consumed internally so callers - # observe the final non-1xx response. - while (raw_response.head.status >= 100 && raw_response.head.status < 200) && raw_response.head.status != 101 - if _request_write_should_wait_for_continue(write_state) && raw_response.head.status == 100 - _request_write_mark_continue_allowed!(write_state::_RequestWriteState) + _set_conn_read_deadline!(conn, request_deadline) + early_final = false + if _request_write_should_wait_for_continue(write_state) && _request_write_continue_state(write_state::_RequestWriteState) == _REQUEST_WRITE_CONTINUE_PENDING + _request_write_mark_continue_suppressed!(write_state::_RequestWriteState) + early_final = true end - @try_ignore begin - body_close!(raw_response.rawbody) + if has_request_body && !_request_write_done(write_state) + early_final = true + _request_write_request_stop!(write_state::_RequestWriteState) + _request_write_disallow_close!(write_state::_RequestWriteState) + _set_conn_write_deadline!(conn, Int64(time_ns())) end - raw_response = _read_transport_incoming_response(reader, transport, conn, request) - end - _set_conn_read_deadline!(conn, request_deadline) - early_final = false - if _request_write_should_wait_for_continue(write_state) && _request_write_continue_state(write_state::_RequestWriteState) == _REQUEST_WRITE_CONTINUE_PENDING - _request_write_mark_continue_suppressed!(write_state::_RequestWriteState) - early_final = true - end - if has_request_body && !_request_write_done(write_state) - early_final = true - _request_write_request_stop!(write_state::_RequestWriteState) - _request_write_disallow_close!(write_state::_RequestWriteState) - _set_conn_write_deadline!(conn, Int64(time_ns())) - end - if has_request_body && !early_final - if request_deadline == 0 - wait(writer_task::Task) - else - status = IOPoll.timedwait(() -> istaskdone(writer_task::Task), max((request_deadline - Int64(time_ns())) / 1.0e9, 0.0); pollint=0.001) - status == :timed_out && throw(IOPoll.DeadlineExceededError()) - wait(writer_task::Task) + if has_request_body && !early_final + if request_deadline == 0 + wait(writer_task::Task) + else + status = IOPoll.timedwait(() -> istaskdone(writer_task::Task), max((request_deadline - Int64(time_ns())) / 1.0e9, 0.0); pollint=0.001) + status == :timed_out && throw(IOPoll.DeadlineExceededError()) + wait(writer_task::Task) + end end - end - if has_request_body && writer_task !== nothing && istaskdone(writer_task::Task) - err = writer_err[] - if err !== nothing && !(early_final && _request_upload_abort_error(err::Exception)) - throw(err::Exception) + if has_request_body && writer_task !== nothing && istaskdone(writer_task::Task) + err = writer_err[] + if err !== nothing && !(early_final && _request_upload_abort_error(err::Exception)) + throw(err::Exception) + end end + reusable = _response_reusable(raw_response, attempt_request) + early_final && (reusable = false) + body = _arm_h1_body!(raw_response.rawbody::H1Body, reusable, request_ctx, cancel_cb) + if _body_immediately_empty(body) + body_close!(body) + end + return _IncomingResponse(raw_response.head, body) + catch err + _remove_cancel_callback!(request_ctx, cancel_cb) + _close_owned_conn!(transport, conn) + if attempt <= transport.max_idle_per_host && was_reused && retry_template !== nothing && _retryable_reused_conn_error(err) + request = _copy_request(retry_template::Request) + attempt += 1 + continue + end + wrapped = err isa Exception ? _wrap_tls_transport_error(err::Exception) : err + wrapped === err ? rethrow() : throw(wrapped) end - reusable = _response_reusable(raw_response, request) - early_final && (reusable = false) - body = _arm_h1_body!(raw_response.rawbody::H1Body, reusable, request_ctx, cancel_cb) - if _body_immediately_empty(body) - body_close!(body) - end - return _IncomingResponse(raw_response.head, body) - catch err - _remove_cancel_callback!(request_ctx, cancel_cb) - _close_owned_conn!(transport, conn) - if attempt == 1 && was_reused && retry_template !== nothing && _retryable_reused_conn_error(err) - return _roundtrip_incoming!( - transport, - address, - _copy_request(retry_template::Request), - secure, - server_name, - proxy_config, - attempt + 1, - ) - end - rethrow(err) end end diff --git a/test/http2_client_tests.jl b/test/http2_client_tests.jl index 3a9607012..654dfe71b 100644 --- a/test/http2_client_tests.jl +++ b/test/http2_client_tests.jl @@ -11,7 +11,7 @@ const TL = Reseau.TLS const _TLS_CERT_PATH = joinpath(@__DIR__, "resources", "unittests.crt") const _TLS_KEY_PATH = joinpath(@__DIR__, "resources", "unittests.key") -function _write_all_h2_tcp!(conn::NC.Conn, bytes::Vector{UInt8})::Nothing +function _write_all_h2_tcp!(conn, bytes::Vector{UInt8})::Nothing total = 0 while total < length(bytes) n = write(conn, bytes[(total + 1):end]) @@ -21,7 +21,7 @@ function _write_all_h2_tcp!(conn::NC.Conn, bytes::Vector{UInt8})::Nothing return nothing end -function _read_exact_h2_tcp!(conn::NC.Conn, n::Int)::Vector{UInt8} +function _read_exact_h2_tcp!(conn, n::Int)::Vector{UInt8} out = Vector{UInt8}(undef, n) offset = 0 while offset < n @@ -34,7 +34,7 @@ function _read_exact_h2_tcp!(conn::NC.Conn, n::Int)::Vector{UInt8} return out end -function _write_frame_to_conn!(conn::NC.Conn, frame::HT.AbstractFrame) +function _write_frame_to_conn!(conn, frame::HT.AbstractFrame) io = IOBuffer() framer = io HT.write_frame!(framer, frame) @@ -113,6 +113,32 @@ function _read_all_h2_body(body::HT.AbstractBody)::Vector{UInt8} return out end +@testset "HTTP/2 client types TLS dial failures as handshake errors (#1353)" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + server_task = errormonitor(Threads.@spawn begin + conn = NC.accept(listener) + try + # Not a TLS server: answer the ClientHello with plaintext so the + # client's record parsing fails during the handshake. + _write_all_h2_tcp!(conn, collect(codeunits("HTTP/1.1 400 Bad Request\r\ncontent-length: 0\r\n\r\n"))) + finally + HTTP.@try_ignore NC.close(conn) + end + return nothing + end) + err = try + HT.connect_h2!(address; secure = true, tls_config = TL.Config(verify_peer = false, verify_hostname = true)) + nothing + catch e + e + end + @test err isa HT.TLSHandshakeError + @test (err::HT.TLSHandshakeError).cause isa TL.TLSError + _wait_task_h2!(server_task) + HTTP.@try_ignore NC.close(listener) +end + @testset "HTTP/2 client request header filtering and authority selection" begin headers = HT.Headers() HT.setheader(headers, "Connection", "close") @@ -436,6 +462,124 @@ end end end +@testset "HTTP/2 high-level client retries a read-loop connection failure" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + accept_count = Ref(0) + server_task = Threads.@spawn begin + for attempt in 1:2 + conn = NC.accept(listener) + accept_count[] += 1 + reader = HT._ConnReader(conn) + try + _ = _read_exact_h2_tcp!(conn, length(HT._H2_PREFACE)) + _ = HT.read_frame!(reader) + _write_frame_to_conn!(conn, HT.SettingsFrame(false, Pair{UInt16,UInt32}[])) + request_headers = _read_next_headers_frame!(reader) + if attempt == 1 + # The read loop wraps this established-connection EOF in a + # ProtocolError. The high-level retry policy must inspect + # that cause and create a second H2 connection. + continue + end + encoder = HT.Encoder() + encoded = HT.encode_header_block( + encoder, + HT.HeaderField[ + HT.HeaderField(":status", "200", false), + HT.HeaderField("content-length", "2", false), + ], + ) + _write_frame_to_conn!(conn, HT.HeadersFrame(request_headers.stream_id, false, true, encoded)) + _write_frame_to_conn!(conn, HT.DataFrame(request_headers.stream_id, true, collect(codeunits("ok")))) + finally + HTTP.@try_ignore NC.close(conn) + end + end + return nothing + end + client = HT.Client(cookiejar = nothing) + try + response = HT.get( + client, + "http://$(address)/retry-read-loop"; + protocol = :h2, + retries = 1, + retry_bucket = false, + ) + @test response.status == 200 + @test String(response.body) == "ok" + _wait_task_h2!(server_task) + @test accept_count[] == 2 + finally + close(client) + HTTP.@try_ignore NC.close(listener) + HTTP.@try_ignore wait(server_task) + end +end + +@testset "HTTP/2 public roundtrip wraps established TLS record truncation" begin + listener = TL.listen( + "tcp", + "127.0.0.1:0", + TL.Config( + verify_peer = false, + cert_file = _TLS_CERT_PATH, + key_file = _TLS_KEY_PATH, + alpn_protocols = ["h2"], + ); + backlog = 8, + ) + port = Int((TL.addr(listener)::NC.SocketAddrV4).port) + address = ND.join_host_port("localhost", port) + server_task = Threads.@spawn begin + tls_conn = nothing + try + tls_conn = TL.accept(listener) + TL.handshake!(tls_conn::TL.Conn) + reader = HT._ConnReader(tls_conn::TL.Conn) + _ = _read_exact_h2_tcp!(tls_conn::TL.Conn, length(HT._H2_PREFACE)) + _ = HT.read_frame!(reader) + _write_frame_to_conn!(tls_conn::TL.Conn, HT.SettingsFrame(false, Pair{UInt16,UInt32}[])) + flush(tls_conn::TL.Conn) + _ = _read_next_headers_frame!(reader) + tcp = TL.net_conn(tls_conn::TL.Conn)::NC.Conn + _write_all_h2_tcp!(tcp, UInt8[0x17, 0x03, 0x03, 0x00, 0x10, 0x00]) + HTTP.@try_ignore NC.close(tcp) + finally + tls_conn === nothing || HTTP.@try_ignore TL.close(tls_conn::TL.Conn) + end + return nothing + end + h2_conn = nothing + try + h2_conn = HT.connect_h2!( + address; + secure = true, + tls_config = TL.Config( + verify_peer = false, + verify_hostname = false, + server_name = "localhost", + alpn_protocols = ["h2"], + ), + ) + request = HT.Request("GET", "/truncated"; host = address, body = HT.EmptyBody(), content_length = 0) + err = try + HT.h2_roundtrip!(h2_conn::HT.H2Connection, request) + nothing + catch caught + caught + end + @test err isa HT.TLSTransportError + @test (err::HT.TLSTransportError).cause isa TL.TLSError + _wait_task_h2!(server_task) + finally + h2_conn === nothing || close(h2_conn::HT.H2Connection) + HTTP.@try_ignore TL.close(listener) + HTTP.@try_ignore wait(server_task) + end +end + @testset "HTTP/2 client requires initial SETTINGS before other frames" begin listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) laddr = NC.addr(listener)::NC.SocketAddrV4 diff --git a/test/http_client_tests.jl b/test/http_client_tests.jl index 7186dd2c1..a52abb169 100644 --- a/test/http_client_tests.jl +++ b/test/http_client_tests.jl @@ -348,6 +348,19 @@ end end end +@testset "HTTP verbose prints RetrySkippedEvent lines (#1353)" begin + trace = HT._VerboseTrace(1) + req = HT.Request("GET", "/skip"; host = "example.com") + output = _capture_stdout_client() do + trace(HT.RetrySkippedEvent(req, "http://example.com/skip", 2, 0, :retry_bucket, HT.Response(503), nothing)) + trace(HT.RetrySkippedEvent(req, "http://example.com/skip", 1, 0, :deadline, nothing, EOFError())) + trace(HT.RetrySkippedEvent(req, "http://example.com/skip", 1, 0, :retry_bucket, nothing, nothing)) + end + @test occursin("[http] retry of attempt 2 skipped (retry_bucket) after status 503", output) + @test occursin("[http] retry of attempt 1 skipped (deadline) after", output) + @test occursin("skipped (retry_bucket) after failure", output) +end + @testset "HTTP client redirect rewrites method" begin listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) laddr = NC.addr(listener)::NC.SocketAddrV4 @@ -2028,6 +2041,25 @@ end end end +@testset "HTTP.open preserves non-Exception callback throws" begin + server = HT.serve!("127.0.0.1", 0; listenany = true) do _ + HT.Response(200, "ok") + end + try + err = try + HT.open(:GET, "http://127.0.0.1:$(HT.port(server))/nonexception") do _ + throw(:sentinel) + end + nothing + catch caught + caught + end + @test err === :sentinel + finally + HT.forceclose(server) + end +end + @testset "HTTP.open per-byte stream reads" begin if _http_windows_ci() @test_skip true diff --git a/test/http_client_transport_tests.jl b/test/http_client_transport_tests.jl index 09deabc90..a0f1243c9 100644 --- a/test/http_client_transport_tests.jl +++ b/test/http_client_transport_tests.jl @@ -6,6 +6,7 @@ const HT = HTTP const NC = Reseau.TCP const ND = Reseau.HostResolvers const IP = Reseau.IOPoll +const TL = Reseau.TLS if !isdefined(@__MODULE__, :_http_windows_ci) @inline function _http_windows_ci()::Bool @@ -82,6 +83,18 @@ function _wait_for_transport_waiter!(transport::HT.Transport, key::String)::Noth return nothing end +function _wait_for_transport_waiter_or_task!(transport::HT.Transport, key::String, task::Task)::Bool + lock(transport.lock) + try + while isempty(get(() -> HT._ConnWaiter[], transport.waiters, key)) && !istaskdone(task) + wait(transport.waiter_condition) + end + return !isempty(get(() -> HT._ConnWaiter[], transport.waiters, key)) + finally + unlock(transport.lock) + end +end + function _transport_debug(msg::AbstractString) _ = msg return nothing @@ -251,6 +264,14 @@ if _http_windows_ci() @testset "HTTP client transport retries idempotent request on stale reused conn" begin @test_skip true end + + @testset "HTTP client transport force-fresh acquire replaces reused conns" begin + @test_skip true + end + + @testset "HTTP client transport retries stale PUT and DELETE requests" begin + @test_skip true + end else @testset "HTTP client transport keep-alive reuse" begin _transport_debug("keep-alive reuse: start") @@ -1026,13 +1047,556 @@ end end end +@testset "HTTP client transport force-fresh acquire replaces reused conns" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + accepted = Channel{NC.Conn}(3) + server_task = Threads.@spawn begin + for _ in 1:3 + put!(accepted, NC.accept(listener)) + end + return nothing + end + transport = HT.Transport(max_idle_per_host = 1, max_idle_total = 1, max_conns_per_host = 1) + plan = HT._proxy_plan(transport.proxy, false, address) + server_conns = NC.Conn[] + first_conn = nothing + fresh_conn = nothing + replacement_conn = nothing + try + first_conn = HT._acquire_conn!(transport, plan, address, false, nothing) + push!(server_conns, take!(accepted)) + + acquire_task = Threads.@spawn try + HT._acquire_conn!( + transport, + plan, + address, + false, + nothing; + force_fresh = true, + ) + finally + lock(transport.lock) + try + notify(transport.waiter_condition) + finally + unlock(transport.lock) + end + end + queued = _wait_for_transport_waiter_or_task!(transport, plan.pool_key, acquire_task) + if !queued + unexpected_conn = fetch(acquire_task) + HT._close_owned_conn!(transport, unexpected_conn::HT.Conn) + end + @test queued + + # The normal handoff path offers `first_conn` to the waiter. A + # force-fresh acquire must transfer that counted slot to a fresh dial. + HT._put_idle_conn!(transport, first_conn::HT.Conn) + first_conn = nothing + fresh_conn = fetch(acquire_task) + @test !(fresh_conn::HT.Conn).reused + @test lock(transport.lock) do + HT._conn_slots_locked(transport, plan.pool_key) == 1 && isempty(transport.waiters) + end + push!(server_conns, take!(accepted)) + @test length(server_conns) == 2 + + # Exercise the other acquisition branch. This time the force-fresh + # caller finds a reused connection already parked in the idle pool. + HT._put_idle_conn!(transport, fresh_conn::HT.Conn) + fresh_conn = nothing + replacement_conn = HT._acquire_conn!( + transport, + plan, + address, + false, + nothing, + Int64(1); + force_fresh = true, + ) + @test !(replacement_conn::HT.Conn).reused + @test lock(transport.lock) do + HT._conn_slots_locked(transport, plan.pool_key) == 1 && isempty(transport.waiters) + end + push!(server_conns, take!(accepted)) + @test length(server_conns) == 3 + _wait_task!(server_task) + finally + first_conn === nothing || HT._close_owned_conn!(transport, first_conn::HT.Conn) + fresh_conn === nothing || HT._close_owned_conn!(transport, fresh_conn::HT.Conn) + replacement_conn === nothing || HT._close_owned_conn!(transport, replacement_conn::HT.Conn) + for conn in server_conns + HTTP.@try_ignore NC.close(conn) + end + close(transport) + HTTP.@try_ignore NC.close(listener) + HTTP.@try_ignore wait(server_task) + end +end + +@testset "HTTP client transport keeps a fresh idle conn while evicting an expired peer" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + accepted = Channel{NC.Conn}(2) + server_task = Threads.@spawn begin + for _ in 1:2 + put!(accepted, NC.accept(listener)) + end + return nothing + end + transport = HT.Transport( + max_idle_per_host = 2, + max_idle_total = 2, + max_conns_per_host = 2, + idle_timeout_ns = 1, + ) + plan = HT._proxy_plan(transport.proxy, false, address) + stale_conn = nothing + fresh_conn = nothing + acquired_conn = nothing + server_conns = NC.Conn[] + try + stale_conn = HT._acquire_conn!(transport, plan, address, false, nothing) + push!(server_conns, take!(accepted)) + fresh_conn = HT._acquire_conn!(transport, plan, address, false, nothing) + push!(server_conns, take!(accepted)) + _wait_task!(server_task) + + HT._put_idle_conn!(transport, stale_conn::HT.Conn) + HT._put_idle_conn!(transport, fresh_conn::HT.Conn) + (stale_conn::HT.Conn).last_used_ns = 0 + (fresh_conn::HT.Conn).last_used_ns = typemax(Int64) + + acquired_conn = HT._acquire_conn!(transport, plan, address, false, nothing) + @test acquired_conn === fresh_conn + @test HT._conn_closed(stale_conn::HT.Conn) + @test (@atomic transport.idle_total) == 0 + @test lock(transport.lock) do + HT._conn_slots_locked(transport, plan.pool_key) == 1 + end + + HT._close_owned_conn!(transport, acquired_conn::HT.Conn) + acquired_conn = nothing + fresh_conn = nothing + @test lock(transport.lock) do + HT._conn_slots_locked(transport, plan.pool_key) == 0 + end + finally + acquired_conn === nothing || HT._close_owned_conn!(transport, acquired_conn::HT.Conn) + fresh_conn === nothing || HT._close_owned_conn!(transport, fresh_conn::HT.Conn) + stale_conn === nothing || HT._close_owned_conn!(transport, stale_conn::HT.Conn) + close(transport) + for conn in server_conns + HTTP.@try_ignore NC.close(conn) + end + HTTP.@try_ignore NC.close(listener) + HTTP.@try_ignore wait(server_task) + end +end + +@testset "HTTP client transport releases a slot after an earlier raw close" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + accepted = Channel{NC.Conn}(1) + server_task = Threads.@spawn put!(accepted, NC.accept(listener)) + transport = HT.Transport(max_idle_per_host = 1, max_idle_total = 1, max_conns_per_host = 1) + plan = HT._proxy_plan(transport.proxy, false, address) + conn = nothing + server_conn = nothing + try + conn = HT._acquire_conn!(transport, plan, address, false, nothing) + server_conn = take!(accepted) + _wait_task!(server_task) + @test HT._close_conn!(conn::HT.Conn) + HT._close_owned_conn!(transport, conn::HT.Conn) + conn = nothing + @test lock(transport.lock) do + HT._conn_slots_locked(transport, plan.pool_key) == 0 + end + finally + conn === nothing || HT._close_owned_conn!(transport, conn::HT.Conn) + close(transport) + server_conn === nothing || HTTP.@try_ignore NC.close(server_conn::NC.Conn) + HTTP.@try_ignore NC.close(listener) + HTTP.@try_ignore wait(server_task) + end +end + +@testset "HTTP client transport retries stale PUT and DELETE requests" begin + for (method, payload) in (("PUT", "payload"), ("DELETE", "")) + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + first_conn_closed = Channel{Nothing}(1) + seen = Tuple{String,String}[] + server_task = Threads.@spawn begin + conn1 = NC.accept(listener) + try + warmup = HT.read_request(HT._ConnReader(conn1)) + push!(seen, (warmup.method, String(_read_all_transport_body_bytes(warmup.body)))) + _write_response_to_conn!(conn1, warmup; body_text = "warmup") + finally + HTTP.@try_ignore NC.close(conn1) + put!(first_conn_closed, nothing) + end + + conn2 = NC.accept(listener) + try + retried = HT.read_request(HT._ConnReader(conn2)) + push!(seen, (retried.method, String(_read_all_transport_body_bytes(retried.body)))) + _write_response_to_conn!(conn2, retried; body_text = "recovered", close_conn = true) + finally + HTTP.@try_ignore NC.close(conn2) + end + return nothing + end + transport = HT.Transport(max_idle_per_host = 1, max_idle_total = 1) + try + warmup = HT.Request("GET", "/warmup"; host = address, body = HT.EmptyBody(), content_length = 0) + warmup_response = HT.roundtrip!(transport, address, warmup) + @test String(_read_all_transport_body_bytes(warmup_response.body)) == "warmup" + take!(first_conn_closed) + + body = isempty(payload) ? HT.EmptyBody() : HT.BytesBody(collect(codeunits(payload))) + request = HT.Request(method, "/retry"; host = address, body = body, content_length = ncodeunits(payload)) + response = HT.roundtrip!(transport, address, request) + @test String(_read_all_transport_body_bytes(response.body)) == "recovered" + _wait_task!(server_task) + @test seen == [("GET", ""), (method, payload)] + finally + close(transport) + HTTP.@try_ignore NC.close(listener) + HTTP.@try_ignore wait(server_task) + end + end +end + end @testset "HTTP client transport treats not-pollable reused errors as retryable" begin @test HT._retryable_method("QUERY") + @test HT._retryable_request(HT.Request("PUT", "/"; body = HT.BytesBody(UInt8[0x78]), content_length = 1)) + @test HT._retryable_request(HT.Request("DELETE", "/"; body = HT.EmptyBody(), content_length = 0)) @test HT._retryable_reused_conn_error(Reseau.IOPoll.NotPollableError()) end +@testset "HTTP client transport types TLS dial failures as handshake errors (#1353)" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + server_task = errormonitor(Threads.@spawn begin + conn = NC.accept(listener) + try + # Not a TLS server: answer the ClientHello with plaintext so the + # client's record parsing fails during the handshake. + _write_all_tcp!(conn, collect(codeunits("HTTP/1.1 400 Bad Request\r\ncontent-length: 0\r\n\r\n"))) + finally + HTTP.@try_ignore NC.close(conn) + end + return nothing + end) + transport = HT.Transport(tls_config = Reseau.TLS.Config(verify_peer = false, verify_hostname = true)) + try + req = HT.Request("GET", "/handshake"; host = address, body = HT.EmptyBody(), content_length = 0) + err = try + HT.roundtrip!(transport, address, req; secure = true, server_name = "localhost") + nothing + catch e + e + end + @test err isa HT.TLSHandshakeError + @test (err::HT.TLSHandshakeError).cause isa Reseau.TLS.TLSError + _wait_task!(server_task) + finally + close(transport) + HTTP.@try_ignore NC.close(listener) + end +end + +@testset "HTTP client transport classifies TLS-wrapped reused errors by cause (#1353)" begin + # A dead reused TLS connection surfaces reads/writes as TLSError wrapping + # the underlying transport failure (e.g. an RST as SystemError); classify + # by the cause so the reused-connection retry engages. + @test HT._retryable_reused_conn_error( + Reseau.TLS.TLSError("read", Int32(0), "unexpected TLS failure", SystemError("read", 0)), + ) + @test HT._retryable_reused_conn_error( + Reseau.TLS.TLSError("read", Int32(0), "unexpected EOF", EOFError()), + ) + # Reproduce Reseau's private mid-record EOF sentinel. Production code must + # classify the public TLSError message instead of naming this private type. + unexpected_eof_cause = ErrorException("opaque TLS record EOF cause") + @test HT._retryable_reused_conn_error( + Reseau.TLS.TLSError("read", Int32(0), "unexpected EOF", unexpected_eof_cause), + ) + # Causeless TLS protocol failures and deadline expiries are not dead-conn + # signatures. + @test !HT._retryable_reused_conn_error( + Reseau.TLS.TLSError("read", Int32(0), "bad record mac", nothing), + ) + @test !HT._retryable_reused_conn_error( + Reseau.TLS.TLSError("read", Int32(0), "i/o timeout", Reseau.IOPoll.DeadlineExceededError()), + ) +end + +@testset "HTTP public client APIs wrap established TLS record truncation" begin + cert_file = joinpath(@__DIR__, "resources", "unittests.crt") + key_file = joinpath(@__DIR__, "resources", "unittests.key") + listener = TL.listen( + "tcp", + "127.0.0.1:0", + TL.Config( + verify_peer = false, + cert_file = cert_file, + key_file = key_file, + ); + backlog = 8, + ) + port = Int((TL.addr(listener)::NC.SocketAddrV4).port) + address = ND.join_host_port("localhost", port) + partial_record = UInt8[0x17, 0x03, 0x03, 0x00, 0x10, 0x00] + server_task = Threads.@spawn begin + for _ in 1:4 + tls_conn = nothing + try + tls_conn = TL.accept(listener) + TL.handshake!(tls_conn::TL.Conn) + request = HT.read_request(HT._ConnReader(tls_conn::TL.Conn)) + if startswith(request.target, "/body") + write(tls_conn::TL.Conn, "HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\n") + flush(tls_conn::TL.Conn) + end + tcp = TL.net_conn(tls_conn::TL.Conn)::NC.Conn + _write_all_tcp!(tcp, partial_record) + HTTP.@try_ignore NC.close(tcp) + finally + tls_conn === nothing || HTTP.@try_ignore TL.close(tls_conn::TL.Conn) + end + end + return nothing + end + transport = HT.Transport( + tls_config = TL.Config(verify_peer = false, verify_hostname = false), + max_idle_per_host = 1, + max_idle_total = 1, + ) + client = HT.Client(transport = transport, cookiejar = nothing, prefer_http2 = false) + try + for api in (:roundtrip, :open), phase in (:head, :body) + target = "/$(phase)-$(api)" + err = if api === :roundtrip + response = nothing + try + request = HT.Request("GET", target; host = address, body = HT.EmptyBody(), content_length = 0) + response = HT.roundtrip!(transport, address, request; secure = true, server_name = "localhost") + phase === :body && HT.body_read!(response.body, Vector{UInt8}(undef, 5)) + nothing + catch caught + caught + finally + response === nothing || HTTP.@try_ignore HT.body_close!(response.body) + end + else + stream = HT.open( + :GET, + "https://$(address)$(target)"; + client = client, + protocol = :h1, + retry = false, + ) + try + HT.startread(stream) + phase === :body && read(stream) + nothing + catch caught + caught + finally + close(stream) + end + end + @test err isa HT.TLSTransportError + @test (err::HT.TLSTransportError).cause isa TL.TLSError + end + _wait_task!(server_task) + finally + close(client) + HTTP.@try_ignore TL.close(listener) + HTTP.@try_ignore wait(server_task) + end +end + +@testset "HTTP client transport survives a fully poisoned idle pool (#1353)" begin + # Pooled connections can die in correlated batches (dialed together, then + # discarded together by the peer while parked). The reused-connection retry + # must keep retrying while failures land on reused connections — through + # every dead pooled connection — until it reaches a fresh dial, instead of + # giving up after a single retry. + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + laddr = NC.addr(listener)::NC.SocketAddrV4 + address = ND.join_host_port("127.0.0.1", Int(laddr.port)) + accept_count = Ref(0) + paths = String[] + both_warmups_read = Channel{Nothing}(2) + warmup_conns_closed = Channel{Nothing}(1) + server_task = errormonitor(Threads.@spawn begin + # Hold both warmup responses until both requests have arrived so the + # client provably opens two separate connections. + conn1 = NC.accept(listener) + accept_count[] += 1 + req1 = HT.read_request(HT._ConnReader(conn1)) + push!(paths, req1.target) + put!(both_warmups_read, nothing) + conn2 = NC.accept(listener) + accept_count[] += 1 + req2 = HT.read_request(HT._ConnReader(conn2)) + push!(paths, req2.target) + put!(both_warmups_read, nothing) + # Keep-alive responses so the client parks both connections. + _write_response_to_conn!(conn1, req1; body_text = "warmup1") + _write_response_to_conn!(conn2, req2; body_text = "warmup2") + # Correlated death: discard both parked connections at once. + take!(warmup_conns_closed) + HTTP.@try_ignore NC.close(conn1) + HTTP.@try_ignore NC.close(conn2) + conn3 = NC.accept(listener) + accept_count[] += 1 + try + req3 = HT.read_request(HT._ConnReader(conn3)) + push!(paths, req3.target) + _write_response_to_conn!(conn3, req3; body_text = "recovered", close_conn = true) + finally + HTTP.@try_ignore NC.close(conn3) + end + return nothing + end) + transport = HT.Transport(max_idle_per_host = 4, max_idle_total = 4) + try + warmup1 = errormonitor(Threads.@spawn begin + req = HT.Request("GET", "/warmup1"; host = address, body = HT.EmptyBody(), content_length = 0) + res = HT.roundtrip!(transport, address, req) + String(_read_all_transport_body_bytes(res.body)) + end) + warmup2 = errormonitor(Threads.@spawn begin + take!(both_warmups_read) # conn1's request is in: this dials conn2 + req = HT.Request("GET", "/warmup2"; host = address, body = HT.EmptyBody(), content_length = 0) + res = HT.roundtrip!(transport, address, req) + String(_read_all_transport_body_bytes(res.body)) + end) + @test fetch(warmup1) == "warmup1" + @test fetch(warmup2) == "warmup2" + take!(both_warmups_read) + put!(warmup_conns_closed, nothing) + # Both parked connections are now dead; the FINs may take a moment to + # be delivered, but reads observe them either way once we try to reuse. + req = HT.Request("GET", "/poisoned"; host = address, body = HT.EmptyBody(), content_length = 0) + res = HT.roundtrip!(transport, address, req) + @test res.status == 200 + @test String(_read_all_transport_body_bytes(res.body)) == "recovered" + _wait_task!(server_task) + @test accept_count[] == 3 + @test paths == ["/warmup1", "/warmup2", "/poisoned"] + finally + close(transport) + HTTP.@try_ignore NC.close(listener) + end +end + +if _http_windows_ci() + @testset "HTTP client transport forces a fresh final retry during concurrent handoff" begin + @test_skip true + end +else +@testset "HTTP client transport forces a fresh final retry during concurrent handoff" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + first_request_read = Channel{Nothing}(1) + blocker_parked = Channel{Nothing}(1) + accept_count = Ref(0) + paths = String[] + server_task = Threads.@spawn begin + conn_a = NC.accept(listener) + accept_count[] += 1 + conn_b = nothing + try + warmup = HT.read_request(HT._ConnReader(conn_a)) + push!(paths, warmup.target) + _read_all_transport_body_bytes(warmup.body) + put!(first_request_read, nothing) + + conn_b = NC.accept(listener) + accept_count[] += 1 + blocker = HT.read_request(HT._ConnReader(conn_b::NC.Conn)) + push!(paths, blocker.target) + _read_all_transport_body_bytes(blocker.body) + + # Return A first so it becomes the victim's reused connection. + _write_response_to_conn!(conn_a, warmup; body_text = "warmup") + victim = HT.read_request(HT._ConnReader(conn_a)) + push!(paths, victim.target) + _read_all_transport_body_bytes(victim.body) + + # Return B while the victim is blocked on A. After B is parked, + # close both connections before allowing the victim to retry. + _write_response_to_conn!(conn_b::NC.Conn, blocker; body_text = "blocker") + take!(blocker_parked) + HTTP.@try_ignore NC.close(conn_b::NC.Conn) + conn_b = nothing + HTTP.@try_ignore NC.close(conn_a) + + conn_c = NC.accept(listener) + accept_count[] += 1 + try + recovered = HT.read_request(HT._ConnReader(conn_c)) + push!(paths, recovered.target) + _read_all_transport_body_bytes(recovered.body) + _write_response_to_conn!(conn_c, recovered; body_text = "recovered", close_conn = true) + finally + HTTP.@try_ignore NC.close(conn_c) + end + finally + conn_b === nothing || HTTP.@try_ignore NC.close(conn_b::NC.Conn) + HTTP.@try_ignore NC.close(conn_a) + end + return nothing + end + transport = HT.Transport( + max_idle_per_host = 1, + max_idle_total = 2, + max_conns_per_host = 2, + ) + blocker_task = nothing + try + warmup_task = errormonitor(Threads.@spawn begin + request = HT.Request("GET", "/warmup"; host = address, body = HT.EmptyBody(), content_length = 0) + response = HT.roundtrip!(transport, address, request) + return String(_read_all_transport_body_bytes(response.body)) + end) + take!(first_request_read) + blocker_task = errormonitor(Threads.@spawn begin + request = HT.Request("GET", "/blocker"; host = address, body = HT.EmptyBody(), content_length = 0) + response = HT.roundtrip!(transport, address, request) + body = String(_read_all_transport_body_bytes(response.body)) + put!(blocker_parked, nothing) + return body + end) + @test fetch(warmup_task) == "warmup" + + victim = HT.Request("GET", "/victim"; host = address, body = HT.EmptyBody(), content_length = 0) + response = HT.roundtrip!(transport, address, victim) + @test String(_read_all_transport_body_bytes(response.body)) == "recovered" + @test fetch(blocker_task::Task) == "blocker" + _wait_task!(server_task) + @test accept_count[] == 3 + @test paths == ["/warmup", "/blocker", "/victim", "/victim"] + finally + isready(blocker_parked) || put!(blocker_parked, nothing) + close(transport) + HTTP.@try_ignore NC.close(listener) + HTTP.@try_ignore wait(server_task) + end +end +end + @testset "close_idle_connections! clears the default and per-client pools" begin server = HTTP.serve!("127.0.0.1", 0) do req return HTTP.Response(200, "ok") diff --git a/test/http_core_tests.jl b/test/http_core_tests.jl index e643dccdd..360ea74df 100644 --- a/test/http_core_tests.jl +++ b/test/http_core_tests.jl @@ -306,6 +306,7 @@ end @test HT.ConnectError <: HT.HTTPError @test HT.DNSError <: HT.HTTPError @test HT.TLSHandshakeError <: HT.HTTPError + @test HT.TLSTransportError <: HT.HTTPError @test HT.AddressInUseError <: HT.HTTPError addr = "127.0.0.1:1" @@ -314,12 +315,24 @@ end @test HT.ConnectError(addr, cause).cause === cause @test HT.DNSError("host.invalid", cause).hostname == "host.invalid" @test HT.TLSHandshakeError(cause).cause === cause + @test HT.TLSTransportError(cause).cause === cause @test HT.AddressInUseError(addr).address == addr - tls_error = Reseau.TLS.TLSError("handshake", Int32(-1), "boom", nothing) + # A bare TLSError at the boundary arose on an established connection + # (handshake failures are typed TLSHandshakeError at the dial sites), so + # it wraps as TLSTransportError, not TLSHandshakeError (#1353). + tls_error = Reseau.TLS.TLSError("read", Int32(-1), "boom", nothing) wrapped_tls_error = HT._wrap_client_transport_error(tls_error) - @test wrapped_tls_error isa HT.TLSHandshakeError - @test (wrapped_tls_error::HT.TLSHandshakeError).cause === tls_error + @test wrapped_tls_error isa HT.TLSTransportError + @test (wrapped_tls_error::HT.TLSTransportError).cause === tls_error + @test HT._wrap_client_transport_error(wrapped_tls_error) === wrapped_tls_error + @test occursin("http tls transport error", sprint(showerror, wrapped_tls_error)) + @test occursin("http tls handshake error", sprint(showerror, HT.TLSHandshakeError(tls_error))) + + # An OpError-wrapped TLSError comes from the dial path and stays a + # handshake error. + op_error = Reseau.HostResolvers.OpError("dial", "tcp", nothing, nothing, tls_error) + @test HT._wrap_client_transport_error(op_error) isa HT.TLSHandshakeError bytes = UInt8[0x41] @test HT._response_body_arg(bytes) === bytes diff --git a/test/http_retry_tests.jl b/test/http_retry_tests.jl index 8582efedf..6362493ec 100644 --- a/test/http_retry_tests.jl +++ b/test/http_retry_tests.jl @@ -112,6 +112,17 @@ end @test_throws ArgumentError HT.RetryBucket(capacity = 0) @test_throws ArgumentError HT.RetryBucket(backoff_scale_factor_ms = -1) @test_throws ArgumentError HT.RetryBucket(max_backoff_secs = -1) + + partitions = Dict{String,HT._RetryPartition}("depleted.example" => HT._RetryPartition(5)) + positional = HT.RetryBucket(25, 20, 10, partitions, ReentrantLock()) + @test positional.partitions === partitions + @test (@atomic :acquire positional.depleted_partitions) == Set(["depleted.example"]) + + converted = HT.RetryBucket(Int32(25), Int16(20), Int8(10), copy(partitions), ReentrantLock()) + @test converted.backoff_scale_factor_ms === 25 + @test converted.max_backoff_secs === 20 + @test converted.capacity === 10 + @test converted.partitions == partitions end @testset "HTTP retry bucket acquire/release is partitioned and case-insensitive" begin @@ -173,6 +184,88 @@ end Base.release(bucket, token1, 0) end +@testset "HTTP retry bucket refunds retried attempts that reach a final response (#1353)" begin + # A retried attempt that reached a non-retryable response refunds its + # reservation in full; retryable responses keep the partial cost; `nothing` + # (the retry never launched) refunds in full. + @test HT._retry_bucket_failure_cost(nothing) == 0 + @test HT._retry_bucket_failure_cost(200) == 0 + @test HT._retry_bucket_failure_cost(404) == 0 + @test HT._retry_bucket_failure_cost(501) == 0 + for retryable_status in (408, 429, 500, 502, 503, 504) + @test HT._retry_bucket_failure_cost(retryable_status) == HT._RETRY_BUCKET_RETRYABLE_RESPONSE_COST + @test HT._retryable_status(retryable_status) + end + @test HT._retry_bucket_response_cost(true) == HT._RETRY_BUCKET_RETRYABLE_RESPONSE_COST + @test HT._retry_bucket_response_cost(false) == 0 + + # Full refund on success: with capacity for exactly one reservation, a + # second acquire only succeeds because the first returned its cost. + bucket = HT.RetryBucket(capacity = 10) + token = Base.acquire(bucket, "svc.example") + Base.release(bucket, token, HT._retry_bucket_failure_cost(200)) + token2 = Base.acquire(bucket, "svc.example") + Base.release(bucket, token2, 0) +end + +@testset "HTTP retry bucket replenishes consumed capacity (#1353)" begin + bucket = HT.RetryBucket(capacity = 20) + @test isempty(@atomic :acquire bucket.depleted_partitions) + + # Replenish before any capacity was ever spent is a lock-free no-op and + # creates no partitions. + HT._retry_bucket_replenish!(bucket, "svc.example") + @test isempty(bucket.partitions) + + token = Base.acquire(bucket, "svc.example") + @test (@atomic :acquire bucket.depleted_partitions) == Set(["svc.example"]) + Base.release(bucket, token, HT._RETRY_BUCKET_ACQUIRE_COST) + @test bucket.partitions["svc.example"].capacity == 10 + + for _ in 1:5 + HT._retry_bucket_replenish!(bucket, "svc.example") + end + @test bucket.partitions["svc.example"].capacity == 15 + @test (@atomic :acquire bucket.depleted_partitions) == Set(["svc.example"]) + + # Case-insensitive, and capped at full capacity. + for _ in 1:10 + HT._retry_bucket_replenish!(bucket, "SVC.example") + end + @test bucket.partitions["svc.example"].capacity == 20 + @test isempty(@atomic :acquire bucket.depleted_partitions) + + # Untouched partitions are not affected by another partition's depletion. + other = Base.acquire(bucket, "other.example") + HT._retry_bucket_replenish!(bucket, "svc.example") + @test bucket.partitions["svc.example"].capacity == 20 + @test (@atomic :acquire bucket.depleted_partitions) == Set(["other.example"]) + Base.release(bucket, other, 0) + @test isempty(@atomic :acquire bucket.depleted_partitions) +end + +@testset "HTTP retry bucket heals only after successful responses" begin + server = HT.serve!("127.0.0.1", 0; listenany = true) do request + return request.target == "/healthy" ? HT.Response(200) : HT.Response(404) + end + bucket = HT.RetryBucket(capacity = 20, backoff_scale_factor_ms = 0, max_backoff_secs = 0) + spent = Base.acquire(bucket, "127.0.0.1") + Base.release(bucket, spent, HT._RETRY_BUCKET_ACQUIRE_COST) + client = HT.Client(transport = HT.Transport(retry_bucket = bucket), cookiejar = nothing) + try + failed = HT.get(client, "http://127.0.0.1:$(HT.port(server))/missing"; retries = 1, status_exception = false) + @test failed.status == 404 + @test bucket.partitions["127.0.0.1"].capacity == 10 + + healthy = HT.get(client, "http://127.0.0.1:$(HT.port(server))/healthy"; retries = 1) + @test healthy.status == 200 + @test bucket.partitions["127.0.0.1"].capacity == 11 + finally + close(client) + HT.forceclose(server) + end +end + @testset "HTTP transport owns an optional default retry bucket" begin default_transport = HT.Transport() @test default_transport.retry_bucket isa HT.RetryBucket @@ -404,12 +497,17 @@ end (status = 418, reason = "I'm a teapot", retry_after = "0"), (status = 200, reason = "OK", body_text = "ok"), ], seen1) - force_retry = (attempt, err, req, resp) -> resp !== nothing && resp.status == 418 ? true : nothing + policy_attempts = Int[] + force_retry = (attempt, err, req, resp) -> begin + push!(policy_attempts, attempt) + return resp !== nothing && resp.status == 418 ? true : nothing + end try response = HT.get("$(base_url1)/hook"; retries = 1, retry_if = force_retry, status_exception = false, retry_bucket = HT.RetryBucket(backoff_scale_factor_ms = 0, max_backoff_secs = 0)) @test response.status == 200 _wait_task_retry!(server_task1) @test seen1 == [("GET", "/hook", ""), ("GET", "/hook", "")] + @test policy_attempts == [1] finally HTTP.@try_ignore NC.close(listener1) end @@ -430,6 +528,267 @@ end end end +@testset "HTTP retry bucket accounts for retry_if decisions" begin + attempts = Ref(0) + server = HT.serve!("127.0.0.1", 0; listenany = true) do _ + attempts[] += 1 + return HT.Response(418, "still retryable by policy") + end + events = Any[] + policy_attempts = Int[] + bucket = HT.RetryBucket(capacity = 10, backoff_scale_factor_ms = 0, max_backoff_secs = 0) + retry_if = (attempt, _, _, response) -> begin + push!(policy_attempts, attempt) + return response !== nothing && response.status == 418 + end + try + response = HT.request( + event -> push!(events, event), + "GET", + "http://127.0.0.1:$(HT.port(server))/custom"; + retries = 3, + retry_if = retry_if, + retry_bucket = bucket, + status_exception = false, + ) + @test response.status == 418 + @test attempts[] == 2 + @test policy_attempts == [1, 2] + skipped = [event for event in events if event isa HT.RetrySkippedEvent] + @test length(skipped) == 1 + @test (only(skipped)::HT.RetrySkippedEvent).reason === :retry_bucket + @test bucket.partitions["127.0.0.1"].capacity == 5 + finally + HT.forceclose(server) + end + + # A custom policy can suppress a built-in response retry. The effective + # policy result must also decide whether the prior reservation keeps cost. + attempts[] = 0 + empty!(policy_attempts) + suppressing_server = HT.serve!("127.0.0.1", 0; listenany = true) do _ + attempts[] += 1 + return HT.Response(503, "built-in retryable") + end + suppressing_bucket = HT.RetryBucket(capacity = 10, backoff_scale_factor_ms = 0, max_backoff_secs = 0) + suppress_second = (attempt, _, _, response) -> begin + push!(policy_attempts, attempt) + return response !== nothing && attempt == 1 + end + try + response = HT.get( + "http://127.0.0.1:$(HT.port(suppressing_server))/suppressed"; + retries = 3, + retry_if = suppress_second, + retry_bucket = suppressing_bucket, + status_exception = false, + ) + @test response.status == 503 + @test attempts[] == 2 + @test policy_attempts == [1, 2] + @test suppressing_bucket.partitions["127.0.0.1"].capacity == 10 + finally + HT.forceclose(suppressing_server) + end +end + +@testset "HTTP retry bucket preserves terminal built-in response charging" begin + attempts = Ref(0) + server = HT.serve!("127.0.0.1", 0; listenany = true) do _ + attempts[] += 1 + return HT.Response(503, "still unavailable") + end + bucket = HT.RetryBucket(capacity = 10, backoff_scale_factor_ms = 0, max_backoff_secs = 0) + events = Any[] + url = "http://127.0.0.1:$(HT.port(server))/terminal-failure" + try + first = HT.get(url; retries = 1, retry_bucket = bucket, status_exception = false) + @test first.status == 503 + @test attempts[] == 2 + @test bucket.partitions["127.0.0.1"].capacity == 5 + + second = HT.request( + event -> push!(events, event), + "GET", + url; + retries = 1, + retry_bucket = bucket, + status_exception = false, + ) + @test second.status == 503 + @test attempts[] == 3 + skipped = [event for event in events if event isa HT.RetrySkippedEvent] + @test length(skipped) == 1 + @test (only(skipped)::HT.RetrySkippedEvent).reason === :retry_bucket + finally + HT.forceclose(server) + end +end + +@testset "HTTP retry bucket charges terminal custom-only response failures" begin + attempts = Ref(0) + server = HT.serve!("127.0.0.1", 0; listenany = true) do _ + attempts[] += 1 + return HT.Response(418, "custom retry failure") + end + bucket = HT.RetryBucket(capacity = 10, backoff_scale_factor_ms = 0, max_backoff_secs = 0) + retry_if = (_, _, _, response) -> response !== nothing && response.status == 418 + url = "http://127.0.0.1:$(HT.port(server))/terminal-custom-failure" + try + first = HT.get( + url; + retries = 1, + retry_if = retry_if, + retry_bucket = bucket, + status_exception = false, + ) + @test first.status == 418 + @test attempts[] == 2 + @test bucket.partitions["127.0.0.1"].capacity == 5 + + second = HT.get( + url; + retries = 1, + retry_if = retry_if, + retry_bucket = bucket, + status_exception = false, + ) + @test second.status == 418 + @test attempts[] == 3 + finally + HT.forceclose(server) + end +end + +@testset "HTTP retry bucket carries an explicit custom decision to the terminal response" begin + attempts = Ref(0) + server = HT.serve!("127.0.0.1", 0; listenany = true) do _ + attempts[] += 1 + return isodd(attempts[]) ? HT.Response(503, "built-in and custom") : HT.Response(418, "custom only") + end + bucket = HT.RetryBucket(capacity = 10, backoff_scale_factor_ms = 0, max_backoff_secs = 0) + retry_if = (_, _, _, response) -> response !== nothing && response.status >= 400 + url = "http://127.0.0.1:$(HT.port(server))/mixed-terminal-failure" + try + first = HT.get( + url; + retries = 1, + retry_if = retry_if, + retry_bucket = bucket, + status_exception = false, + ) + @test first.status == 418 + @test attempts[] == 2 + @test bucket.partitions["127.0.0.1"].capacity == 5 + + second = HT.get( + url; + retries = 1, + retry_if = retry_if, + retry_bucket = bucket, + status_exception = false, + ) + @test second.status == 503 + @test attempts[] == 3 + finally + HT.forceclose(server) + end +end + +@testset "HTTP retry tracing failures release reserved capacity" begin + for throw_on in (HT.RetryEvent, HT.RequestEvent, HT.ResponseHeadEvent) + attempts = Ref(0) + server = HT.serve!("127.0.0.1", 0; listenany = true) do _ + attempts[] += 1 + return attempts[] == 1 ? HT.Response(503, "retry") : HT.Response(200, "ok") + end + bucket = HT.RetryBucket(capacity = 10, backoff_scale_factor_ms = 0, max_backoff_secs = 0) + transport = HT.Transport( + retry_bucket = bucket, + max_conns_per_host = 1, + max_idle_per_host = 1, + max_idle_total = 1, + ) + client = HT.Client(transport = transport, cookiejar = nothing) + trace_err = ErrorException("trace failed") + trace = event -> begin + if event isa throw_on + matching_attempt = throw_on === HT.RetryEvent || getfield(event, :attempt) == 2 + matching_attempt && throw(trace_err) + end + return nothing + end + try + err = try + HT.request( + trace, + "GET", + "http://127.0.0.1:$(HT.port(server))/trace"; + client = client, + retries = 1, + status_exception = false, + ) + nothing + catch caught + caught + end + @test err === trace_err + @test bucket.partitions["127.0.0.1"].capacity == 10 + @test isempty(@atomic :acquire bucket.depleted_partitions) + @test lock(transport.lock) do + isempty(transport.conns_per_host) + end + finally + close(client) + HT.forceclose(server) + end + end +end + +@testset "HTTP retry policy failures preserve response ownership cleanup" begin + attempts = Ref(0) + server = HT.serve!("127.0.0.1", 0; listenany = true) do _ + attempts[] += 1 + return attempts[] == 1 ? HT.Response(503, "retry") : HT.Response(200, "ok") + end + bucket = HT.RetryBucket(capacity = 10, backoff_scale_factor_ms = 0, max_backoff_secs = 0) + transport = HT.Transport( + retry_bucket = bucket, + max_conns_per_host = 1, + max_idle_per_host = 1, + max_idle_total = 1, + ) + client = HT.Client(transport = transport, cookiejar = nothing) + policy_err = ErrorException("policy failed") + retry_if = (attempt, _, _, response) -> begin + attempt == 2 && response !== nothing && throw(policy_err) + return response !== nothing && response.status == 503 + end + try + err = try + HT.get( + client, + "http://127.0.0.1:$(HT.port(server))/policy"; + retries = 2, + retry_if = retry_if, + status_exception = false, + ) + nothing + catch caught + caught + end + @test err === policy_err + @test bucket.partitions["127.0.0.1"].capacity == 10 + @test isempty(@atomic :acquire bucket.depleted_partitions) + @test lock(transport.lock) do + isempty(transport.conns_per_host) + end + finally + close(client) + HT.forceclose(server) + end +end + @testset "HTTP retry_if sees RequestRetryError for request-path failures" begin seen_err = Ref{Any}(nothing) hook = (attempt, err, req, resp) -> begin @@ -453,6 +812,11 @@ end address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) base_url = "http://$(address)" seen = Tuple{String, String, String}[] + hook_calls = Ref(0) + force_retry = (_, _, _, _) -> begin + hook_calls[] += 1 + return true + end server_task = _serve_retry_sequence(listener, [(status = 503, reason = "Service Unavailable", retry_after = "0")], seen) try response = HT.post( @@ -460,11 +824,13 @@ end body = _OneShotIO("payload"), retries = 1, retry_non_idempotent = true, + retry_if = force_retry, status_exception = false, ) @test response.status == 503 _wait_task_retry!(server_task) @test seen == [("POST", "/streaming", "payload")] + @test hook_calls[] == 0 finally HTTP.@try_ignore NC.close(listener) end @@ -515,10 +881,11 @@ end request = HT.Request("GET", "/deadline"; host="example.com", context=HT.RequestContext(deadline_ns=1)) response = HT.Response(503; headers=["Retry-After" => "60"]) - armed, token, delay_ns = HT._arm_request_retry!(controller, "example.com:80", request, 1, response) + armed, token, delay_ns, skip_reason = HT._arm_request_retry!(controller, "example.com:80", request, 1, response) @test !armed @test token === nothing @test delay_ns == 60_000_000_000 + @test skip_reason === :deadline # Full refund: with capacity 15 a second reservation (cost 10) only # succeeds if the abandoned one returned its 10. @@ -526,6 +893,115 @@ end Base.release(bucket, refunded, 0) end +@testset "retry delay rejects a deadline crossed during sleep" begin + request = HT.Request("GET", "/deadline"; host = "example.com", context = HT.RequestContext(deadline_ns = 200)) + ticks = Int64[100, 201] + tick_index = Ref(0) + clock_ns = () -> begin + tick_index[] += 1 + return ticks[tick_index[]] + end + slept = Ref{Int64}(0) + sleep_ns = delay_ns -> (slept[] = delay_ns) + + @test !HT._sleep_retry_delay!(request, Int64(100); clock_ns = clock_ns, sleep_ns = sleep_ns) + @test slept[] == 100 +end + +@testset "HTTP request retry that recovers refunds the retry bucket (#1353)" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + seen = Tuple{String, String, String}[] + scenarios = [ + (status = 503, reason = "Service Unavailable", retry_after = "0"), + (status = 200, reason = "OK", body_text = "ok"), + ] + server_task = _serve_retry_sequence(listener, scenarios, seen) + try + bucket = HT.RetryBucket(capacity = 20, backoff_scale_factor_ms = 0, max_backoff_secs = 0) + response = HT.get("http://$(address)/refund"; retries = 2, retry_bucket = bucket, status_exception = false) + @test response.status == 200 + _wait_task_retry!(server_task) + # The armed retry reserved 10 and recovered with a 200, so the + # reservation was refunded in full instead of consumed (#1353). + @test bucket.partitions["127.0.0.1"].capacity == 20 + @test isempty(@atomic :acquire bucket.depleted_partitions) + finally + HTTP.@try_ignore NC.close(listener) + end +end + +@testset "HTTP request trace emits RetrySkippedEvent when the bucket denies (#1353)" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + seen = Tuple{String, String, String}[] + scenarios = [ + (status = 503, reason = "Service Unavailable", retry_after = "0"), + (status = 503, reason = "Service Unavailable", retry_after = "0"), + ] + server_task = _serve_retry_sequence(listener, scenarios, seen) + events = Any[] + try + # Capacity for exactly one reservation: the first retry consumes 10 and + # releases 5 back after the retried attempt's 503, so arming the second + # retry (cost 10 > 5) is denied by the bucket. + bucket = HT.RetryBucket(capacity = 10, backoff_scale_factor_ms = 0, max_backoff_secs = 0) + response = HT.request(event -> push!(events, event), "GET", "http://$(address)/denied"; retries = 2, retry_bucket = bucket, status_exception = false) + @test response.status == 503 + _wait_task_retry!(server_task) + skipped = [event for event in events if event isa HT.RetrySkippedEvent] + @test length(skipped) == 1 + skip_event = skipped[1]::HT.RetrySkippedEvent + @test skip_event.reason === :retry_bucket + @test skip_event.attempt == 2 + @test skip_event.err === nothing + @test (skip_event.response::HT.Response).status == 503 + @test skip_event.redirect_count == 0 + finally + HTTP.@try_ignore NC.close(listener) + end +end + +@testset "HTTP request trace emits RetrySkippedEvent for request-path failures (#1353)" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + server_task = errormonitor(Threads.@spawn begin + conn = NC.accept(listener) + try + # Read the request, then close without responding: the request + # fails with a retryable request-path error on a fresh connection. + _ = HT.read_request(HT._ConnReader(conn)) + finally + HTTP.@try_ignore NC.close(conn) + end + return nothing + end) + events = Any[] + bucket = HT.RetryBucket(capacity = 10, backoff_scale_factor_ms = 0, max_backoff_secs = 0) + drained = Base.acquire(bucket, "127.0.0.1") # empty the partition up front + try + err = try + HT.request(event -> push!(events, event), "GET", "http://$(address)/skip"; retries = 2, retry_bucket = bucket) + nothing + catch e + e + end + @test err isa Exception + @test HT.isrecoverable(err::Exception) + _wait_task_retry!(server_task) + skipped = [event for event in events if event isa HT.RetrySkippedEvent] + @test length(skipped) == 1 + skip_event = skipped[1]::HT.RetrySkippedEvent + @test skip_event.reason === :retry_bucket + @test skip_event.attempt == 1 + @test skip_event.response === nothing + @test skip_event.err isa Exception + finally + Base.release(bucket, drained, 0) + HTTP.@try_ignore NC.close(listener) + end +end + @testset "HTTP.open retries idempotent buffered requests" begin listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) @@ -605,6 +1081,27 @@ end @test HT.isrecoverable(HT.RequestRetryError(EOFError())) @test !HT.isrecoverable(HT.RequestRetryError(ArgumentError("nope"))) + # unwraps the public TLS wrappers (and TLSError causes inside them) so + # downstream retry loops can classify wrapped errors caught from + # HTTP.request (#1353) + reset_tls = Reseau.TLS.TLSError("read", Int32(0), "unexpected TLS failure", SystemError("read", 0)) + @test HT.isrecoverable(reset_tls) + @test HT.isrecoverable(HT.TLSTransportError(reset_tls)) + @test HT.isrecoverable(HT.TLSHandshakeError(reset_tls)) + @test !HT.isrecoverable(HT.TLSTransportError(Reseau.TLS.TLSError("read", Int32(0), "bad record mac", nothing))) + @test !HT.isrecoverable(HT.TLSHandshakeError(ErrorException("cert rejected"))) + + unexpected_eof_cause = ErrorException("opaque TLS record EOF cause") + unexpected_eof = Reseau.TLS.TLSError("read", Int32(0), "unexpected EOF", unexpected_eof_cause) + @test HT.isrecoverable(HT.TLSTransportError(unexpected_eof)) + + h2_tls = Reseau.TLS.TLSError("read", Int32(0), "unexpected TLS failure", SystemError("read", 0)) + h2_read_error = HT.ProtocolError("HTTP/2 read loop failed", h2_tls) + @test HT.isrecoverable(h2_read_error) + wrapped_h2 = HT._wrap_client_transport_error(h2_read_error) + @test wrapped_h2 isa HT.TLSTransportError + @test (wrapped_h2::HT.TLSTransportError).cause === h2_tls + # matches the internal classifier the built-in policy uses for err in (EOFError(), HT.ParseError("x"), ArgumentError("y"), Reseau.IOPoll.DeadlineExceededError()) @test HT.isrecoverable(err) == HT._retryable_request_error(err)