Beyond BullMQ's Built-In Rate Limiter: Custom Redis-Powered Throttling with Lua Scripts

BullMQ's built-in rate limiter (limiter.max / limiter.duration) is a solid queue-level backstop — a fixed-window counter that prevents your workers from processing more than N jobs per second. For many setups, that's sufficient.

But production queues face constraints that a single fixed-window counter cannot express: per-user API rate limits on Stripe, per-endpoint fairness for webhook delivery, Redis CPU spikes that demand workers slow down, and thundering herds when a rate limit lifts. In this post, we'll build custom Redis-based rate limiters that layer sophisticated policies alongside BullMQ — not replacing its built-in limiter, but augmenting it with fine-grained control.

The Gap: Why One Limiter Isn't Enough

BullMQ's built-in rate limiter is a queue-level, fixed-window counter. It answers: "How many jobs per second for this queue?" Real-world queue infrastructure needs finer questions answered:

• "How many Stripe API calls per tenant per second?"
• "What's the fair rate for this webhook endpoint over a rolling 60-second window?"
• "Should workers slow down because Redis CPU just hit 80%?"
• "How do we smoothly recover after a rate limit lifts without overwhelming the downstream?"

Each question requires a different strategy, and all of them benefit from Redis Lua scripts for atomic, race-condition-free operations across distributed worker instances.

1. Custom Token Bucket with Redis Lua

A token bucket allows controlled bursts up to a configured capacity while maintaining an average rate via periodic refill. Unlike BullMQ's fixed window, it scopes naturally to any dimension: per tenant, per downstream API, per job type.

The Lua Script

-- token_bucket.lua
-- KEYS[1]   = bucket key (e.g., "ratelimit:api:stripe:{tenantId}")
-- ARGV[1]   = max capacity
-- ARGV[2]   = refill rate (tokens per refill interval)
-- ARGV[3]   = refill interval in seconds
-- ARGV[4]   = current timestamp in seconds
-- Returns   = { allowed: 0|1, remaining: number, retryAfterMs: number }

local key          = KEYS[1]
local capacity     = tonumber(ARGV[1])
local refillRate   = tonumber(ARGV[2])
local refillSec    = tonumber(ARGV[3])
local now          = tonumber(ARGV[4])

local fields = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens      = tonumber(fields[1])
local lastRefill  = tonumber(fields[2])

if tokens == nil then
    tokens = capacity
    lastRefill = now
end

local elapsed = now - lastRefill
local refillCount = math.floor(elapsed / refillSec)

if refillCount > 0 then
    tokens = math.min(capacity, tokens + (refillCount * refillRate))
    lastRefill = lastRefill + (refillCount * refillSec)
end

local allowed = 0
local retryAfter = 0

if tokens >= 1 then
    tokens = tokens - 1
    allowed = 1
else
    local tokensNeeded = 1 - tokens
    local secondsUntilRefill = refillSec * math.ceil(tokensNeeded / refillRate)
    retryAfter = math.ceil(secondsUntilRefill * 1000)
end

redis.call('HMSET', key, 'tokens', tostring(tokens), 'last_refill', tostring(lastRefill))

local drainTime = math.ceil(capacity / refillRate) * refillSec
redis.call('EXPIRE', key, math.max(60, drainTime + 3600))

return { allowed, tokens, retryAfter }

TypeScript Wrapper

import { createClient, RedisClientType } from 'redis';
import * as fs from 'fs';
import * as path from 'path';

interface TokenBucketConfig {
  redisClient: RedisClientType;
  capacity: number;
  refillRate: number;
  refillIntervalSeconds: number;
  keyPrefix?: string;
}

interface TokenBucketResult {
  allowed: boolean;
  remaining: number;
  retryAfterMs: number;
}

export class TokenBucketRateLimiter {
  private client: RedisClientType;
  private capacity: number;
  private refillRate: number;
  private refillIntervalSeconds: number;
  private keyPrefix: string;
  private scriptSha: string | null = null;
  private scriptSource: string;

  constructor(config: TokenBucketConfig) {
    this.client = config.redisClient;
    this.capacity = config.capacity;
    this.refillRate = config.refillRate;
    this.refillIntervalSeconds = config.refillIntervalSeconds;
    this.keyPrefix = config.keyPrefix || 'ratelimit:token-bucket';
    this.scriptSource = fs.readFileSync(
      path.join(__dirname, 'token_bucket.lua'),
      'utf-8'
    );
  }

  async initialize(): Promise<void> {
    this.scriptSha = await this.client.scriptLoad(this.scriptSource);
  }

  private buildKey(scope: string): string {
    return `${this.keyPrefix}:${scope}`;
  }

  async allow(scope: string): Promise<TokenBucketResult> {
    const now = Date.now() / 1000;
    const key = this.buildKey(scope);

    const runScript = async (script: string, sha: string | null): Promise<(number | string)[]> => {
      if (sha) {
        try {
          return await this.client.evalSha(sha, {
            keys: [key],
            arguments: [
              String(this.capacity),
              String(this.refillRate),
              String(this.refillIntervalSeconds),
              String(now),
            ],
          });
        } catch {
          // Fall through to EVAL if SHA was evicted
        }
      }
      return await this.client.eval(script, {
        keys: [key],
        arguments: [
          String(this.capacity),
          String(this.refillRate),
          String(this.refillIntervalSeconds),
          String(now),
        ],
      });
    };

    const result = await runScript(this.scriptSource, this.scriptSha);
    return {
      allowed: Number(result[0]) === 1,
      remaining: Number(result[1]),
      retryAfterMs: Number(result[2]),
    };
  }
}

Integration with a BullMQ Worker

Here's how a Stripe API worker uses two token bucket layers — a global 100 req/s bucket matching Stripe's documented limit, and a per-tenant fairness bucket to prevent noisy tenants from starving others:

import { Worker, Job } from 'bullmq';

interface StripeJobData {
  tenantId: string;
  action: 'create_customer' | 'charge' | 'refund';
  payload: Record<string, unknown>;
}

const stripeGlobalLimiter = new TokenBucketRateLimiter({
  redisClient,
  capacity: 100,
  refillRate: 100,
  refillIntervalSeconds: 1,
  keyPrefix: 'ratelimit:stripe:global',
});

const tenantFairnessLimiter = new TokenBucketRateLimiter({
  redisClient,
  capacity: 10,
  refillRate: 10,
  refillIntervalSeconds: 1,
  keyPrefix: 'ratelimit:stripe:per-tenant',
});

await stripeGlobalLimiter.initialize();
await tenantFairnessLimiter.initialize();

const stripeWorker = new Worker<StripeJobData>(
  'stripe-operations',
  async (job: Job<StripeJobData>) => {
    const { tenantId } = job.data;

    // Layer 1: Per-tenant fairness — no single tenant monopolizes capacity
    const tenantCheck = await tenantFairnessLimiter.allow(tenantId);
    if (!tenantCheck.allowed) {
      await stripeWorker.rateLimit(tenantCheck.retryAfterMs);
      throw Worker.RateLimitError();
    }

    // Layer 2: Global Stripe API limit
    const apiCheck = await stripeGlobalLimiter.allow('stripe-api');
    if (!apiCheck.allowed) {
      await stripeWorker.rateLimit(apiCheck.retryAfterMs);
      throw Worker.RateLimitError();
    }

    await callStripeApi(job.data);
  },
  {
    connection: { host: 'localhost', port: 6379 },
    limiter: { max: 150, duration: 1000 },
    concurrency: 50,
  }
);

async function callStripeApi(data: StripeJobData): Promise<void> {
  // Actual Stripe API call
  console.log(`Processing ${data.action} for tenant ${data.tenantId}`);
}

Notice BullMQ's built-in limiter (max: 150) still acts as the third layer — a safety net that prevents total queue throughput from exceeding hardware capacity even if the custom token buckets have a bug.

2. Sliding Window Log with Redis Sorted Sets

BullMQ's built-in limiter is a fixed-window counter. The sliding window log is the fairest algorithm — because every request is individually timestamped, there are no boundary bursts. The trade-off is higher memory: O(requests) per key.

The Lua Script

-- sliding_window_log.lua
-- KEYS[1]   = log key (e.g., "ratelimit:sliding-log:webhook:{endpoint}")
-- ARGV[1]   = max requests allowed in window
-- ARGV[2]   = window duration in seconds
-- ARGV[3]   = current timestamp in milliseconds
-- ARGV[4]   = unique member identifier (e.g., "req:{uuid}")
-- Returns   = { allowed: 0|1, count: number, retryAfterMs: number }

local key           = KEYS[1]
local maxRequests   = tonumber(ARGV[1])
local windowSeconds = tonumber(ARGV[2])
local now           = tonumber(ARGV[3])
local member        = ARGV[4]

local windowStart = now - (windowSeconds * 1000)

redis.call('ZREMRANGEBYSCORE', key, 0, windowStart)

local currentCount = redis.call('ZCARD', key)

if currentCount < maxRequests then
    redis.call('ZADD', key, now, member)
    redis.call('EXPIRE', key, windowSeconds * 2)
    return { 1, currentCount + 1, 0 }
else
    local oldestEntries = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
    local oldestTimestamp = tonumber(oldestEntries[2])
    local retryAfter = math.max(0, oldestTimestamp - windowStart)
    return { 0, currentCount, math.ceil(retryAfter) }
end

TypeScript Wrapper

import { RedisClientType } from 'redis';
import * as fs from 'fs';
import * as path from 'path';
import { randomUUID } from 'crypto';

interface SlidingWindowConfig {
  redisClient: RedisClientType;
  maxRequests: number;
  windowSeconds: number;
  keyPrefix?: string;
}

interface SlidingWindowResult {
  allowed: boolean;
  currentCount: number;
  retryAfterMs: number;
}

export class SlidingWindowLogRateLimiter {
  private client: RedisClientType;
  private maxRequests: number;
  private windowSeconds: number;
  private keyPrefix: string;
  private scriptSha: string | null = null;
  private scriptSource: string;

  constructor(config: SlidingWindowConfig) {
    this.client = config.redisClient;
    this.maxRequests = config.maxRequests;
    this.windowSeconds = config.windowSeconds;
    this.keyPrefix = config.keyPrefix || 'ratelimit:sliding-log';
    this.scriptSource = fs.readFileSync(
      path.join(__dirname, 'sliding_window_log.lua'),
      'utf-8'
    );
  }

  async initialize(): Promise<void> {
    this.scriptSha = await this.client.scriptLoad(this.scriptSource);
  }

  private buildKey(scope: string): string {
    return `${this.keyPrefix}:${scope}`;
  }

  async check(scope: string): Promise<SlidingWindowResult> {
    const now = Date.now();
    const member = `req:${randomUUID()}`;
    const key = this.buildKey(scope);

    const result = await this.evalScript(key, [String(this.maxRequests), String(this.windowSeconds), String(now), member]);

    return {
      allowed: Number(result[0]) === 1,
      currentCount: Number(result[1]),
      retryAfterMs: Number(result[2]),
    };
  }

  private async evalScript(key: string, args: string[]): Promise<(number | string)[]> {
    if (this.scriptSha) {
      try {
        return await this.client.evalSha(this.scriptSha, { keys: [key], arguments: args });
      } catch {
        // SHA evicted — fall through
      }
    }
    return await this.client.eval(this.scriptSource, { keys: [key], arguments: args });
  }

  async getWindowCount(scope: string): Promise<number> {
    const now = Date.now();
    const windowStart = now - (this.windowSeconds * 1000);
    const key = this.buildKey(scope);
    await this.client.zRemRangeByScore(key, 0, windowStart);
    return await this.client.zCard(key);
  }
}

Worker Integration: Pre-Flight Check

interface WebhookJobData {
  endpointId: string;
  targetUrl: string;
  payload: unknown;
}

const endpointRateLimiter = new SlidingWindowLogRateLimiter({
  redisClient,
  maxRequests: 100,
  windowSeconds: 60,
  keyPrefix: 'ratelimit:webhook',
});

await endpointRateLimiter.initialize();

const webhookWorker = new Worker<WebhookJobData>(
  'webhook-delivery',
  async (job: Job<WebhookJobData>) => {
    const { endpointId } = job.data;

    const check = await endpointRateLimiter.check(endpointId);
    if (!check.allowed) {
      await webhookWorker.rateLimit(check.retryAfterMs);
      throw Worker.RateLimitError();
    }

    const response = await fetch(job.data.targetUrl, {
      method: 'POST',
      body: JSON.stringify(job.data.payload),
      headers: { 'Content-Type': 'application/json' },
    });

    if (response.status === 429) {
      const retryAfter = parseRetryAfter(response.headers.get('retry-after'));
      await webhookWorker.rateLimit(retryAfter);
      throw Worker.RateLimitError();
    }
  },
  {
    connection: { host: 'localhost', port: 6379 },
    limiter: { max: 1000, duration: 1000 },
    concurrency: 100,
  }
);

function parseRetryAfter(header: string | null): number {
  if (!header) return 5000;
  const seconds = parseInt(header, 10);
  return isNaN(seconds) ? 5000 : seconds * 1000;
}

3. Dynamic Adaptive Rate Limiting

Static rate limits assume the system's capacity never changes. In reality, Redis CPU spikes, downstream APIs slow down, and database connection pools fill up. An adaptive controller monitors these signals and self-regulates by adjusting BullMQ's global rate limit at runtime.

import { RedisClientType } from 'redis';
import { Worker } from 'bullmq';

interface AdaptiveControllerConfig {
  redisClient: RedisClientType;
  worker: Worker;
  minRateLimit: number;
  maxRateLimit: number;
  cooldownMs: number;
  signalThresholds: {
    redisCpuPercent: number;
    downstreamP99Ms: number;
    dbPoolPercent: number;
  };
}

export class AdaptiveRateController {
  private client: RedisClientType;
  private worker: Worker;
  private minRate: number;
  private maxRate: number;
  private cooldownMs: number;
  private thresholds: AdaptiveControllerConfig['signalThresholds'];
  private lastAdjustment: number = 0;
  private currentMax: number;

  constructor(config: AdaptiveControllerConfig) {
    this.client = config.redisClient;
    this.worker = config.worker;
    this.minRate = config.minRateLimit;
    this.maxRate = config.maxRateLimit;
    this.cooldownMs = config.cooldownMs;
    this.thresholds = config.signalThresholds;
    this.currentMax = config.maxRateLimit;
  }

  async start(pollIntervalMs: number = 5000): Promise<void> {
    const poll = async () => {
      try {
        await this.evaluateAndAdjust();
      } catch (error) {
        console.error('[AdaptiveRate] Error:', error);
      }
      setTimeout(poll, pollIntervalMs);
    };
    poll();
  }

  private async evaluateAndAdjust(): Promise<void> {
    const now = Date.now();
    if (now - this.lastAdjustment < this.cooldownMs) return;

    const [redisCpuRatio, downstreamP99, dbPoolUtil] = await Promise.all([
      this.getRedisCpuRatio(),
      this.getDownstreamP99(),
      this.getDbPoolUtilization(),
    ]);

    const pressureScore = Math.max(
      redisCpuRatio / (this.thresholds.redisCpuPercent / 100),
      downstreamP99 / this.thresholds.downstreamP99Ms,
      dbPoolUtil / (this.thresholds.dbPoolPercent / 100)
    );

    let newMax: number;

    if (pressureScore > 1.0) {
      // Over threshold — aggressive reduction
      const reduction = Math.min(0.75, (pressureScore - 1.0) * 0.5);
      newMax = Math.max(this.minRate, Math.round(this.currentMax * (1 - reduction)));
    } else if (pressureScore > 0.7) {
      // Yellow zone — hold current rate
      newMax = this.currentMax;
    } else {
      // Green zone — gradual recovery toward max
      const recovery = (1 - pressureScore) * 0.2;
      newMax = Math.min(this.maxRate, Math.round(this.currentMax * (1 + recovery)));
    }

    if (newMax !== this.currentMax) {
      this.currentMax = newMax;
      await this.worker.queue.setGlobalRateLimit(this.currentMax, 1000);
      this.lastAdjustment = now;
      console.log(
        `[AdaptiveRate] Pressure=${pressureScore.toFixed(2)} Rate=${this.currentMax}/s`
      );
    }
  }

  private async getRedisCpuRatio(): Promise<number> {
    const info = await this.client.info('cpu');
    const sysMatch = info.match(/used_cpu_sys:(\d+\.?\d*)/);
    const userMatch = info.match(/used_cpu_user:(\d+\.?\d*)/);
    if (!sysMatch || !userMatch) return 0;
    return (parseFloat(sysMatch[1]) + parseFloat(userMatch[1]));
  }

  private async getDownstreamP99(): Promise<number> {
    // Read from your metrics system (Prometheus, StatsD, etc.)
    const raw = await this.client.get('metrics:downstream:p99');
    return raw ? parseFloat(raw) : 0;
  }

  private async getDbPoolUtilization(): Promise<number> {
    const raw = await this.client.get('metrics:db:pool-utilization');
    return raw ? parseFloat(raw) : 0;
  }
}

Wire it into a worker:

const adaptiveWorker = new Worker(
  'adaptive-queue',
  async (job) => { await processJob(job); },
  {
    connection: { host: 'localhost', port: 6379 },
    limiter: { max: 200, duration: 1000 },
    concurrency: 100,
  }
);

const controller = new AdaptiveRateController({
  redisClient,
  worker: adaptiveWorker,
  minRateLimit: 10,
  maxRateLimit: 500,
  cooldownMs: 10000,
  signalThresholds: {
    redisCpuPercent: 70,
    downstreamP99Ms: 2000,
    dbPoolPercent: 80,
  },
});

await controller.start(5000);

4. Cost Analysis: Which Pattern to Pick

Each approach has different operational characteristics. Here's how they compare:

Algorithm	Redis Commands Per Check	Memory Per Key	Cluster-Safe
Token bucket	HMGET + HMSET + EXPIRE (3 ops via Lua)	~200 bytes per scope	Yes (1 key)
Sliding window log	ZREMRANGEBYSCORE + ZADD + ZCARD + EXPIRE (4 ops via Lua)	~80 bytes × request count	Yes (1 key)
BullMQ built-in	INCR + EXPIRE (2 ops)	~64 bytes per queue	Yes
Adaptive control	INFO cpu + 2 GET (3 ops)	~100 bytes per metric	N/A (centralized)

Key insight: The sliding window log is the only O(n) memory algorithm — at 1,000 req/s with a 60-second window, that's 60,000 sorted-set members taking ~4.8 MB per scope. Use it only for low-to-medium volume endpoints. The token bucket is always O(1) per scope and is the safest default.

Lua scripts block the Redis event loop while executing, but at ~25 lines of Lua these complete in under 50 microseconds. The real risk is ZREMRANGEBYSCORE on a large sorted set — mitigate by setting aggressive TTLs and consider periodic compaction in a background job.

5. Putting It All Together: Multi-Layered Architecture

The most robust production setup combines all three patterns into a defense-in-depth pipeline:

Clients → Express (L1: Token Bucket / IP) → BullMQ Queue (L2: Built-in limiter) → Workers (L3: Per-processor sliding window + adaptive control) → Downstream APIs

• L1 — Ingress (Express/Fastify middleware): Rejects abusive clients with HTTP 429 before they ever reach the queue. Saves Redis connection slots and job serialization cost.
• L2 — Queue (BullMQ's limiter option): Ensures the queue never exceeds total processing capacity regardless of L1 leakage.
• L3 — Processor (Custom token bucket + sliding window + adaptive control): Handles per-tenant quotas, per-endpoint fairness, and dynamic throttling based on live system signals.

This layered design means a bug in one layer can't cause total system failure — each layer is an independent safety net.

Conclusion

BullMQ's built-in rate limiter is an essential first wall: it's simple, free, and effective. But production queue systems face heterogeneous constraints — per-user quotas, per-endpoint fairness, adaptive throttling based on live signals — that a single fixed-window counter cannot express.

The four custom Redis patterns covered here fill that gap:

1. Token bucket (Redis Lua) — atomic per-scope burst control with O(1) memory
2. Sliding window log (sorted sets) — the fairest algorithm for per-endpoint fairness
3. Adaptive control — workers that self-regulate based on Redis CPU, downstream latency, and DB pool pressure
4. Multi-layered architecture — defense in depth from ingress to processor

Each pattern uses Redis Lua scripts for atomicity, keeps overhead bounded, and works alongside BullMQ's built-in limiter. In production, combine them: let BullMQ handle the coarse queue-level throttle, and deploy custom Redis rate limiters for the fine-grained, context-aware policies your system demands.

---

Try QueueHub for Advanced Rate Limiting Observability

QueueHub gives you real-time visibility into your BullMQ queues — including rate limit state, worker saturation, job processing latency, and Redis health metrics. Monitor your custom rate limiters alongside BullMQ's built-in controls from a single dashboard. Start free at queuehub.tech.