The Resize Torture Test

The Resize Torture Test: a feature that changes a limiter's ceiling while fibers are waiting on it had better say exactly what it guarantees - and then survive an attempt to break the guarantee. Three assaults: per-epoch ceilings under load, a mid-flight shrink, and a grow that must actually wake the waiters.

source on github

a real captured run

  epoch ceiling 1   observed max 1   held
  epoch ceiling 5   observed max 5   held
  epoch ceiling 2   observed max 2   held
  epoch ceiling 4   observed max 4   held
  epoch ceiling 1   observed max 1   held
  epoch ceiling 3   observed max 3   held

  shrink 5->2 mid-flight: wave-2 admissions saw max 2 concurrent (<= 2, held)
  grow 1->3 with 2 queued: last admission at 14ms (woken by resize, held)

  3 assaults, 0 cracks. the guarantees, as proven: an epoch's
  ceiling binds every admission inside it; shrinking drains rather
  than evicts, and nothing new is admitted above the new mark;
  growing wakes waiters immediately instead of leaving them on
  the old schedule. resize without these proofs is a data race
  with a friendly method name.

source

# frozen_string_literal: true

# The Resize Torture Test: a feature that changes a limiter's ceiling
# while fibers are waiting on it had better say exactly what it
# guarantees - and then survive an attempt to break the guarantee.
# Three assaults: per-epoch ceilings under load, a mid-flight shrink,
# and a grow that must actually wake the waiters.
#
#   bundle exec ruby examples/resize_torture.rb
#
# Runs offline; exits 1 if any guarantee cracks.

require class="s">"bundler/setup"
require class="s">"agentic"
require class="s">"async"

violations = []

# --- assault 1: every epoch's ceiling holds under saturating load -------------
# Resize through jagged ceilings; within each epoch, run far more jobs
# than lanes and record the max observed concurrency ourselves - we
# don't trust high_water, we recompute it.
limiter = Agentic:class="y">:RateLimit.new(1)
EPOCHS = [1, 5, 2, 4, 1, 3].freeze

Sync do
  EPOCHS.each do |ceiling|
    limiter.resize(ceiling)
    concurrent = 0
    observed_max = 0

    (ceiling * 4).times.map {
      Async do
        limiter.acquire do
          concurrent += 1
          observed_max = [observed_max, concurrent].max
          sleep(0.003)
          concurrent -= 1
        end
      end
    }.each(&class="y">:wait)

    if observed_max > ceiling
      violations << class="s">"epoch ceiling #{ceiling}: observed #{observed_max} concurrent"
    end
    puts format(class="s">"  epoch ceiling %-3d observed max %-3d %s",
      ceiling, observed_max, (observed_max > ceiling) ? class="s">"VIOLATED" : class="s">"held")
  end
end

# --- assault 2: shrink mid-flight admits nobody above the new mark ------------
# Fill 5 lanes, shrink to 2 while all 5 are running, then submit a
# second wave. Every wave-2 admission must see <= 2 concurrent
# (in-flight holders from wave 1 drain; nothing new joins them).
puts
shrinker = Agentic:class="y">:RateLimit.new(5)
wave2_snapshots = []

Sync do
  concurrent = 0
  wave1 = 5.times.map {
    Async do
      shrinker.acquire do
        concurrent += 1
        sleep(0.03)
        concurrent -= 1
      end
    end
  }
  Async do
    sleep(0.005) # let wave 1 occupy all 5 lanes
    shrinker.resize(2)
  end.wait

  wave2 = 5.times.map {
    Async do
      shrinker.acquire do
        concurrent += 1
        wave2_snapshots << concurrent
        sleep(0.003)
        concurrent -= 1
      end
    end
  }
  (wave1 + wave2).each(&class="y">:wait)
end

if wave2_snapshots.any? { |snapshot| snapshot > 2 }
  violations << class="s">"post-shrink admission saw #{wave2_snapshots.max} concurrent"
end
puts format(class="s">"  shrink 5->2 mid-flight: wave-2 admissions saw max %d concurrent %s",
  wave2_snapshots.max, (wave2_snapshots.any? { |s| s > 2 }) ? class="s">"VIOLATED" : class="s">"(<= 2, held)")

# --- assault 3: grow must wake the already-waiting -----------------------------
# One lane, three long jobs queued behind it. Grow to 3; the queued
# jobs must be admitted promptly, not on the old schedule.
grower = Agentic:class="y">:RateLimit.new(1)
admissions = []

Sync do
  t0 = Process.clock_gettime(Process:class="y">:CLOCK_MONOTONIC)
  jobs = 3.times.map {
    Async do
      grower.acquire do
        admissions << Process.clock_gettime(Process:class="y">:CLOCK_MONOTONIC) - t0
        sleep(0.05)
      end
    end
  }
  Async do
    sleep(0.01)
    grower.resize(3)
  end.wait
  jobs.each(&class="y">:wait)
end

# Serial schedule would admit job 3 at ~0.10s; waking on grow admits it ~0.01s
late = admissions.max
if late > 0.04
  violations << class="s">"grow did not wake waiters (last admission at #{(late * 1000).round}ms)"
end
puts format(class="s">"  grow 1->3 with 2 queued: last admission at %.0fms %s",
  late * 1000, (late > 0.04) ? class="s">"VIOLATED (serial schedule)" : class="s">"(woken by resize, held)")

puts
if violations.empty?
  puts class="s">"  3 assaults, 0 cracks. the guarantees, as proven: an epoch's"
  puts class="s">"  ceiling binds every admission inside it; shrinking drains rather"
  puts class="s">"  than evicts, and nothing new is admitted above the new mark;"
  puts class="s">"  growing wakes waiters immediately instead of leaving them on"
  puts class="s">"  the old schedule. resize without these proofs is a data race"
  puts class="s">"  with a friendly method name."
else
  puts class="s">"  CRACKED: #{violations.join("; class="s">")}"
end
exit(violations.empty? ? 0 : 1)