From 626b9d65e7b41a5ebbdd6fac1891581210415058 Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Fri, 28 Aug 2026 10:31:55 -0600 Subject: [PATCH 1/4] Fix retry-budget exhaustion and correlated reused-connection death MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A client sending periodic bursts against a peer that silently discards idle pooled connections saw transient errors escape HTTP.request with default retry=true after ~50 burst cycles, permanently (#1353). Root cause: every armed high-level retry consumed 10 of the per-host RetryBucket's 500 tokens even when the retried attempt succeeded, the bucket never refilled, and once empty the denial was silent — so the budget was a fuse that burned out in ~50 recoveries, after which every transient failure surfaced raw. - RetryBucket: refund the reservation when a retried attempt reaches a non-retryable response (retryable 408/429/5xx responses keep partial cost, exception outcomes keep full cost), and credit 1 unit per successful non-retried request so a drained partition heals from healthy traffic. A depleted-partition counter keeps the per-request replenish check lock-free while all partitions are full. - Transport: retry a replayable idempotent request while failures keep landing on *reused* pooled connections (bounded by max_idle_per_host + 1 acquisitions) instead of exactly once, so a correlated-death batch is burned through down to a fresh dial without consuming high-level retry budget; classify TLSError on reused connections by its cause so dead reused TLS connections take this path too. - Add RetrySkippedEvent so a denied (:retry_bucket) or deadline- preempted (:deadline) retry is observable in request traces. - Add TLSTransportError for TLS I/O failures on established connections; handshake failures are now typed TLSHandshakeError at the dial sites instead of one blanket wrap that mislabeled mid-request read errors as handshake errors. isrecoverable unwraps both. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 29 +++++++ docs/src/api/client.md | 1 + docs/src/api/core.md | 1 + src/HTTP.jl | 3 +- src/http2_client.jl | 9 +- src/http_client.jl | 58 ++++++++++++- src/http_client_retry.jl | 23 +++++- src/http_core.jl | 32 ++++++- src/http_retry.jl | 67 +++++++++++++-- src/http_transport.jl | 54 +++++++++--- test/http_client_transport_tests.jl | 96 +++++++++++++++++++++ test/http_core_tests.jl | 18 +++- test/http_retry_tests.jl | 124 +++++++++++++++++++++++++++- 13 files changed, 482 insertions(+), 33 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 51e4a6551..443a6df30 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,12 +10,40 @@ 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`. Successful retries now refund their reservation, + and each successful non-retried request restores one unit of previously + consumed budget. ([#1353]) +- The HTTP/1 transport now retries a replayable idempotent request for as long + as failures land on *reused* pooled connections (bounded by + `max_idle_per_host + 1` connection acquisitions) instead of exactly once. + 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`) are now classified by + their underlying cause in the transport's reused-connection retry, instead + of skipping that retry and consuming high-level retry budget. ([#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 +854,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/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..f6da6e851 100644 --- a/src/http2_client.jl +++ b/src/http2_client.jl @@ -1216,7 +1216,14 @@ function _connect_h2_from_tcp!( 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 + TLS.handshake!(tls_conn) + catch err + # Type TLS failures at the site where the phase is known (see + # the matching wrap in the HTTP/1 transport's _new_conn_tls!). + err isa TLS.TLSError && throw(TLSHandshakeError(err::TLS.TLSError)) + rethrow() + end stream_reader = _ConnReader(tls_conn::TLS.Conn) else stream_reader = _ConnReader(tcp) diff --git a/src/http_client.jl b/src/http_client.jl index 794b2b77c..864a17b53 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 @@ -904,7 +945,7 @@ function _do_incoming!( retry_token = nothing 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) + 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 @@ -912,6 +953,7 @@ function _do_incoming!( retry_token = next_token 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) @@ -925,8 +967,15 @@ function _do_incoming!( _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)) + if retry_controller !== nothing && retry_controller.bucket !== nothing + response_bucket = retry_controller.bucket::RetryBucket + if retry_token !== nothing + release(response_bucket, retry_token::RetryToken, _retry_bucket_failure_cost(status_response.status)) + elseif !_retryable_status(status_response.status) + # A successful non-retried request slowly heals retry budget + # consumed by an earlier failure burst (#1353). + _retry_bucket_replenish!(response_bucket, _retry_partition_for_address(current_address)) + end end retry_token = nothing if retry_controller !== nothing @@ -939,7 +988,7 @@ function _do_incoming!( rethrow() end if should_retry - scheduled, next_token, delay_ns = _arm_request_retry!(retry_controller, current_address, current_request, retry_attempt, status_response) + 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 @@ -950,6 +999,7 @@ function _do_incoming!( 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 end if !_is_redirect_status(response.head.status) diff --git a/src/http_client_retry.jl b/src/http_client_retry.jl index 11c9c2645..473f1a609 100644 --- a/src/http_client_retry.jl +++ b/src/http_client_retry.jl @@ -64,6 +64,14 @@ function _retryable_request_error(err::Exception)::Bool 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 cause = (current::TLS.TLSError).cause cause === nothing && return false @@ -85,7 +93,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 +and of the public [`TLSHandshakeError`](@ref)/[`TLSTransportError`](@ref) +wrappers. A request *deadline* being exceeded (`IOPoll.DeadlineExceededError`) is treated as non-recoverable, as is anything else. @@ -183,6 +193,11 @@ function _sleep_retry_delay!(request::Request, delay_ns::Int64)::Bool 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 +213,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..7c0a33497 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,7 +298,12 @@ 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)) if err isa TLS.TLSHandshakeTimeoutError @@ -296,7 +322,7 @@ function _wrap_client_transport_error(err, operation::AbstractString="request", return DNSError(err.name, err) end if err isa TLS.TLSError - return TLSHandshakeError(err) + return TLSTransportError(err) end return err end diff --git a/src/http_retry.jl b/src/http_retry.jl index 28c36b10a..13638c6dc 100644 --- a/src/http_retry.jl +++ b/src/http_retry.jl @@ -27,6 +27,13 @@ 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 for retryable responses (429/5xx), 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 request — 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 +41,10 @@ mutable struct RetryBucket capacity::Int partitions::Dict{String,_RetryPartition} lock::ReentrantLock + # Number of partitions currently below full capacity. Maintained under + # `lock`; read without it as a fast path so per-request replenish checks + # stay lock-free while every partition is full. + @atomic depleted::Int end """Handle returned by `acquire` and consumed by `release` to refund retry budget.""" @@ -74,9 +85,24 @@ function RetryBucket(; Int(capacity), Dict{String,_RetryPartition}(), ReentrantLock(), + 0, ) end +# Set a partition's capacity while keeping the bucket's depleted-partition +# count in sync. Must be called with `bucket.lock` held. +@inline function _retry_partition_set_capacity!(bucket::RetryBucket, 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 + @atomic bucket.depleted += 1 + elseif !was_full && now_full + @atomic bucket.depleted -= 1 + 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 +123,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, state, state.capacity - _RETRY_BUCKET_ACQUIRE_COST) return RetryToken(bucket, partition_key, _RETRY_BUCKET_ACQUIRE_COST, false) end end @@ -106,12 +132,14 @@ 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 release(bucket::RetryBucket, token::RetryToken, failure_cost::Int)::Nothing @@ -123,7 +151,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, state, min(bucket.capacity, state.capacity + refund)) token.released = true return nothing finally @@ -131,6 +159,33 @@ 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, and the depleted-partition fast path keeps this lock-free while +every partition is full. +""" +function _retry_bucket_replenish!(bucket::RetryBucket, partition)::Nothing + (@atomic :monotonic bucket.depleted) == 0 && return nothing + partition_key = _retry_bucket_partition_key(partition) + 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_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_transport.jl b/src/http_transport.jl index 6fbca02ce..3b7f5651c 100644 --- a/src/http_transport.jl +++ b/src/http_transport.jl @@ -1021,8 +1021,13 @@ function _new_conn_tls!( 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 + 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 @@ -1486,13 +1491,29 @@ end end @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`). Classify by the cause so dead + # reused connections are retried here instead of consuming the + # caller's retry budget (#1353). + 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 @@ -1730,6 +1751,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 — up to +`max_idle_per_host + 1` acquisitions, enough to burn through a fully poisoned +pool and reach a fresh dial (#1353). A failure on a freshly dialed connection +propagates. + Throws parser, protocol, transport, TLS, and timeout exceptions depending on where the exchange fails. """ @@ -1741,9 +1771,12 @@ 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 + if retry_template === nothing && attempt == 1 && _retryable_request(request) + retry_template = _copy_request(request) + end 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) @@ -1884,7 +1917,7 @@ function _roundtrip_incoming!( 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) + if attempt <= transport.max_idle_per_host && was_reused && retry_template !== nothing && _retryable_reused_conn_error(err) return _roundtrip_incoming!( transport, address, @@ -1893,6 +1926,7 @@ function _roundtrip_incoming!( server_name, proxy_config, attempt + 1, + retry_template, ) end rethrow(err) diff --git a/test/http_client_transport_tests.jl b/test/http_client_transport_tests.jl index 09deabc90..a97bd2db9 100644 --- a/test/http_client_transport_tests.jl +++ b/test/http_client_transport_tests.jl @@ -1033,6 +1033,102 @@ end @test HT._retryable_reused_conn_error(Reseau.IOPoll.NotPollableError()) 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()), + ) + # 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 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 + @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..85b3417fc 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,23 @@ 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 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..d9c0057d1 100644 --- a/test/http_retry_tests.jl +++ b/test/http_retry_tests.jl @@ -173,6 +173,63 @@ 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 + + # 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 (@atomic bucket.depleted) == 0 + + # 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 bucket.depleted) == 1 + 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 bucket.depleted) == 1 + + # 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 (@atomic bucket.depleted) == 0 + + # 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 + Base.release(bucket, other, 0) + @test (@atomic bucket.depleted) == 0 +end + @testset "HTTP transport owns an optional default retry bucket" begin default_transport = HT.Transport() @test default_transport.retry_bucket isa HT.RetryBucket @@ -515,10 +572,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 +584,60 @@ end Base.release(bucket, refunded, 0) 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 (@atomic bucket.depleted) == 0 + 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.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 +717,16 @@ 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"))) + # 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) From 506af7f98411c8cd0867ad2829521878e68ced81 Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Fri, 28 Aug 2026 10:56:24 -0600 Subject: [PATCH 2/4] Cover TLS dial-site wrapping and RetrySkippedEvent emission paths Add tests for the handshake-phase TLSHandshakeError wrap on both the HTTP/1 transport dial and connect_h2!, the request-path (exception) RetrySkippedEvent emission, and the verbose trace formatting of RetrySkippedEvent. Co-Authored-By: Claude Fable 5 --- test/http2_client_tests.jl | 26 +++++++++++++++++++ test/http_client_tests.jl | 13 ++++++++++ test/http_client_transport_tests.jl | 32 +++++++++++++++++++++++ test/http_retry_tests.jl | 40 +++++++++++++++++++++++++++++ 4 files changed, 111 insertions(+) diff --git a/test/http2_client_tests.jl b/test/http2_client_tests.jl index 3a9607012..c3db8c9a2 100644 --- a/test/http2_client_tests.jl +++ b/test/http2_client_tests.jl @@ -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") diff --git a/test/http_client_tests.jl b/test/http_client_tests.jl index 7186dd2c1..812c1760d 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 diff --git a/test/http_client_transport_tests.jl b/test/http_client_transport_tests.jl index a97bd2db9..a17c71c8f 100644 --- a/test/http_client_transport_tests.jl +++ b/test/http_client_transport_tests.jl @@ -1033,6 +1033,38 @@ end @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 diff --git a/test/http_retry_tests.jl b/test/http_retry_tests.jl index d9c0057d1..883020dc4 100644 --- a/test/http_retry_tests.jl +++ b/test/http_retry_tests.jl @@ -638,6 +638,46 @@ end 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)) From 30462e6410dbb37d7df51102030144b9ecc1c05b Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Fri, 28 Aug 2026 12:47:08 -0600 Subject: [PATCH 3/4] fix(client): harden retry and transport recovery Make retry-budget accounting exact across built-in and custom policies. Recover poisoned pooled connections without leaking slots or reusing a failed connection. Normalize established TLS failures across HTTP/1 and HTTP/2 public client boundaries. Add regression coverage for terminal accounting, trace failures, one-shot bodies, concurrent pool handoff, and truncated TLS records. Fixes #1353 --- CHANGELOG.md | 26 +- docs/src/guides/client.md | 1 + src/http2_client.jl | 36 ++- src/http_client.jl | 147 +++++++--- src/http_client_retry.jl | 73 +++-- src/http_core.jl | 30 +- src/http_retry.jl | 105 +++++-- src/http_stream.jl | 39 +-- src/http_transport.jl | 428 +++++++++++++++------------ test/http2_client_tests.jl | 124 +++++++- test/http_client_tests.jl | 19 ++ test/http_client_transport_tests.jl | 436 ++++++++++++++++++++++++++++ test/http_core_tests.jl | 1 + test/http_retry_tests.jl | 349 +++++++++++++++++++++- 14 files changed, 1482 insertions(+), 332 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 443a6df30..3444f3903 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,19 +31,29 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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`. Successful retries now refund their reservation, - and each successful non-retried request restores one unit of previously - consumed budget. ([#1353]) + raw despite `retry=true`. Successful retries now refund their reservation. + Retry responses use the effective built-in or custom `retry_if` decision + while another configured retry remains. A terminal armed response uses the + built-in classification without invoking `retry_if` after retry slots are + exhausted. Retries explicitly requested by `retry_if` conservatively keep + cost on terminal non-success responses. + Each successful non-retried request restores one unit of + previously consumed budget. Retry reservations and response connections are + also released if a trace or retry-policy callback throws, and the request + deadline is rechecked after backoff sleep. ([#1353]) - The HTTP/1 transport now retries a replayable idempotent request for as long - as failures land on *reused* pooled connections (bounded by - `max_idle_per_host + 1` connection acquisitions) instead of exactly once. + 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`) are now classified by - their underlying cause in the transport's reused-connection retry, instead - of skipping that retry and consuming high-level retry budget. ([#1353]) + 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 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/http2_client.jl b/src/http2_client.jl index f6da6e851..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,14 +1219,14 @@ 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) 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 - # Type TLS failures at the site where the phase is known (see - # the matching wrap in the HTTP/1 transport's _new_conn_tls!). + # 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 @@ -1660,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 @@ -1671,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 @@ -1891,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 864a17b53..5b94f6e63 100644 --- a/src/http_client.jl +++ b/src/http_client.jl @@ -886,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 @@ -939,18 +950,29 @@ 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) + 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)) @@ -958,42 +980,54 @@ function _do_incoming!( 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 - response_bucket = retry_controller.bucket::RetryBucket - if retry_token !== nothing - release(response_bucket, retry_token::RetryToken, _retry_bucket_failure_cost(status_response.status)) - elseif !_retryable_status(status_response.status) - # A successful non-retried request slowly heals retry budget - # consumed by an earlier failure burst (#1353). - _retry_bucket_replenish!(response_bucket, _retry_partition_for_address(current_address)) - end - 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 + 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 @@ -1001,6 +1035,22 @@ function _do_incoming!( 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 @@ -1169,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 @@ -2190,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 @@ -2206,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 473f1a609..ef4f7b509 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,6 +52,12 @@ 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 @@ -73,6 +71,7 @@ function _retryable_request_error(err::Exception)::Bool continue end if current isa TLS.TLSError + (current::TLS.TLSError).message == "unexpected EOF" && return true cause = (current::TLS.TLSError).cause cause === nothing && return false current = cause::Exception @@ -93,9 +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 -and of the public [`TLSHandshakeError`](@ref)/[`TLSTransportError`](@ref) -wrappers. +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. @@ -130,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}) @@ -179,17 +202,23 @@ 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 diff --git a/src/http_core.jl b/src/http_core.jl index 7c0a33497..769565e7d 100644 --- a/src/http_core.jl +++ b/src/http_core.jl @@ -305,7 +305,9 @@ 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 @@ -321,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 TLSTransportError(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 13638c6dc..61b050d6c 100644 --- a/src/http_retry.jl +++ b/src/http_retry.jl @@ -12,6 +12,15 @@ 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) +@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 @@ -30,10 +39,13 @@ 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 for retryable responses (429/5xx), 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 request — so a burst of real -failures can drain a partition, but healthy traffic always heals it. +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 @@ -41,10 +53,10 @@ mutable struct RetryBucket capacity::Int partitions::Dict{String,_RetryPartition} lock::ReentrantLock - # Number of partitions currently below full capacity. Maintained under - # `lock`; read without it as a fast path so per-request replenish checks - # stay lock-free while every partition is full. - @atomic depleted::Int + # 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.""" @@ -85,20 +97,67 @@ function RetryBucket(; Int(capacity), Dict{String,_RetryPartition}(), ReentrantLock(), - 0, + Set{String}(), ) end -# Set a partition's capacity while keeping the bucket's depleted-partition -# count in sync. Must be called with `bucket.lock` held. -@inline function _retry_partition_set_capacity!(bucket::RetryBucket, state::_RetryPartition, new_capacity::Int)::Nothing +# 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 - @atomic bucket.depleted += 1 + depleted = copy(@atomic :acquire bucket.depleted_partitions) + push!(depleted, partition_key) + @atomic :release bucket.depleted_partitions = depleted elseif !was_full && now_full - @atomic bucket.depleted -= 1 + depleted = copy(@atomic :acquire bucket.depleted_partitions) + delete!(depleted, partition_key) + @atomic :release bucket.depleted_partitions = depleted end return nothing end @@ -123,7 +182,7 @@ function acquire(bucket::RetryBucket, partition) if state.capacity < _RETRY_BUCKET_ACQUIRE_COST throw(RetryDeniedError(partition_key)) end - _retry_partition_set_capacity!(bucket, state, 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 @@ -142,6 +201,10 @@ end 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 token.bucket === bucket || throw(ArgumentError("retry token does not belong to the provided retry bucket")) lock(bucket.lock) @@ -151,7 +214,7 @@ end reserved = _retry_bucket_reserved_cost(token) consumed = min(reserved, max(0, failure_cost)) refund = reserved - consumed - _retry_partition_set_capacity!(bucket, state, 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 @@ -167,19 +230,21 @@ 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, and the depleted-partition fast path keeps this lock-free while -every partition is full. +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 - (@atomic :monotonic bucket.depleted) == 0 && return 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_state, partition_state.capacity + 1) + _retry_partition_set_capacity!(bucket, partition_key, partition_state, partition_state.capacity + 1) return nothing finally unlock(bucket.lock) 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 3b7f5651c..aecf6c4c1 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,7 +1023,7 @@ 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()) + 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 @@ -1054,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, @@ -1064,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 @@ -1100,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, @@ -1116,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 @@ -1478,18 +1526,12 @@ 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 # Iterative cause-unwrapping (not recursion) keeps this resolvable for # trimmed static compilation. @@ -1507,6 +1549,7 @@ end # (e.g. an RST as `SystemError`). Classify by the cause so dead # reused connections are retried here instead of consuming the # caller's retry budget (#1353). + (current::TLS.TLSError).message == "unexpected EOF" && return true cause = (current::TLS.TLSError).cause cause === nothing && return false current = cause::Exception @@ -1651,12 +1694,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 @@ -1755,10 +1799,10 @@ 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 — up to -`max_idle_per_host + 1` acquisitions, enough to burn through a fully poisoned -pool and reach a fresh dial (#1353). A failure on a freshly dialed connection -propagates. +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. @@ -1773,163 +1817,161 @@ function _roundtrip_incoming!( attempt::Int=1, retry_template::Union{Nothing,Request}=nothing, ) - request_deadline = _request_deadline_ns(request) if retry_template === nothing && attempt == 1 && _retryable_request(request) retry_template = _copy_request(request) end - 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 - 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 <= transport.max_idle_per_host && 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, - retry_template, - ) - end - rethrow(err) end end diff --git a/test/http2_client_tests.jl b/test/http2_client_tests.jl index c3db8c9a2..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) @@ -462,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 812c1760d..a52abb169 100644 --- a/test/http_client_tests.jl +++ b/test/http_client_tests.jl @@ -2041,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 a17c71c8f..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,10 +1047,237 @@ 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 @@ -1075,6 +1323,12 @@ end @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( @@ -1085,6 +1339,92 @@ end ) 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 @@ -1161,6 +1501,102 @@ end 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 85b3417fc..360ea74df 100644 --- a/test/http_core_tests.jl +++ b/test/http_core_tests.jl @@ -325,6 +325,7 @@ end wrapped_tls_error = HT._wrap_client_transport_error(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))) diff --git a/test/http_retry_tests.jl b/test/http_retry_tests.jl index 883020dc4..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 @@ -185,6 +196,8 @@ end @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. @@ -197,7 +210,7 @@ end @testset "HTTP retry bucket replenishes consumed capacity (#1353)" begin bucket = HT.RetryBucket(capacity = 20) - @test (@atomic bucket.depleted) == 0 + @test isempty(@atomic :acquire bucket.depleted_partitions) # Replenish before any capacity was ever spent is a lock-free no-op and # creates no partitions. @@ -205,7 +218,7 @@ end @test isempty(bucket.partitions) token = Base.acquire(bucket, "svc.example") - @test (@atomic bucket.depleted) == 1 + @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 @@ -213,21 +226,44 @@ end HT._retry_bucket_replenish!(bucket, "svc.example") end @test bucket.partitions["svc.example"].capacity == 15 - @test (@atomic bucket.depleted) == 1 + @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 (@atomic bucket.depleted) == 0 + @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 (@atomic bucket.depleted) == 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 @@ -461,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 @@ -487,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 @@ -510,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( @@ -517,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 @@ -584,6 +893,21 @@ 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)) @@ -601,7 +925,7 @@ end # 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 (@atomic bucket.depleted) == 0 + @test isempty(@atomic :acquire bucket.depleted_partitions) finally HTTP.@try_ignore NC.close(listener) end @@ -767,6 +1091,17 @@ end @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) From 0726d936601d37ed94c7e526da4b58e7a6dbdf1f Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Fri, 28 Aug 2026 14:21:49 -0600 Subject: [PATCH 4/4] Centralize the TLS truncation-message classification Both retry classifiers matched Reseau's truncated-TLS-stream message with a duplicated string literal; hoist it to a single documented constant so a Reseau wording change is a one-line fix, and note that the end-to-end truncation tests pin the coupling. Also smooth the retry-budget CHANGELOG entry into readable prose. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 21 +++++++++++---------- src/http_client_retry.jl | 2 +- src/http_retry.jl | 8 ++++++++ src/http_transport.jl | 9 +++++---- 4 files changed, 25 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3444f3903..5fbce7d2a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,16 +31,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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`. Successful retries now refund their reservation. - Retry responses use the effective built-in or custom `retry_if` decision - while another configured retry remains. A terminal armed response uses the - built-in classification without invoking `retry_if` after retry slots are - exhausted. Retries explicitly requested by `retry_if` conservatively keep - cost on terminal non-success responses. - Each successful non-retried request restores one unit of - previously consumed budget. Retry reservations and response connections are - also released if a trace or retry-policy callback throws, and the request - deadline is rechecked after backoff sleep. ([#1353]) + 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 diff --git a/src/http_client_retry.jl b/src/http_client_retry.jl index ef4f7b509..b6df3dd24 100644 --- a/src/http_client_retry.jl +++ b/src/http_client_retry.jl @@ -71,7 +71,7 @@ function _retryable_request_error(err::Exception)::Bool continue end if current isa TLS.TLSError - (current::TLS.TLSError).message == "unexpected EOF" && return true + (current::TLS.TLSError).message == _TLS_TRUNCATED_STREAM_MESSAGE && return true cause = (current::TLS.TLSError).cause cause === nothing && return false current = cause::Exception diff --git a/src/http_retry.jl b/src/http_retry.jl index 61b050d6c..70c4f21a3 100644 --- a/src/http_retry.jl +++ b/src/http_retry.jl @@ -12,6 +12,14 @@ 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" diff --git a/src/http_transport.jl b/src/http_transport.jl index aecf6c4c1..cabd84662 100644 --- a/src/http_transport.jl +++ b/src/http_transport.jl @@ -1546,10 +1546,11 @@ const _retryable_method = _retryable_request_method 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`). Classify by the cause so dead - # reused connections are retried here instead of consuming the - # caller's retry budget (#1353). - (current::TLS.TLSError).message == "unexpected EOF" && return true + # (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