Scale: Redis-backed rate limiting + presence TTLs

.env.example

@@ -30,8 +30,10 @@
 
 # --- realtime: in-memory (default) | redis ---
 # Unset = single-process in-memory pub/sub + presence (the zero-config default).
-# Set REDIS_URL to fan websocket messages and presence across multiple server
-# processes for horizontal scale. The API and SDK are identical either way.
+# Set REDIS_URL to scale horizontally across multiple server processes: websocket
+# messages + presence fan out via Redis (presence entries auto-expire if a process
+# dies), and the per-IP rate limit becomes a shared, cluster-wide window. The API
+# and SDK are identical either way.
 # REDIS_URL=redis://localhost:6379
 
 # --- file storage: disk (default) | s3 ---

README.md

@@ -444,7 +444,7 @@ scale with an identical API/SDK.
 
 | Variable    | Default   | Purpose                                                       |
 | ----------- | --------- | ------------------------------------------------------------- |
-| `REDIS_URL` | _(unset)_ | Redis connection string; enables cross-process realtime fan-out |
+| `REDIS_URL` | _(unset)_ | Redis connection string; enables cross-process realtime fan-out, cluster-wide presence (entries auto-expire if a process dies), and a shared per-IP rate-limit window |
 
 ```bash
 REDIS_URL=redis://localhost:6379   # then run N server processes behind a load balancer

public/docs.html

@@ -333,7 +333,7 @@ <h3>Core &amp; access</h3>
               <tr><td class="ep">API_KEY</td><td>—</td><td>Lock the instance with one bearer key</td></tr>
               <tr><td class="ep">ADMIN_KEY</td><td>—</td><td>Enable multi-tenant projects</td></tr>
               <tr><td class="ep">CORS_ORIGIN</td><td>* (open)</td><td>Comma-separated browser origin allowlist</td></tr>
-              <tr><td class="ep">REDIS_URL</td><td>—</td><td>Cross-process realtime (messages + presence) for horizontal scale</td></tr>
+              <tr><td class="ep">REDIS_URL</td><td>—</td><td>Horizontal scale: cross-process realtime (messages + presence) + a shared per-IP rate-limit window</td></tr>
               <tr><td class="ep">LOG_LEVEL</td><td>info</td><td>trace … fatal, or silent</td></tr>
             </tbody>
           </table>

src/modules/realtime/transport.ts

@@ -71,21 +71,40 @@ interface Envelope {
   frame: unknown;
 }
 
+export interface RealtimeTransportOptions {
+  /** How long a roster entry lives without a heartbeat (ms). */
+  ttlMs?: number;
+  /** How often this process re-asserts its local members (ms). */
+  heartbeatMs?: number;
+}
+
+const DEFAULT_TTL_MS = Number(process.env.REALTIME_PRESENCE_TTL_MS ?? 30_000);
+
 /**
  * Redis transport: PUBLISH every frame to a shared topic so other processes can
- * deliver it to their local sockets, and keep each channel's roster in a Redis
- * hash so presence snapshots span the whole cluster. Frames we publish ourselves
- * are tagged with this process's `origin` and skipped on receive (we already
+ * deliver it to their local sockets, and keep each channel's roster in Redis so
+ * presence snapshots span the whole cluster. Frames we publish ourselves are
+ * tagged with this process's `origin` and skipped on receive (we already
  * delivered them locally).
+ *
+ * The roster is a sorted set per channel scored by expiry (member = cid), plus a
+ * hash of cid → identity. Each process heartbeats its own members to push their
+ * expiry forward; if a process dies, its entries simply age out — so a hard
+ * crash can't leave ghosts in the roster (Redis 7.0 has no per-field hash TTL).
  */
 class RedisTransport implements RealtimeTransport {
   readonly id = 'redis';
   private readonly pub: Redis;
   private readonly sub: Redis;
   private readonly origin = `${Math.random().toString(36).slice(2)}${Date.now().toString(36)}`;
   private handler: (channel: string, frame: unknown) => void = () => {};
+  private readonly ttlMs: number;
+  // Members this process owns, so the heartbeat can keep them alive: channel → cids.
+  private readonly local = new Map<string, Set<string>>();
+  private readonly heartbeat: NodeJS.Timeout;
 
-  constructor(redisUrl: string) {
+  constructor(redisUrl: string, opts: RealtimeTransportOptions = {}) {
+    this.ttlMs = opts.ttlMs ?? DEFAULT_TTL_MS;
     this.pub = new Redis(redisUrl, { maxRetriesPerRequest: null });
     this.sub = this.pub.duplicate();
     void this.sub.subscribe(PUBSUB_CHANNEL);
@@ -99,6 +118,9 @@ class RedisTransport implements RealtimeTransport {
       if (env.origin === this.origin) return; // we already delivered locally
       this.handler(env.channel, env.frame);
     });
+
+    this.heartbeat = setInterval(() => void this.beat(), opts.heartbeatMs ?? Math.max(1000, Math.floor(this.ttlMs / 3)));
+    this.heartbeat.unref?.();
   }
 
   onRemote(handler: (channel: string, frame: unknown) => void): void {
@@ -112,27 +134,66 @@ class RedisTransport implements RealtimeTransport {
     });
   }
 
+  private zsetKey = (channel: string): string => ROSTER_PREFIX + channel;
+  private idKey = (channel: string): string => `${ROSTER_PREFIX}${channel}:id`;
+
   async addMember(channel: string, member: Member): Promise<void> {
-    await this.pub.hset(ROSTER_PREFIX + channel, member.cid, member.identity ?? '');
+    let set = this.local.get(channel);
+    if (!set) this.local.set(channel, (set = new Set()));
+    set.add(member.cid);
+    await this.pub
+      .multi()
+      .zadd(this.zsetKey(channel), Date.now() + this.ttlMs, member.cid)
+      .hset(this.idKey(channel), member.cid, member.identity ?? '')
+      .exec();
   }
 
   async removeMember(channel: string, cid: string): Promise<void> {
-    await this.pub.hdel(ROSTER_PREFIX + channel, cid);
+    this.local.get(channel)?.delete(cid);
+    await this.pub.multi().zrem(this.zsetKey(channel), cid).hdel(this.idKey(channel), cid).exec();
   }
 
   async members(channel: string): Promise<Member[]> {
-    const map = await this.pub.hgetall(ROSTER_PREFIX + channel);
-    return Object.entries(map).map(([cid, identity]) => ({ cid, identity: (identity as string) || null }));
+    const now = Date.now();
+    // Evict any expired entries (e.g. a crashed process's members) first.
+    const expired = await this.pub.zrangebyscore(this.zsetKey(channel), 0, now);
+    if (expired.length) {
+      await this.pub.multi().zrem(this.zsetKey(channel), ...expired).hdel(this.idKey(channel), ...expired).exec();
+    }
+    const cids = await this.pub.zrangebyscore(this.zsetKey(channel), `(${now}`, '+inf');
+    if (cids.length === 0) return [];
+    const identities = await this.pub.hmget(this.idKey(channel), ...cids);
+    return cids.map((cid, i) => ({ cid, identity: identities[i] || null }));
+  }
+
+  /** Push each locally-owned member's expiry forward so live members never age out. */
+  private async beat(): Promise<void> {
+    const now = Date.now();
+    for (const [channel, cids] of this.local) {
+      if (cids.size === 0) continue;
+      const args: (string | number)[] = [];
+      for (const cid of cids) args.push(now + this.ttlMs, cid);
+      try {
+        await this.pub.zadd(this.zsetKey(channel), ...args);
+      } catch {
+        /* transient; the next beat will retry */
+      }
+    }
   }
 
   async close(): Promise<void> {
+    clearInterval(this.heartbeat);
+    this.local.clear();
     // Graceful QUIT drains in-flight commands and disables reconnection, so we
-    // don't leave pending hgetall/publish to reject after shutdown.
+    // don't leave pending commands to reject after shutdown.
     await Promise.allSettled([this.pub.quit(), this.sub.quit()]);
   }
 }
 
 /** Construct the realtime transport selected by REDIS_URL (defaults to local). */
-export function getRealtimeTransport(redisUrl = config.realtime.redisUrl): RealtimeTransport {
-  return redisUrl ? new RedisTransport(redisUrl) : new LocalTransport();
+export function getRealtimeTransport(
+  redisUrl = config.realtime.redisUrl,
+  opts: RealtimeTransportOptions = {},
+): RealtimeTransport {
+  return redisUrl ? new RedisTransport(redisUrl, opts) : new LocalTransport();
 }

src/rate-limit.ts

@@ -1,3 +1,4 @@
+import { Redis } from 'ioredis';
 import type { FastifyInstance } from 'fastify';
 
 export interface RateLimitOptions {
@@ -7,14 +8,19 @@ export interface RateLimitOptions {
 }
 
 /**
- * In-memory, per-IP fixed-window rate limiter. No dependency, single-process —
- * exactly the scope of one VM. Sets X-RateLimit-* headers and answers 429 with
- * the standard error envelope once a client exceeds its window.
+ * Per-IP fixed-window rate limiter. In-memory by default (single process — the
+ * zero-config VM); when a Redis URL is given the window counter lives in Redis
+ * so the limit holds across every process in a horizontally-scaled deployment.
+ * Either way it sets X-RateLimit-* headers and answers 429 once a client exceeds
+ * its window.
  */
-export function registerRateLimit(app: FastifyInstance, opts: RateLimitOptions): void {
+export function registerRateLimit(app: FastifyInstance, opts: RateLimitOptions, redisUrl: string | null = null): void {
   if (opts.max <= 0) return; // disabled
+  if (redisUrl) registerRedisLimiter(app, opts, redisUrl);
+  else registerMemoryLimiter(app, opts);
+}
 
-  const { max, windowMs } = opts;
+function registerMemoryLimiter(app: FastifyInstance, { max, windowMs }: RateLimitOptions): void {
   const hits = new Map<string, { count: number; resetAt: number }>();
 
   // Periodically drop expired buckets so the map can't grow unbounded.
@@ -35,14 +41,51 @@ export function registerRateLimit(app: FastifyInstance, opts: RateLimitOptions):
       hits.set(req.ip, bucket);
     }
     bucket.count += 1;
-
-    reply.header('X-RateLimit-Limit', max);
-    reply.header('X-RateLimit-Remaining', Math.max(0, max - bucket.count));
-    reply.header('X-RateLimit-Reset', Math.ceil(bucket.resetAt / 1000));
-
+    setRateHeaders(reply, max, bucket.count, bucket.resetAt - now);
     if (bucket.count > max) {
       reply.header('Retry-After', Math.ceil((bucket.resetAt - now) / 1000));
       return reply.code(429).send({ error: 'rate_limited', message: 'Too many requests.' });
     }
   });
 }
+
+function registerRedisLimiter(app: FastifyInstance, { max, windowMs }: RateLimitOptions, redisUrl: string): void {
+  const redis = new Redis(redisUrl, { maxRetriesPerRequest: null });
+  app.addHook('onClose', async () => {
+    redis.disconnect();
+  });
+
+  app.addHook('onRequest', async (req, reply) => {
+    if (req.url === '/health') return;
+
+    const key = `zero:rl:${req.ip}`;
+    let count: number;
+    let ttl: number;
+    try {
+      // INCR the window counter and read its TTL in one round trip.
+      const res = await redis.multi().incr(key).pttl(key).exec();
+      count = Number(res?.[0]?.[1] ?? 0);
+      ttl = Number(res?.[1]?.[1] ?? -1);
+      // First hit of a new window: start the expiry.
+      if (ttl < 0) {
+        await redis.pexpire(key, windowMs);
+        ttl = windowMs;
+      }
+    } catch {
+      // Fail open: a Redis hiccup must not take the whole API down.
+      return;
+    }
+
+    setRateHeaders(reply, max, count, ttl);
+    if (count > max) {
+      reply.header('Retry-After', Math.ceil(ttl / 1000));
+      return reply.code(429).send({ error: 'rate_limited', message: 'Too many requests.' });
+    }
+  });
+}
+
+function setRateHeaders(reply: { header: (k: string, v: number) => void }, max: number, count: number, ttlMs: number): void {
+  reply.header('X-RateLimit-Limit', max);
+  reply.header('X-RateLimit-Remaining', Math.max(0, max - count));
+  reply.header('X-RateLimit-Reset', Math.ceil((Date.now() + ttlMs) / 1000));
+}

src/server.ts

@@ -136,7 +136,8 @@ export async function buildServer(opts: BuildOptions = {}): Promise<FastifyInsta
   await app.register(multipart, { limits: { fileSize: config.maxUploadBytes } });
   await app.register(websocket);
 
-  registerRateLimit(app, opts.rateLimit ?? config.rateLimit);
+  // Share the limit across processes when Redis is configured (multi-process scale).
+  registerRateLimit(app, opts.rateLimit ?? config.rateLimit, config.realtime.redisUrl);
 
   // Resolve the tenant for each request and enforce auth.
   //  - Admin paths (/v1/admin/*) authenticate with ADMIN_KEY.

test/realtime-redis.test.ts

@@ -1,7 +1,9 @@
 import { test } from 'node:test';
 import assert from 'node:assert/strict';
 import type { AddressInfo } from 'node:net';
+import { Redis } from 'ioredis';
 import { buildServer } from '../src/server.js';
+import { getRealtimeTransport } from '../src/modules/realtime/transport.js';
 
 // Cross-process realtime requires Redis. Skipped unless REDIS_URL is set (so the
 // default suite needs no Redis); CI runs it with a redis service via `test:redis`.
@@ -64,3 +66,49 @@ test('realtime (redis): messages and presence fan out across two processes', { s
     await s2.close();
   }
 });
+
+test('rate limit (redis): the per-IP window is shared across processes', { skip: !REDIS }, async () => {
+  // A tiny shared limit across two instances; the 4th request overall is blocked
+  // even though it lands on the second instance (proving the counter is shared).
+  const rl = { max: 3, windowMs: 60_000 };
+  const s1 = await buildServer({ logger: false, rateLimit: rl });
+  const s2 = await buildServer({ logger: false, rateLimit: rl });
+  await s1.listen({ host: '127.0.0.1', port: 0 });
+  await s2.listen({ host: '127.0.0.1', port: 0 });
+  const u1 = `http://127.0.0.1:${(s1.server.address() as AddressInfo).port}/v1`;
+  const u2 = `http://127.0.0.1:${(s2.server.address() as AddressInfo).port}/v1`;
+  const cleaner = new Redis(REDIS as string, { maxRetriesPerRequest: null });
+  try {
+    await cleaner.del('zero:rl:127.0.0.1'); // start from a clean window
+
+    assert.equal((await fetch(u1)).status, 200); // 1
+    assert.equal((await fetch(u1)).status, 200); // 2
+    assert.equal((await fetch(u2)).status, 200); // 3 (other instance)
+    const fourth = await fetch(u2); // 4 → over the shared limit
+    assert.equal(fourth.status, 429, 'the shared window should block across processes');
+    assert.equal(fourth.headers.get('x-ratelimit-limit'), '3');
+  } finally {
+    cleaner.disconnect();
+    await s1.close();
+    await s2.close();
+  }
+});
+
+test('presence (redis): a crashed process’s members age out of the roster', { skip: !REDIS }, async () => {
+  const channel = `ttl-room-${Date.now()}`;
+  // t1 owns a member with a short TTL; t2 is an independent reader.
+  const t1 = getRealtimeTransport(REDIS, { ttlMs: 600, heartbeatMs: 150 });
+  const t2 = getRealtimeTransport(REDIS);
+  try {
+    await t1.addMember(channel, { cid: 'ghost', identity: 'casper' });
+    assert.deepEqual((await t2.members(channel)).map((m) => m.cid), ['ghost']);
+
+    // Simulate a hard crash: stop t1 (its heartbeat) WITHOUT a graceful leave.
+    await t1.close();
+    await wait(900); // > ttl, with no heartbeat refreshing the entry
+
+    assert.deepEqual(await t2.members(channel), [], 'a stale member should expire');
+  } finally {
+    await t2.close();
+  }
+});