Fix retry bucket corruption on Julia 1.10

.github/workflows/ci.yml

@@ -26,6 +26,7 @@ jobs:
       fail-fast: false
       matrix:
         version:
+          - '1.10'
           - '1'
           - 'pre'
         os:

Project.toml

@@ -1,6 +1,6 @@
 name = "HTTP"
 uuid = "cd3eb016-35fb-5094-929b-558a96fad6f3"
-version = "2.6.6"
+version = "2.6.7"
 authors = ["Jacob Quinn", "contributors: https://github.com/JuliaWeb/HTTP.jl/graphs/contributors"]
 
 [deps]

src/http_retry.jl

@@ -61,10 +61,11 @@ mutable struct RetryBucket
     capacity::Int
     partitions::Dict{String,_RetryPartition}
     lock::ReentrantLock
-    # Copy-on-write snapshot of partition keys below full capacity. Published
-    # snapshots are treated as immutable. Writers publish under `lock`; readers
-    # use the snapshot to avoid locking for healthy traffic to unrelated keys.
-    @atomic depleted_partitions::Set{String}
+    # Number of partition keys below full capacity. Writers update it under
+    # `lock`; readers use it to avoid locking while every partition is full.
+    # Keep this field pointer-free: atomic references to GC-managed containers
+    # can corrupt the referenced object on Julia 1.10 (#1355).
+    @atomic depleted_partitions::Int
 end
 
 """Handle returned by `acquire` and consumed by `release` to refund retry budget."""
@@ -105,7 +106,7 @@ function RetryBucket(;
         Int(capacity),
         Dict{String,_RetryPartition}(),
         ReentrantLock(),
-        Set{String}(),
+        0,
     )
 end
 
@@ -118,9 +119,7 @@ function RetryBucket(
     partitions::Dict{String,_RetryPartition},
     lock::ReentrantLock,
 )
-    depleted_partitions = Set(
-        key for (key, state) in partitions if state.capacity < capacity
-    )
+    depleted_partitions = count(state -> state.capacity < capacity, values(partitions))
     return RetryBucket(
         backoff_scale_factor_ms,
         max_backoff_secs,
@@ -131,6 +130,27 @@ function RetryBucket(
     )
 end
 
+# Preserve the six-field constructor exposed in HTTP 2.6.6. The set argument
+# is accepted only for compatibility; the pointer-free depleted-partition
+# count is derived from `partitions` so the count invariant holds even when
+# the caller's set disagrees with the partition states.
+function RetryBucket(
+    backoff_scale_factor_ms::Int,
+    max_backoff_secs::Int,
+    capacity::Int,
+    partitions::Dict{String,_RetryPartition},
+    lock::ReentrantLock,
+    ::Set{String},
+)
+    return RetryBucket(
+        backoff_scale_factor_ms,
+        max_backoff_secs,
+        capacity,
+        partitions,
+        lock,
+    )
+end
+
 function RetryBucket(
     backoff_scale_factor_ms,
     max_backoff_secs,
@@ -147,25 +167,23 @@ function RetryBucket(
     )
 end
 
-# Set a partition's capacity while keeping the published depleted-key snapshot
-# in sync. Must be called with `bucket.lock` held.
+# Set a partition's capacity while keeping the depleted-partition count in sync.
+# Must be called with `bucket.lock` held.
 @inline function _retry_partition_set_capacity!(
     bucket::RetryBucket,
-    partition_key::String,
     state::_RetryPartition,
     new_capacity::Int,
 )::Nothing
     was_full = state.capacity >= bucket.capacity
     now_full = new_capacity >= bucket.capacity
     state.capacity = new_capacity
     if was_full && !now_full
-        depleted = copy(@atomic :acquire bucket.depleted_partitions)
-        push!(depleted, partition_key)
-        @atomic :release bucket.depleted_partitions = depleted
+        depleted = @atomic :monotonic bucket.depleted_partitions
+        @atomic :release bucket.depleted_partitions = depleted + 1
     elseif !was_full && now_full
-        depleted = copy(@atomic :acquire bucket.depleted_partitions)
-        delete!(depleted, partition_key)
-        @atomic :release bucket.depleted_partitions = depleted
+        depleted = @atomic :monotonic bucket.depleted_partitions
+        depleted > 0 || error("retry bucket depleted-partition count underflow")
+        @atomic :release bucket.depleted_partitions = depleted - 1
     end
     return nothing
 end
@@ -190,7 +208,7 @@ function acquire(bucket::RetryBucket, partition)
         if state.capacity < _RETRY_BUCKET_ACQUIRE_COST
             throw(RetryDeniedError(partition_key))
         end
-        _retry_partition_set_capacity!(bucket, partition_key, state, 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
@@ -222,7 +240,7 @@ end
         reserved = _retry_bucket_reserved_cost(token)
         consumed = min(reserved, max(0, failure_cost))
         refund = reserved - consumed
-        _retry_partition_set_capacity!(bucket, token.partition, state, min(bucket.capacity, state.capacity + refund))
+        _retry_partition_set_capacity!(bucket, state, min(bucket.capacity, state.capacity + refund))
         token.released = true
         return nothing
     finally
@@ -238,21 +256,20 @@ non-retried request, capped at the bucket's full capacity. This is the slow
 recovery path that lets a partition legitimately drained by a burst of real
 failures regain retry budget from healthy traffic instead of staying empty for
 the transport's lifetime. Partitions that have never spent capacity are left
-untouched. The published depleted-key snapshot also keeps this lock-free for
-healthy traffic to other partitions.
+untouched. The depleted-partition count keeps this lock-free while all
+partitions are healthy.
 """
 function _retry_bucket_replenish!(bucket::RetryBucket, partition)::Nothing
     depleted = @atomic :acquire bucket.depleted_partitions
-    isempty(depleted) && return nothing
+    depleted == 0 && return nothing
     partition_key = _retry_bucket_partition_key(partition)
-    partition_key in depleted || return nothing
     lock(bucket.lock)
     try
         state = get(() -> nothing, bucket.partitions, partition_key)
         state === nothing && return nothing
         partition_state = state::_RetryPartition
         partition_state.capacity >= bucket.capacity && return nothing
-        _retry_partition_set_capacity!(bucket, partition_key, partition_state, partition_state.capacity + 1)
+        _retry_partition_set_capacity!(bucket, partition_state, partition_state.capacity + 1)
         return nothing
     finally
         unlock(bucket.lock)

test/http_retry_tests.jl

@@ -116,7 +116,15 @@ end
     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"])
+    @test (@atomic :acquire positional.depleted_partitions) == 1
+
+    six_field = HT.RetryBucket(25, 20, 10, partitions, ReentrantLock(), Set(["depleted.example"]))
+    @test (@atomic :acquire six_field.depleted_partitions) == 1
+
+    # The six-field constructor derives the count from the partition states;
+    # a stale legacy set must not break the count invariant.
+    stale_set = HT.RetryBucket(25, 20, 10, partitions, ReentrantLock(), Set{String}())
+    @test (@atomic :acquire stale_set.depleted_partitions) == 1
 
     converted = HT.RetryBucket(Int32(25), Int16(20), Int8(10), copy(partitions), ReentrantLock())
     @test converted.backoff_scale_factor_ms === 25
@@ -210,38 +218,64 @@ end
 
 @testset "HTTP retry bucket replenishes consumed capacity (#1353)" begin
     bucket = HT.RetryBucket(capacity = 20)
-    @test isempty(@atomic :acquire bucket.depleted_partitions)
+    @test (@atomic :acquire bucket.depleted_partitions) == 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 :acquire bucket.depleted_partitions) == Set(["svc.example"])
+    @test (@atomic :acquire bucket.depleted_partitions) == 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 :acquire bucket.depleted_partitions) == Set(["svc.example"])
+    @test (@atomic :acquire bucket.depleted_partitions) == 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 isempty(@atomic :acquire bucket.depleted_partitions)
+    @test (@atomic :acquire bucket.depleted_partitions) == 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
-    @test (@atomic :acquire bucket.depleted_partitions) == Set(["other.example"])
+    @test (@atomic :acquire bucket.depleted_partitions) == 1
     Base.release(bucket, other, 0)
-    @test isempty(@atomic :acquire bucket.depleted_partitions)
+    @test (@atomic :acquire bucket.depleted_partitions) == 0
+end
+
+@testset "HTTP retry bucket uses a pointer-free concurrent fast path (#1355)" begin
+    bucket = HT.RetryBucket(capacity = 20)
+    @test isbitstype(fieldtype(HT.RetryBucket, :depleted_partitions))
+    workers = max(4, 2 * Threads.nthreads())
+    @sync begin
+        for worker in 1:workers
+            Threads.@spawn begin
+                key = "svc-$worker.example"
+                for _ in 1:2_000
+                    token = Base.acquire(bucket, key)
+                    Base.release(bucket, token, HT._RETRY_BUCKET_ACQUIRE_COST)
+                    for _ in 1:HT._RETRY_BUCKET_ACQUIRE_COST
+                        HT._retry_bucket_replenish!(bucket, key)
+                    end
+                end
+            end
+        end
+        Threads.@spawn for _ in 1:100
+            GC.gc(false)
+            yield()
+        end
+    end
+    @test all(state.capacity == bucket.capacity for state in values(bucket.partitions))
+    @test (@atomic :acquire bucket.depleted_partitions) == 0
 end
 
 @testset "HTTP retry bucket heals only after successful responses" begin
@@ -734,7 +768,7 @@ end
             end
             @test err === trace_err
             @test bucket.partitions["127.0.0.1"].capacity == 10
-            @test isempty(@atomic :acquire bucket.depleted_partitions)
+            @test (@atomic :acquire bucket.depleted_partitions) == 0
             @test lock(transport.lock) do
                 isempty(transport.conns_per_host)
             end
@@ -779,7 +813,7 @@ end
         end
         @test err === policy_err
         @test bucket.partitions["127.0.0.1"].capacity == 10
-        @test isempty(@atomic :acquire bucket.depleted_partitions)
+        @test (@atomic :acquire bucket.depleted_partitions) == 0
         @test lock(transport.lock) do
             isempty(transport.conns_per_host)
         end
@@ -925,7 +959,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 isempty(@atomic :acquire bucket.depleted_partitions)
+        @test (@atomic :acquire bucket.depleted_partitions) == 0
     finally
         HTTP.@try_ignore NC.close(listener)
     end

test/public_api_tests.jl

@@ -48,8 +48,11 @@ using HTTP
         @test true   # `public` unsupported before 1.11; nothing to assert
     end
 
-    # Internals must stay private regardless of Julia version.
-    @test !Base.ispublic(HTTP, :_retryable_request_error)
-    @test !Base.ispublic(HTTP, :_normalize_local_addr)
-    @test !Base.ispublic(HTTP.WebSockets, :_ws_mask_into!)
+    if isdefined(Base, :ispublic)
+        @test !Base.ispublic(HTTP, :_retryable_request_error)
+        @test !Base.ispublic(HTTP, :_normalize_local_addr)
+        @test !Base.ispublic(HTTP.WebSockets, :_ws_mask_into!)
+    else
+        @test true
+    end
 end