///|
/// REST rate limiting strategy.
///
/// The design is requester-driven: `acquire` blocks (asynchronously) in the
/// calling task until the request may proceed, and `release` must be called
/// exactly once per successful `acquire`, with the response's rate-limit
/// headers when available. There is no background actor task, so a cancelled
/// caller cleans up naturally.
///
/// Implementations other than `InMemoryRateLimiter` (e.g. a cross-process
/// store) can be plugged into the HTTP client via this trait.
pub(open) trait RateLimiter {
/// Wait until a request for `bucket_key` may be sent. `bucket_key` is the
/// client's route-level bucket (method + path template + major parameter).
/// `global_exempt` skips the global request-per-second limit (interaction
/// and webhook endpoints are exempt per Discord's documentation).
async fn acquire(Self, bucket_key : String, global_exempt~ : Bool) -> Unit
/// Report the outcome of the request admitted by the matching `acquire`.
/// `headers` must have lowercase header names; pass an empty map when the
/// request failed before a response arrived (`status=0`).
async fn release(
Self,
bucket_key : String,
status~ : Int,
headers~ : Map[String, String],
) -> Unit
}
///|
priv struct Bucket {
/// Serializes requests within one bucket: held from admission until the
/// response headers are reported. This keeps bucket accounting exact at the
/// cost of per-bucket sequencing (different major parameters get different
/// buckets, so cross-channel/guild parallelism is preserved).
gate : @async.Semaphore
mut known : Bool
mut remaining : Int
mut reset_at_ms : Int64
}
///|
/// In-process `RateLimiter`: per-bucket accounting learned from response
/// headers plus a global sliding-window limit (default 50 requests/second).
pub struct InMemoryRateLimiter {
priv buckets : Map[String, Bucket]
global_limit : Int
mut global_window_start : Int64
mut global_count : Int
mut global_blocked_until : Int64
}
///|
/// Create a limiter with empty bucket accounting and a global window of
/// `global_limit` requests per second.
pub fn InMemoryRateLimiter::InMemoryRateLimiter(
global_limit? : Int = 50,
) -> InMemoryRateLimiter {
{
buckets: Map([]),
global_limit,
global_window_start: 0,
global_count: 0,
global_blocked_until: 0,
}
}
///|
async fn InMemoryRateLimiter::wait_for_global(
self : InMemoryRateLimiter,
) -> Unit {
for ;; {
let now = @clock.now_ms()
if now < self.global_blocked_until {
@async.sleep((self.global_blocked_until - now).to_int() + 1)
continue
}
if now - self.global_window_start >= 1000L {
self.global_window_start = now
self.global_count = 0
}
if self.global_count < self.global_limit {
self.global_count += 1
break
}
let wait = self.global_window_start + 1000L - now
@async.sleep(wait.to_int() + 1)
}
}
///|
/// Wait for bucket capacity — and the global window unless `global_exempt` —
/// before the request is sent.
pub impl RateLimiter for InMemoryRateLimiter with fn acquire(
self,
bucket_key,
global_exempt~,
) {
if !global_exempt {
self.wait_for_global()
}
let bucket = match self.buckets.get(bucket_key) {
Some(b) => b
None => {
let b = Bucket::{
gate: Semaphore(1),
known: false,
remaining: 0,
reset_at_ms: 0,
}
self.buckets[bucket_key] = b
b
}
}
bucket.gate.acquire()
// From here on the gate is held; make sure a cancellation while waiting
// for the bucket reset releases it.
errdefer bucket.gate.release()
if bucket.known && bucket.remaining <= 0 {
let wait = bucket.reset_at_ms - @clock.now_ms()
if wait > 0 {
@async.sleep(wait.to_int() + 1)
}
// optimistic: the window has reset; the next response corrects us
bucket.remaining = 1
}
}
///|
/// Feed the response's `x-ratelimit-*` headers back into the bucket
/// accounting.
pub impl RateLimiter for InMemoryRateLimiter with fn release(
self,
bucket_key,
status~,
headers~,
) {
guard self.buckets.get(bucket_key) is Some(bucket) else { return }
if headers.get("x-ratelimit-remaining") is Some(remaining_s) {
bucket.remaining = @string.parse_int(remaining_s) catch { _ => 0 }
bucket.known = true
}
// Prefer the relative reset to be robust against wall-clock jumps.
if headers.get("x-ratelimit-reset-after") is Some(after_s) {
let after = @string.parse_double(after_s) catch { _ => 0.0 }
bucket.reset_at_ms = @clock.now_ms() + (after * 1000.0).to_int64()
}
if status == 429 {
let retry_after_ms = match headers.get("retry-after") {
Some(s) => {
let seconds = @string.parse_double(s) catch { _ => 1.0 }
(seconds * 1000.0).to_int64()
}
None => 1000L
}
if headers.get("x-ratelimit-global") is Some(_) {
let blocked_until = @clock.now_ms() + retry_after_ms
if blocked_until > self.global_blocked_until {
self.global_blocked_until = blocked_until
}
} else {
bucket.known = true
bucket.remaining = 0
bucket.reset_at_ms = @clock.now_ms() + retry_after_ms
}
}
bucket.gate.release()
}