The Shared Rate Limit

The Shared Rate Limit: two plans run concurrently in one reactor, but the API key they share allows only 3 requests in flight. A single Async::Semaphore, passed to both, enforces the provider's ceiling across plan boundaries - because rate limits belong to credentials, not to orchestrators.

source on github

a real captured run

SHARED RATE LIMIT: two plans, one credential, ceiling 3

  plan 1: completed, 8 tasks
  plan 2: completed, 8 tasks

  wall time: 346ms for 16 calls of ~60ms each
  in-flight high-water mark: 3 (ceiling 3) - held
  calls interleaved across plans: yes

each orchestrator had concurrency_limit 10; the credential said 3.
the credential won, across both plans, because the semaphore lives
with the resource it protects - not with either scheduler.

source

# frozen_string_literal: true

# The Shared Rate Limit: two plans run concurrently in one reactor, but
# the API key they share allows only 3 requests in flight. A single
# Async::Semaphore, passed to both, enforces the provider's ceiling
# across plan boundaries - because rate limits belong to credentials,
# not to orchestrators.
#
#   bundle exec ruby examples/shared_rate_limit.rb
#
# Runs offline; the proof is the high-water mark.

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

API_CEILING = 3

# The shared credential: Agentic::RateLimit (this example's original
# feature request, granted) plus a call log for the interleaving proof
class RateLimitedApi
  attr_reader class="y">:limit

  def initialize(ceiling)
    @limit = Agentic:class="y">:RateLimit.new(ceiling)
    @calls = []
  end

  def high_water = @limit.high_water

  def call(plan, name, latency)
    @limit.acquire do
      @calls << class="s">"#{plan}/#{name}"
      sleep(latency)
      class="s">"#{name}class="y">:ok"
    end
  end

  def interleaved?
    plans = @calls.map { |c| c.split(class="s">"/").first }
    plans.uniq.size > 1 && plans != plans.sort
  end
end

def build_plan(label, task_count, api)
  orchestrator = Agentic:class="y">:PlanOrchestrator.new(concurrency_limit: 10)
  task_count.times do |i|
    orchestrator.add_task(Agentic:class="y">:Task.new(
      description: class="s">"#{label}-#{i}",
      agent_spec: {class="s">"name" => label, class="s">"instructions" => class="s">"call the API"},
      payload: 0.04 + (i % 3) * 0.02
    ), agent: ->(t) { api.call(label, t.description, t.payload) })
  end
  orchestrator
end

api = RateLimitedApi.new(API_CEILING)

puts class="s">"SHARED RATE LIMIT: two plans, one credential, ceiling #{API_CEILING}"
puts

wall = nil
Sync do
  ingest = build_plan(class="s">"ingest", 8, api)
  enrich = build_plan(class="s">"enrich", 8, api)

  started = Process.clock_gettime(Process:class="y">:CLOCK_MONOTONIC)

  # Both plans run as siblings in this reactor; each would happily use
  # 10 slots, but the shared semaphore is the credential's law
  plans = [ingest, enrich].map do |orchestrator|
    Async { orchestrator.execute_plan }
  end
  results = plans.map(&class="y">:wait)
  wall = Process.clock_gettime(Process:class="y">:CLOCK_MONOTONIC) - started

  results.each_with_index do |result, i|
    puts format(class="s">"  plan %d: %s, %d tasks", i + 1, result.status,
      result.results.count { |_, r| r.successful? })
  end
end

puts
puts format(class="s">"  wall time: %dms for 16 calls of ~60ms each", wall * 1000)
puts format(class="s">"  in-flight high-water mark: %d (ceiling %d) %s",
  api.high_water, API_CEILING, (api.high_water <= API_CEILING) ? class="s">"- held" : class="s">"- BREACHED")
puts class="s">"  calls interleaved across plans: #{api.interleaved? ? "yesclass="s">" : "noclass="s">"}"
puts
puts class="s">"each orchestrator had concurrency_limit 10; the credential said 3."
puts class="s">"the credential won, across both plans, because the semaphore lives"
puts class="s">"with the resource it protects - not with either scheduler."