diff --git a/.env.example b/.env.example index bb0f3af..88d2851 100644 --- a/.env.example +++ b/.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 --- diff --git a/README.md b/README.md index b6db00a..c5398a4 100644 --- a/README.md +++ b/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 diff --git a/public/docs.html b/public/docs.html index 2e8418f..c3a12e0 100644 --- a/public/docs.html +++ b/public/docs.html @@ -333,7 +333,7 @@

Core & access

API_KEY—Lock the instance with one bearer key ADMIN_KEY—Enable multi-tenant projects CORS_ORIGIN* (open)Comma-separated browser origin allowlist - REDIS_URL—Cross-process realtime (messages + presence) for horizontal scale + REDIS_URL—Horizontal scale: cross-process realtime (messages + presence) + a shared per-IP rate-limit window LOG_LEVELinfotrace … fatal, or silent diff --git a/src/modules/realtime/transport.ts b/src/modules/realtime/transport.ts index 271e000..4f49834 100644 --- a/src/modules/realtime/transport.ts +++ b/src/modules/realtime/transport.ts @@ -71,12 +71,26 @@ 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'; @@ -84,8 +98,13 @@ class RedisTransport implements RealtimeTransport { 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>(); + 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 { - 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 { - 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 { - 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 { + 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 { + 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(); } diff --git a/src/rate-limit.ts b/src/rate-limit.ts index 62457e7..00a2e39 100644 --- a/src/rate-limit.ts +++ b/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(); // 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)); +} diff --git a/src/server.ts b/src/server.ts index c601636..4edb32a 100644 --- a/src/server.ts +++ b/src/server.ts @@ -136,7 +136,8 @@ export async function buildServer(opts: BuildOptions = {}): Promise { + // 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(); + } +});