Rate limiter for external APIs.

// npm install queen-mq        (Node 22+)
import { Queen, Stream, tokenBucketGate } from 'queen-mq'

const url = 'http://localhost:6632'
const q = new Queen({ url, handleSignals: false })

await q.queue('rl.requests').config({ leaseTime: 2, retryLimit: 1000 }).create()
await q.queue('rl.approved').create()

// Push 30 messages on partition 'tenant-A' as a burst.
await q.queue('rl.requests').partition('tenant-A').push(
  Array.from({ length: 30 }, (_, i) => ({ data: { seq: i } }))
)

// Token bucket: burst 10, refill 5/sec.
const stream = await Stream
  .from(q.queue('rl.requests'))
  .gate(tokenBucketGate({ capacity: 10, refillPerSec: 5 }))
  .to(q.queue('rl.approved'))
  .run({
    queryId:   'docs-rl-quickstart',
    url,
    batchSize: 10,
    reset:     true
  })

// Drain the approved queue and observe arrival order + timing.
const t0 = Date.now()
const arrivals = []
const cg = 'docs-rl-quickstart-cg'
while (arrivals.length < 30) {
  const batch = await q.queue('rl.approved')
    .group(cg).batch(10).wait(true).timeoutMillis(2000).pop()
  if (!batch || batch.length === 0) continue
  for (const m of batch) arrivals.push({ seq: m.data.seq, at: Date.now() - t0 })
  await q.ack(batch, true, { group: cg })
}
await stream.stop()

console.log('first 10 ms:', arrivals.slice(0, 10).map(a => a.at).join(' '))
console.log('last  10 ms:', arrivals.slice(-10).map(a => a.at).join(' '))
// Output (verified):
//   first 10 ms: 128 128 128 128 128 128 128 128 128 128
//   last  10 ms: 5961 5961 5961 5961 5961 5961 5961 5961 5961 5961
//   30 messages, all in seq order [0..29].

# pip install queen-mq        (Python 3.10+)
import asyncio
from queen import Queen, Stream, token_bucket_gate

URL = "http://localhost:6632"

async def main():
    async with Queen(url=URL) as q:
        await q.queue("rl.requests").config(lease_time=2, retry_limit=1000).create()
        await q.queue("rl.approved").create()

        # Push 30 messages on partition 'tenant-A' as a burst.
        await q.queue("rl.requests").partition("tenant-A").push(
            [{"data": {"seq": i}} for i in range(30)]
        )

        # Token bucket: burst 10, refill 5/sec.
        stream = await (
            Stream.from_(q.queue("rl.requests"))
            .gate(token_bucket_gate(capacity=10, refill_per_sec=5))
            .to(q.queue("rl.approved"))
            .run(query_id="docs-rl-quickstart-py", url=URL, batch_size=10, reset=True)
        )

        cg = "docs-rl-quickstart-cg"
        arrivals = []
        while len(arrivals) < 30:
            batch = await q.queue("rl.approved").group(cg).batch(10).wait(True).timeout_millis(2000).pop()
            if not batch:
                continue
            for m in batch:
                arrivals.append(m.get("data", {}).get("seq"))
            await q.ack(batch, True, {"group": cg})
        await stream.stop()
        print("seqs:", arrivals)
        # All 30 messages, in seq order [0..29].

asyncio.run(main())

// go get github.com/smartpricing/queen/clients/client-go
package main

import (
    "context"
    "fmt"

    queen "github.com/smartpricing/queen/clients/client-go"
    "github.com/smartpricing/queen/clients/client-go/streams"
    "github.com/smartpricing/queen/clients/client-go/streams/helpers"
)

func main() {
    url := "http://localhost:6632"
    q, _ := queen.New(url)
    defer q.Close(context.Background())

    _, _ = q.Queue("rl.requests").Config(queen.QueueConfig{
        LeaseTime: 2, RetryLimit: 1000,
    }).Create().Execute(context.Background())
    _, _ = q.Queue("rl.approved").Create().Execute(context.Background())

    // Push 30 messages on partition 'tenant-A' as a burst.
    for i := 0; i < 30; i++ {
        _, _ = q.Queue("rl.requests").Partition("tenant-A").Push(
            map[string]interface{}{"seq": i},
        ).Execute(context.Background())
    }

    // Token bucket: burst 10, refill 5/sec.
    gateFn := helpers.TokenBucketGate(helpers.TokenBucketGateOptions{
        Capacity: 10, RefillPerSec: 5,
    })
    stream := streams.From(q.Queue("rl.requests").AsStreamSource()).
        Gate(gateFn).
        To(q.Queue("rl.approved"))

    runner, _ := stream.Run(context.Background(), streams.RunOptions{
        QueryID: "docs-rl-quickstart-go", URL: url, BatchSize: 10, Reset: true,
    })

    cg := "docs-rl-quickstart-cg"
    drained := 0
    for drained < 30 {
        batch, _ := q.Queue("rl.approved").Group(cg).Batch(10).Wait(true).TimeoutMillis(2000).Pop(context.Background())
        if len(batch) == 0 { continue }
        for _, m := range batch {
            fmt.Println("seq:", m.Data["seq"])
        }
        drained += len(batch)
        _, _ = q.Ack(context.Background(), batch, true, queen.AckOptions{ConsumerGroup: cg})
    }
    runner.Stop()
}

import { Queen, Stream, tokenBucketGate } from 'queen-mq'

const q = new Queen({ url: 'http://localhost:6632', handleSignals: false })

await q.queue('rl.cost.requests').config({ leaseTime: 2, retryLimit: 1000 }).create()
await q.queue('rl.cost.approved').create()

// 100 requests, weight 1..5 each (deterministic for reproducibility).
const items = Array.from({ length: 100 }, (_, i) => ({
  data: { seq: i, weight: (i % 5) + 1 }
}))
await q.queue('rl.cost.requests').partition('tenant-A').push(items)

const stream = await Stream
  .from(q.queue('rl.cost.requests'))
  .gate(tokenBucketGate({
    capacity:     100,
    refillPerSec: 50,
    costFn:       msg => msg.weight        // ★ cost varies per request
  }))
  .to(q.queue('rl.cost.approved'))
  .run({ queryId: 'docs-rl-cost', url: 'http://localhost:6632', batchSize: 30, reset: true })

// Verified: drained 100 requests, 300 total weight in 5.71s
//           weight rate sustained: 52.6 weight/sec (target 50)
//           order intact: true

import { Queen, Stream, tokenBucketGate } from 'queen-mq'

const q = new Queen({ url: 'http://localhost:6632', handleSignals: false })

await q.queue('rl.cascade.src').config({ leaseTime: 2, retryLimit: 1000 }).create()
await q.queue('rl.cascade.mid').config({ leaseTime: 2, retryLimit: 1000 }).create()
await q.queue('rl.cascade.ok').create()

// 60 requests, each carrying 5 items.
const items = Array.from({ length: 60 }, (_, i) => ({
  data: {
    seq: i,
    items: Array.from({ length: 5 }, (_, j) => ({ id: `${i}-${j}` }))
  }
}))
await q.queue('rl.cascade.src').partition('tenant-A').push(items)

// Gate 1: 10 req/sec
const gate1 = await Stream
  .from(q.queue('rl.cascade.src'))
  .gate(tokenBucketGate({
    capacity:     20,
    refillPerSec: 10,
    costFn:       () => 1                     // 1 token per request
  }))
  .to(q.queue('rl.cascade.mid'))
  .run({ queryId: 'docs-rl-casc-1', url: 'http://localhost:6632',
         batchSize: 50, reset: true })

// Gate 2: 30 msg/sec (cost = items.length)
const gate2 = await Stream
  .from(q.queue('rl.cascade.mid'))
  .gate(tokenBucketGate({
    capacity:     60,
    refillPerSec: 30,
    costFn:       msg => msg.items.length     // weight by batch size
  }))
  .to(q.queue('rl.cascade.ok'))
  .run({ queryId: 'docs-rl-casc-2', url: 'http://localhost:6632',
         batchSize: 50, reset: true })

// Verified: 60 requests in 10.00s
//           req rate: 6.0/sec (gate1 limit 10)   ← gate2 is binding
//           msg rate: 30.0/sec (gate2 limit 30)
//           order intact: true

import { Queen, Stream, slidingWindowGate } from 'queen-mq'

const q = new Queen({ url: 'http://localhost:6632', handleSignals: false })

// IMPORTANT: leaseTime should be much smaller than the window (rule of thumb: ×5).
await q.queue('rl.sw.requests').config({ leaseTime: 1, retryLimit: 1000 }).create()
await q.queue('rl.sw.approved').create()

await q.queue('rl.sw.requests').partition('tenant-A').push(
  Array.from({ length: 60 }, (_, i) => ({ data: { seq: i } }))
)

const stream = await Stream
  .from(q.queue('rl.sw.requests'))
  .gate(slidingWindowGate({ limit: 20, windowSec: 5 }))   // max 20 in any 5s window
  .to(q.queue('rl.sw.approved'))
  .run({ queryId: 'docs-rl-sw', url: 'http://localhost:6632',
         batchSize: 50, reset: true })

// Verified: drained 60 messages in 12.41s
//           max arrivals in any 5s window: 29 (limit 20)
//           sustained rate: 4.75 msg/sec (target 4 = 20/5)
//           order intact: true

// Same gate definition, run N times. Only the FIRST runner uses reset:true
// so a re-run of the program starts clean. The rest attach with reset:false
// so they don't wipe state the first one already started building.
async function spawnRateLimiter({ runners, queryId, sourceQueue, sinkQueue,
                                 capacity, refillPerSec, costFn }) {
  const url = 'http://localhost:6632'
  const q = new Queen({ url, handleSignals: false })

  const buildStream = () => Stream
    .from(q.queue(sourceQueue))
    .gate(tokenBucketGate({ capacity, refillPerSec, costFn }))
    .to(q.queue(sinkQueue))

  const streams = []
  streams.push(await buildStream().run({
    queryId, url, batchSize: capacity, maxPartitions: 16, reset: true
  }))
  for (let i = 1; i < runners; i++) {
    streams.push(await buildStream().run({
      queryId, url, batchSize: capacity, maxPartitions: 16, reset: false
    }))
  }
  return streams
}

// Spawn 4 runners locally; throughput scales ~linearly.
const streams = await spawnRateLimiter({
  runners:      4,
  queryId:      'ota-rate-limiter.v1',
  sourceQueue:  'ota.requests',
  sinkQueue:    'ota.approved',
  capacity:     200,
  refillPerSec: 100,
  costFn:       () => 1
})

-- All keys for a query, sorted by remaining tokens (lowest = closest to throttle).
SELECT
    s.key                         AS tenant_or_partition,
    (s.value->>'tokens')::float   AS tokens_remaining,
    (s.value->>'allowedTotal')::int AS lifetime_allowed,
    (s.value->>'consumedTotal')::int AS lifetime_cost,
    s.updated_at
FROM queen_streams.state s
JOIN queen_streams.queries q ON q.id = s.query_id
WHERE q.name = 'ota-rate-limiter.v1'
ORDER BY tokens_remaining ASC
LIMIT 20;

SELECT s.key, (s.value->>'tokens')::float AS tokens
FROM queen_streams.state s
JOIN queen_streams.queries q ON q.id = s.query_id
WHERE q.name = 'ota-rate-limiter.v1'
  AND (s.value->>'tokens')::float < 1
ORDER BY s.updated_at DESC;