Rate limiting

Throttle outbound work with a token-bucket RateLimiter — a sustained rate plus a burst allowance, shared by reference so several callers run under one budget.

Version

v 1.0.14

The rate-limiting module (cocoindex::resources::rate_limit) provides a token-bucket RateLimiter for throttling outbound work — typically the API calls a pipeline makes to an external service — so it stays within that service’s rate limit.

use cocoindex::resources::rate_limit::RateLimiter; // also in cocoindex::prelude

RateLimiter

A token-bucket limiter. Acquire tokens with limiter.acquire(n).await?; the call returns once n tokens are available, otherwise it waits.

impl RateLimiter {
    pub fn new(max_rows_per_second: f64, burst_window: Duration) -> Result<Self>;
    pub async fn acquire(&self, n: u32) -> Result<()>;
}

Constructor parameters:

• max_rows_per_second — the sustained rate, in tokens (think: units of work) per second. Fractional rates are supported (e.g. 2.5 → one token every 0.4s). Must be positive.

• burst_window — the burst allowance, as a Duration. Up to max_rows_per_second × burst_window (in seconds) tokens may accumulate while the limiter is idle, letting a sudden fan-out proceed immediately before settling to the sustained rate. Duration::from_secs(1) gives a one-second burst; Duration::ZERO gives no burst (strict spacing between tokens).

acquire(n) — wait until n tokens are available, then consume them. A single RateLimiter is safe to share across concurrent callers; they are served in FIFO order, so no caller can be starved.

let limiter = RateLimiter::new(20.0, Duration::from_secs(1))?;

// Pace outbound calls at ~20/s (with a 1s burst).
limiter.acquire(1).await?; // one token

One budget per instance

The budget lives in the RateLimiter object, not in its settings. So the unit of sharing is the instance: functions and components handed the same RateLimiter draw from one shared budget, while two separately-constructed limiters are two independent budgets — even with identical max_rows_per_second.

RateLimiter is Clone, and every clone shares the one bucket. Cloning is the way to hand a limiter to several closures; it does not create a second budget.

The idiomatic way to share one budget is to bind a RateLimiter to a context key and retrieve it with ctx.get_key wherever you make outbound calls. Every component then draws from the same budget, with no limiter to thread through each signature:

use std::time::Duration;

use cocoindex::prelude::*;

cocoindex::context_key!(static RATE_LIMIT: RateLimiter);

#[cocoindex::lifespan]
async fn lifespan(builder: EnvironmentBuilder) -> Result<EnvironmentBuilder> {
    Ok(builder.provide_key(&RATE_LIMIT, RateLimiter::new(50.0, Duration::from_secs(1))?))
}

#[cocoindex::function]
async fn fetch_one(ctx: &Ctx, item: Item) -> Result<Data> {
    ctx.get_key(&RATE_LIMIT)?.acquire(1).await?;
    call_external_api(item).await
}

Because the key resolves to one shared instance, every acquire call across the app draws from the same 50/s budget.

Pacing a batched function

A batched function whose every batch is one request can pace itself on the limiter: call acquire(1) at the start of the batched body. The body runs once per batch, so a token pays for one batch rather than one call, and a scarce quota is spent on batches rather than on single items.

When to use it

Reach for a RateLimiter whenever a pipeline makes outbound calls to a rate-limited service — an API you query from a #[cocoindex::function], an embedding or LLM endpoint, a bucket or database with a request cap — and you want the whole pipeline to stay within one budget rather than each call site guessing at its own sleep. Because a fan-out (e.g. one component per source item) can otherwise issue requests as fast as the engine schedules them, a shared limiter is the simplest way to keep concurrent components cooperatively under quota.