---
title: "Rate limiting with a token bucket gate"
description: "An SMS provider takes five messages a second: a token bucket gate in a stream lets five through, and the rest come back after their lease, in order."
---

> Queen MQ documentation, for AI agents
> Complete self-contained summary of Queen MQ: https://queenmq.com/llms-brief.txt
> Fetch that first when the question is about the product rather than about this page.
> Index of all pages: https://queenmq.com/llms.txt

# Rate limiting with a token bucket gate

An SMS provider accepts five messages a second for each customer. Send more and it refuses them, or blocks the account. The messages must still all go out, in order.

A stream with a gate decides, message by message, whether a message may go on now. The token bucket gate keeps its state per partition, in the broker: here five tokens per customer, refilled at five a second. A message the gate refuses stops its partition there, and comes back, with the ones after it, when its lease ends.

JavaScript:

```js
// The SMS provider accepts 5 messages a second for each customer.
// A refused message comes back when its lease ends: here after 1 second.
await queen.queue('sms-out').config({ leaseTime: 1 }).create()

const limiter = await Stream
  .from(queen.queue('sms-out'))                          // one partition per customer
  .gate(tokenBucketGate({ capacity: 5, refillPerSec: 5 }))  // 5 at once, then 5 a second
  .foreach(async (sms) => sendSms(sms))                  // only what the gate lets through
  .run({ queryId: 'sms-gate', url, subscriptionMode: 'all' })

// Twelve messages for one customer, all at once.
await queen.queue('sms-out').partition('customer-42')
  .push(Array.from({ length: 12 }, (_, i) => ({ data: { n: i + 1, text: `#${i + 1}` } })))
```

Python:

```python
# The SMS provider accepts 5 messages a second for each customer.
# A refused message comes back when its lease ends: here after 1 second.
await queen.queue("sms-out").config({"leaseTime": 1}).create()

limiter = await (
    Stream.from_(queen.queue("sms-out"))                    # one partition per customer
    .gate(token_bucket_gate(capacity=5, refill_per_sec=5))  # 5 at once, then 5 a second
    .foreach(send_sms)                                      # only what the gate allows
    .run(query_id="sms-gate", url=url, subscription_mode="all")
)

# Twelve messages for one customer, all at once.
await queen.queue("sms-out").partition("customer-42").push(
    [{"data": {"n": i + 1, "text": f"#{i + 1}"}} for i in range(12)]
)
```

Go:

```go
// The SMS provider accepts 5 messages a second for each customer.
// A refused message comes back when its lease ends: here after 1 second.
_, err = client.Queue("sms-out").Config(queen.QueueConfig{LeaseTime: 1}).
	Create().Execute(ctx)
if err != nil {
	log.Fatal(err)
}

limiter, err := streams.
	From(client.Queue("sms-out").AsStreamSource()). // one partition per customer
	Gate(helpers.TokenBucketGate(helpers.TokenBucketGateOptions{
		Capacity: 5, RefillPerSec: 5, // 5 at once, then 5 a second
	})).
	Foreach(func(sms interface{}, _ streams.EmitCtx) error {
		return sendSms(sms) // only what the gate lets through
	}).
	Run(ctx, streams.RunOptions{QueryID: "sms-gate", URL: url,
		SubscriptionMode: queen.SubscriptionModeAll})
if err != nil {
	log.Fatal(err)
}

// Twelve messages for one customer, all at once.
burst := make([]interface{}, 12)
for i := range burst {
	burst[i] = map[string]interface{}{"n": i + 1, "text": fmt.Sprintf("#%d", i+1)}
}
_, err = client.Queue("sms-out").Partition("customer-42").Push(burst).Execute(ctx)
if err != nil {
	log.Fatal(err)
}
```

Rust:

```rust
use queen_mq::streams::{GateCtx, Record};

// The Rust client has no tokenBucketGate helper; this is the same bucket. Its
// state is kept per partition in the broker, and saved only when a message passes.
fn token_bucket_gate(
    capacity: f64,
    refill_per_sec: f64,
) -> impl Fn(&Record, &mut GateCtx<'_>) -> bool + Send + Sync + 'static {
    move |_msg, ctx| {
        let now = ctx.stream_time_ms as f64;
        let elapsed = ((now - ctx.num("lastRefillAt", now)) / 1000.0).max(0.0);
        let tokens = (ctx.num("tokens", capacity) + elapsed * refill_per_sec).min(capacity);
        ctx.set_num("lastRefillAt", now);
        if tokens < 1.0 {
            return false; // refused: it comes back, in order, when its lease ends
        }
        ctx.set_num("tokens", tokens - 1.0);
        true
    }
}

// The SMS provider accepts 5 messages a second for each customer.
// A refused message comes back when its lease ends: here after 1 second.
queen
    .queue("sms-out")
    .configure(QueueOptions { lease_time: Some(1), ..Default::default() })
    .await?;

let limiter = Stream::from(queen.queue("sms-out")) // one partition per customer
    .gate(token_bucket_gate(5.0, 5.0))      // 5 at once, then 5 a second
    .foreach(|sms, _| async move { send_sms(sms).await }) // only what the gate lets through
    .run(&queen, RunOptions::new("sms-gate").subscription_mode(SubscriptionMode::All))
    .await?;

// Twelve messages for one customer, all at once.
queen
    .queue("sms-out")
    .partition("customer-42")
    .push_many((1..=12).map(|n| json!({ "n": n, "text": format!("#{n}") })))
    .await?;
```

PHP: PHP has no streams client. A stream reads any queue, whoever pushes to it, so run this step from JavaScript, Python, Go or Rust.

C++: C++ has no streams client. A stream reads any queue, whoever pushes to it, so run this step from JavaScript, Python, Go or Rust.

Five messages went at once, and the rest followed at about five a second, so the twelfth left about two seconds after the first: all twelve, in order.

**Figure.** Twelve SMS for customer-42 pushed at once. The gate starts with 5 tokens. At 0 seconds messages 1 to 5 pass; message 6 is refused, which stops the partition: 6 to 12 wait for the lease, 1 second. At 1 second the bucket has refilled 5 tokens (4.99 if the lease ends a few milliseconds early, and then only 4 pass) and messages 6 to 10 pass; 11 is refused. At 2 seconds 11 and 12 pass. All twelve go out, in order, 5 a second.

The token bucket gate: 5, 5, then 2, in order. A refill a few milliseconds short lets 4 through, not 5.

**Why:** the bucket lives in the broker, per customer, and survives restarts: no Redis next to it. Waiting messages stay in their queue, in order, and never need a queue of your own.

One thing to know: the refused messages wait for their lease, so size the two together. The most a gate can let through is the smaller of `refillPerSec` and `capacity` divided by the lease in seconds, so keep `capacity` near `refillPerSec` times the lease.

Built from: [Streams](/writing-applications/streams/#rate-limits-with-a-gate), [Queues](/writing-applications/queues/) (the lease).

Source: https://queenmq.com/patterns/rate-limiting/index.mdx