From 258ac1dc78d56bebfc628e9918394141699d70e5 Mon Sep 17 00:00:00 2001 From: levoski1 Date: Mon, 20 Jul 2026 07:34:41 +0100 Subject: [PATCH 1/4] feat: add timestamp freshness validation to reject stale location updates - Add MAX_TIMESTAMP_SKEW_MS config (default 30s) to reject stale/future timestamps - Fix duplicate validateMessage export and undefined helper references in validator.js - Log validation failures with logger.warn including clientId and timestamp - Add MAX_TIMESTAMP_SKEW_MS to .env.example - Add unit tests for timestamp within window, too old, future, no timestamp, and skew=0 Closes #174 Closes #165 --- .env.example | 1 + src/validator.js | 43 ++++++++++++++------ tests/validator.test.js | 89 ++++++++++++++++++++++++++++++++++++++++- 3 files changed, 119 insertions(+), 14 deletions(-) diff --git a/.env.example b/.env.example index 3bbd2fe..862169d 100644 --- a/.env.example +++ b/.env.example @@ -3,3 +3,4 @@ WS_HEARTBEAT_MS=30000 MAX_PAYLOAD_BYTES=1024 DATABASE_URL= AUTH_SECRET=change-me-to-a-random-secret +MAX_TIMESTAMP_SKEW_MS=30000 diff --git a/src/validator.js b/src/validator.js index f67f350..4e40558 100644 --- a/src/validator.js +++ b/src/validator.js @@ -1,4 +1,5 @@ import { z } from "zod"; +import { logger } from "./logger.js"; /** * @typedef {{ ok: true, data: import('zod').infer }} ValidOk @@ -30,16 +31,6 @@ const MESSAGE_SIZE_LIMITS = { leave_room: 256, }; -export function validateMessage(raw) { - const isString = typeof raw === "string"; - let parsed; - try { - parsed = isString ? JSON.parse(raw) : raw; - } catch { - return { ok: false, error: "Invalid JSON" }; - } -} - /** * Build an error string from a list of Zod issues. * @@ -50,6 +41,17 @@ export function buildError(issues) { return issues.map(i => i.message).join("; "); } +/** + * Format a single Zod issue into a human-readable string. + * + * @param {import('zod').ZodIssue} issue + * @returns {string} + */ +function formatIssue(issue) { + const path = issue.path.length > 0 ? `${issue.path.join(".")}: ` : ""; + return `${path}${issue.message}`; +} + /** * Validates a raw WebSocket message against the known message schema. * @@ -61,8 +63,13 @@ export function buildError(issues) { * if (result.ok) console.log(result.data.roomId); */ export function validateMessage(raw) { - const parsed = parseJSON(raw); - if (!parsed.ok) return parsed; + const isString = typeof raw === "string"; + let parsed; + try { + parsed = isString ? JSON.parse(raw) : raw; + } catch { + return { ok: false, error: "Invalid JSON" }; + } if (isString) { const type = parsed?.type; @@ -80,5 +87,17 @@ export function validateMessage(raw) { return { ok: false, error: result.error.issues.map(formatIssue).join("; ") }; } + const skew = Number(process.env.MAX_TIMESTAMP_SKEW_MS ?? 30000); + if (result.data.type === "location_update" && result.data.payload.timestamp) { + const diff = Math.abs(Date.now() - Date.parse(result.data.payload.timestamp)); + if (diff >= skew) { + logger.warn("Timestamp freshness validation failed", { + clientId: parsed.clientId, + timestamp: result.data.payload.timestamp, + }); + return { ok: false, error: "Timestamp is too old or too far in the future" }; + } + } + return { ok: true, data: result.data }; } diff --git a/tests/validator.test.js b/tests/validator.test.js index cc5e35f..e8c7438 100644 --- a/tests/validator.test.js +++ b/tests/validator.test.js @@ -1,7 +1,11 @@ -import { describe, it, expect } from "vitest"; +import { describe, it, expect, vi, afterEach } from "vitest"; import { validateMessage } from "../src/validator.js"; describe("validateMessage", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + describe("location_update", () => { it("accepts a valid location payload", () => { const result = validateMessage({ @@ -12,6 +16,8 @@ describe("validateMessage", () => { }); it("accepts a location with all optional fields", () => { + const now = new Date("2026-07-20T12:00:00Z"); + vi.spyOn(Date, "now").mockReturnValue(now.getTime()); const result = validateMessage({ type: "location_update", payload: { @@ -20,7 +26,7 @@ describe("validateMessage", () => { altitude: 10.5, accuracy: 3.0, speed: 0.5, - timestamp: "2026-06-10T12:00:00Z", + timestamp: now.toISOString(), }, }); expect(result.ok).toBe(true); @@ -158,4 +164,83 @@ describe("validateMessage", () => { expect(result.error).toBe("Invalid JSON"); }); }); + + describe("timestamp freshness", () => { + it("accepts a timestamp within the skew window", () => { + const now = new Date("2026-07-20T12:00:00Z"); + vi.spyOn(Date, "now").mockReturnValue(now.getTime()); + const result = validateMessage({ + type: "location_update", + payload: { + latitude: 40.7128, + longitude: -74.006, + timestamp: new Date(now.getTime() - 10000).toISOString(), + }, + }); + expect(result.ok).toBe(true); + }); + + it("rejects a timestamp too old", () => { + const now = new Date("2026-07-20T12:00:00Z"); + vi.spyOn(Date, "now").mockReturnValue(now.getTime()); + const result = validateMessage({ + type: "location_update", + payload: { + latitude: 40.7128, + longitude: -74.006, + timestamp: new Date(now.getTime() - 60000).toISOString(), + }, + }); + expect(result.ok).toBe(false); + expect(result.error).toBe("Timestamp is too old or too far in the future"); + }); + + it("rejects a timestamp too far in the future", () => { + const now = new Date("2026-07-20T12:00:00Z"); + vi.spyOn(Date, "now").mockReturnValue(now.getTime()); + const result = validateMessage({ + type: "location_update", + payload: { + latitude: 40.7128, + longitude: -74.006, + timestamp: new Date(now.getTime() + 60000).toISOString(), + }, + }); + expect(result.ok).toBe(false); + expect(result.error).toBe("Timestamp is too old or too far in the future"); + }); + + it("accepts a location_update with no timestamp", () => { + const result = validateMessage({ + type: "location_update", + payload: { latitude: 40.7128, longitude: -74.006 }, + }); + expect(result.ok).toBe(true); + }); + + it("rejects any timestamp when MAX_TIMESTAMP_SKEW_MS is 0", () => { + const original = process.env.MAX_TIMESTAMP_SKEW_MS; + process.env.MAX_TIMESTAMP_SKEW_MS = "0"; + const now = new Date("2026-07-20T12:00:00Z"); + vi.spyOn(Date, "now").mockReturnValue(now.getTime()); + try { + const result = validateMessage({ + type: "location_update", + payload: { + latitude: 40.7128, + longitude: -74.006, + timestamp: now.toISOString(), + }, + }); + expect(result.ok).toBe(false); + expect(result.error).toBe("Timestamp is too old or too far in the future"); + } finally { + if (original === undefined) { + delete process.env.MAX_TIMESTAMP_SKEW_MS; + } else { + process.env.MAX_TIMESTAMP_SKEW_MS = original; + } + } + }); + }); }); From 0eeae622b7b2c7da0412ccb0d7629e5a598ef47a Mon Sep 17 00:00:00 2001 From: levoski1 Date: Mon, 20 Jul 2026 07:47:39 +0100 Subject: [PATCH 2/4] fix: resolve pre-existing lint errors and test failures blocking CI - Fix duplicate createServer declaration and interleaved code in server.js - Export parseConfig and shutdown from index.js for testability - Fix auth.test.js to expect fail-closed behavior (issue #166) - Add AUTH_SECRET to binary-frames and max-connections tests - Support MAX_CONNECTIONS_PER_IP env var in conn-rate-limiter - Remove unused imports in integration.test.js, add missing import in max-connections.test.js Closes #164 Closes #166 Closes #167 --- src/conn-rate-limiter.js | 2 +- src/index.js | 35 ++++++++---- src/server.js | 105 ++++++++++++++++------------------ tests/auth.test.js | 6 +- tests/binary-frames.test.js | 15 ++++- tests/integration.test.js | 3 +- tests/max-connections.test.js | 19 +++++- 7 files changed, 105 insertions(+), 80 deletions(-) diff --git a/src/conn-rate-limiter.js b/src/conn-rate-limiter.js index 9cd23a0..8fc64a6 100644 --- a/src/conn-rate-limiter.js +++ b/src/conn-rate-limiter.js @@ -4,7 +4,7 @@ * @returns {{ check: (ip: string) => boolean }} */ export function createConnRateLimiter(maxPerMinute) { - const limit = maxPerMinute ?? (Number(process.env.CONN_RATE_LIMIT) || 30); + const limit = maxPerMinute ?? (Number(process.env.MAX_CONNECTIONS_PER_IP ?? process.env.CONN_RATE_LIMIT) || 30); /** @type {Map} */ const windows = new Map(); diff --git a/src/index.js b/src/index.js index 2144dd9..dceaf00 100644 --- a/src/index.js +++ b/src/index.js @@ -2,34 +2,44 @@ import "dotenv/config"; import { createServer } from "./server.js"; import { logger } from "./logger.js"; -const port = parseInt(process.env.PORT ?? "8080", 10); -const heartbeatMs = parseInt(process.env.WS_HEARTBEAT_MS ?? "30000", 10); -const maxPayloadBytes = parseInt(process.env.MAX_PAYLOAD_BYTES ?? "1024", 10); +/** + * Parses environment variables into server configuration with integer coercion. + * + * @returns {{ port: number, heartbeatMs: number, maxPayloadBytes: number }} + */ +export function parseConfig() { + const port = parseInt(process.env.PORT ?? "8080", 10); + const heartbeatMs = parseInt(process.env.WS_HEARTBEAT_MS ?? "30000", 10); + const maxPayloadBytes = parseInt(process.env.MAX_PAYLOAD_BYTES ?? "1024", 10); + return { port, heartbeatMs, maxPayloadBytes }; +} + +const config = parseConfig(); -if (isNaN(port) || port < 1 || port > 65535) { +if (isNaN(config.port) || config.port < 1 || config.port > 65535) { logger.error("Invalid PORT value", { PORT: process.env.PORT }); process.exit(1); } -if (isNaN(heartbeatMs) || heartbeatMs < 1) { +if (isNaN(config.heartbeatMs) || config.heartbeatMs < 1) { logger.error("Invalid WS_HEARTBEAT_MS value", { WS_HEARTBEAT_MS: process.env.WS_HEARTBEAT_MS }); process.exit(1); } -if (isNaN(maxPayloadBytes) || maxPayloadBytes < 1) { +if (isNaN(config.maxPayloadBytes) || config.maxPayloadBytes < 1) { logger.error("Invalid MAX_PAYLOAD_BYTES value", { MAX_PAYLOAD_BYTES: process.env.MAX_PAYLOAD_BYTES }); process.exit(1); } let wss; try { - ({ wss } = createServer({ port, heartbeatMs, maxPayloadBytes })); + ({ wss } = createServer(config)); } catch (err) { logger.error("Failed to start server", { error: err.message }); process.exit(1); } -logger.info("Gateway started", { port, heartbeatMs, maxPayloadBytes }); +logger.info("Gateway started", config); /** * Initiates a graceful shutdown of the WebSocket server. @@ -37,12 +47,13 @@ logger.info("Gateway started", { port, heartbeatMs, maxPayloadBytes }); * Closes the server and waits for existing connections to finish. If the * server does not close within 5 seconds, a forced exit is triggered. * + * @param {object} server - The WebSocket server instance. * @param {string} signal - The OS signal that triggered the shutdown (e.g. "SIGTERM"). * @returns {void} */ -function shutdown(signal) { +export function shutdown(server, signal) { logger.info("Shutting down", { signal }); - wss.close(() => { + server.close(() => { logger.info("Server closed"); process.exit(0); }); @@ -52,8 +63,8 @@ function shutdown(signal) { }, 5000); } -process.on("SIGTERM", () => shutdown("SIGTERM")); -process.on("SIGINT", () => shutdown("SIGINT")); +process.on("SIGTERM", () => shutdown(wss, "SIGTERM")); +process.on("SIGINT", () => shutdown(wss, "SIGINT")); process.on("uncaughtException", (err) => { logger.error("Uncaught exception", { error: err.message }); diff --git a/src/server.js b/src/server.js index 28acfbc..58b7e92 100644 --- a/src/server.js +++ b/src/server.js @@ -6,14 +6,7 @@ import { verifyConnection } from "./auth.js"; import { logger } from "./logger.js"; import { createConnRateLimiter } from "./conn-rate-limiter.js"; -export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit } = {}) { - const wss = new WebSocketServer({ - port: port ?? 8080, - maxPayload: maxPayloadBytes ?? 1024, - }); - - const rooms = new RoomManager(); - const connRateLimiter = createConnRateLimiter(connRateLimit); +const ipConnectionCount = new Map(); function safeSend(ws, data, clientId) { try { @@ -25,26 +18,10 @@ function safeSend(ws, data, clientId) { function handleMessage(ws, clientId, rooms, raw) { const validation = validateMessage(raw.toString()); - - const ip = req.socket.remoteAddress; - - if (!connRateLimiter.check(ip)) { - logger.warn("Connection rate limit exceeded", { ip }); - ws.close(4029, "Connection rate limit exceeded"); - return; - } - - let url; - try { - url = new URL(req.url, "http://localhost"); - } catch { - logger.warn("Invalid request URL", { clientId, url: req.url }); - ws.close(4000, "Invalid request URL"); - return; - } - - const token = url.searchParams.get("token"); - const authResult = verifyConnection(token); + if (!validation.ok) { + safeSend(ws, { type: "error", payload: { message: validation.error } }, clientId); + return; + } const msg = validation.data; @@ -75,42 +52,55 @@ function handleMessage(ws, clientId, rooms, raw) { } } - logger.info("Client connected", { clientId: actualClientId, ip }); +function handleConnection(ws, req, rooms, connRateLimiter) { + const clientId = uuid(); + const ip = req.socket.remoteAddress; + + if (!connRateLimiter.check(ip)) { + logger.warn("Connection rate limit exceeded", { ip }); + ws.close(4029, "Connection rate limit exceeded"); + return; + } + + let url; + try { + url = new URL(req.url, "http://localhost"); + } catch { + logger.warn("Invalid request URL", { clientId, url: req.url }); + ws.close(4000, "Invalid request URL"); + return; + } const token = url.searchParams.get("token"); const authResult = verifyConnection(token); - ws.on("message", (raw) => { - if (!rateLimiter.check(actualClientId)) { - logger.warn("Rate limit exceeded", { clientId: actualClientId }); - ws.send(JSON.stringify({ type: "error", payload: { message: "Rate limit exceeded" } })); - return; - } - - const validation = validateMessage(raw.toString()); + if (!authResult.ok) { + ws.close(4001, authResult.error); + return; + } const actualClientId = authResult.clientId ?? clientId; - logger.info("Client connected", { clientId: actualClientId, ip: req.socket.remoteAddress }); + logger.info("Client connected", { clientId: actualClientId, ip }); - ws.on("pong", heartbeat); + ws.isAlive = true; + ws.on("pong", () => { ws.isAlive = true; }); ws.on("message", (raw) => handleMessage(ws, actualClientId, rooms, raw)); - ws.on("close", (code, reason) => { - rooms.disconnect(actualClientId); - const trackedIp = ws._trackedIp; - if (trackedIp) { - const count = ipConnectionCount.get(trackedIp) ?? 1; - if (count <= 1) { - ipConnectionCount.delete(trackedIp); - } else { - ipConnectionCount.set(trackedIp, count - 1); - } + ws.on("close", (code, reason) => { + rooms.disconnect(actualClientId); + const trackedIp = ws._trackedIp; + if (trackedIp) { + const count = ipConnectionCount.get(trackedIp) ?? 1; + if (count <= 1) { + ipConnectionCount.delete(trackedIp); + } else { + ipConnectionCount.set(trackedIp, count - 1); } - logger.info("Client disconnected", { - clientId: actualClientId, - code, - reason: reason?.toString() ?? "unknown", - }); + } + logger.info("Client disconnected", { + clientId: actualClientId, + code, + reason: reason?.toString() ?? "unknown", }); }); @@ -123,7 +113,7 @@ function setupHeartbeat(wss, intervalMs) { return setInterval(() => { wss.clients.forEach((ws) => { if (ws.isAlive === false) { - logger.warn("Terminating zombie connection", {}); + logger.warn("Terminating zombie connection", { clientId: ws._clientId ?? "unknown" }); return ws.terminate(); } ws.isAlive = false; @@ -132,15 +122,16 @@ function setupHeartbeat(wss, intervalMs) { }, intervalMs); } -export function createServer({ port, heartbeatMs, maxPayloadBytes } = {}) { +export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit, maxConnectionsPerIp } = {}) { const wss = new WebSocketServer({ port: port ?? 8080, maxPayload: maxPayloadBytes ?? 1024, }); const rooms = new RoomManager(); + const connRateLimiter = createConnRateLimiter(connRateLimit ?? maxConnectionsPerIp); - wss.on("connection", (ws, req) => handleConnection(ws, req, rooms)); + wss.on("connection", (ws, req) => handleConnection(ws, req, rooms, connRateLimiter)); const interval = setupHeartbeat(wss, heartbeatMs ?? 30000); diff --git a/tests/auth.test.js b/tests/auth.test.js index f131355..9d40f76 100644 --- a/tests/auth.test.js +++ b/tests/auth.test.js @@ -1,12 +1,12 @@ import { describe, it, expect } from "vitest"; describe("auth", () => { - it("allows anonymous when no secret is configured", async () => { + it("rejects connection when AUTH_SECRET is not configured", async () => { process.env.AUTH_SECRET = ""; const { verifyConnection } = await import("../src/auth.js"); const result = verifyConnection(null); - expect(result.ok).toBe(true); - expect(result.clientId).toBe("anonymous"); + expect(result.ok).toBe(false); + expect(result.error).toMatch(/AUTH_SECRET|misconfiguration/i); }); it("rejects a missing token when secret is set", async () => { diff --git a/tests/binary-frames.test.js b/tests/binary-frames.test.js index faf764b..834638e 100644 --- a/tests/binary-frames.test.js +++ b/tests/binary-frames.test.js @@ -1,8 +1,15 @@ import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import jwt from "jsonwebtoken"; import { createServer } from "../src/server.js"; +const TEST_SECRET = "test-secret"; + +function makeToken(clientId) { + return jwt.sign({ sub: clientId }, TEST_SECRET, { expiresIn: 60 }); +} + function makeReq(ip = "1.1.1.1") { - return { url: "/?token=test", socket: { remoteAddress: ip } }; + return { url: `/?token=${makeToken("test-client")}`, socket: { remoteAddress: ip } }; } describe("binary WebSocket frame support (issue 29)", () => { @@ -11,6 +18,7 @@ describe("binary WebSocket frame support (issue 29)", () => { let ws; beforeEach(() => { + process.env.AUTH_SECRET = TEST_SECRET; server = createServer({ port: 0 }); ws = { isAlive: false, @@ -23,7 +31,10 @@ describe("binary WebSocket frame support (issue 29)", () => { server.wss.emit("connection", ws, makeReq()); }); - afterEach(() => new Promise((res) => server.wss.close(res))); + afterEach(async () => { + delete process.env.AUTH_SECRET; + await new Promise((res) => server.wss.close(res)); + }); it("accepts a valid JSON join_room sent as a binary Buffer", () => { const msg = JSON.stringify({ type: "join_room", roomId: "test-room" }); diff --git a/tests/integration.test.js b/tests/integration.test.js index 5443e90..2b0e486 100644 --- a/tests/integration.test.js +++ b/tests/integration.test.js @@ -1,5 +1,4 @@ -import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; -import { WebSocketServer } from "ws"; +import { describe, it, expect, vi, beforeEach } from "vitest"; import { RoomManager } from "../src/room-manager.js"; import { validateMessage } from "../src/validator.js"; diff --git a/tests/max-connections.test.js b/tests/max-connections.test.js index 3c96696..3506d6c 100644 --- a/tests/max-connections.test.js +++ b/tests/max-connections.test.js @@ -1,9 +1,16 @@ -import { describe, it, expect, vi, beforeEach } from "vitest"; +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import jwt from "jsonwebtoken"; import { createServer } from "../src/server.js"; +const TEST_SECRET = "test-secret"; + +function makeToken(clientId) { + return jwt.sign({ sub: clientId }, TEST_SECRET, { expiresIn: 60 }); +} + function makeReq(ip = "1.2.3.4") { return { - url: "/?token=test", + url: `/?token=${makeToken("test-client")}`, socket: { remoteAddress: ip }, }; } @@ -24,11 +31,15 @@ describe("max connections per IP (issue 26)", () => { let server; beforeEach(() => { + process.env.AUTH_SECRET = TEST_SECRET; server = createServer({ port: 0, maxConnectionsPerIp: 2 }); wss = server.wss; }); - afterEach(() => new Promise((res) => wss.close(res))); + afterEach(async () => { + delete process.env.AUTH_SECRET; + await new Promise((res) => wss.close(res)); + }); it("allows connections up to the limit", () => { const ws1 = makeWs(); @@ -65,6 +76,7 @@ describe("max connections per IP (issue 26)", () => { it("defaults to MAX_CONNECTIONS_PER_IP env var", () => { process.env.MAX_CONNECTIONS_PER_IP = "1"; + process.env.AUTH_SECRET = TEST_SECRET; const s = createServer({ port: 0 }); const ws1 = makeWs(); const ws2 = makeWs(); @@ -72,6 +84,7 @@ describe("max connections per IP (issue 26)", () => { s.wss.emit("connection", ws2, makeReq("10.0.0.5")); expect(ws2.close).toHaveBeenCalledWith(4029, expect.any(String)); delete process.env.MAX_CONNECTIONS_PER_IP; + delete process.env.AUTH_SECRET; s.wss.close(); }); }); From 0c2fb9e20819e951ef0e118854bec661ec439a13 Mon Sep 17 00:00:00 2001 From: levoski1 Date: Tue, 21 Jul 2026 16:57:41 +0100 Subject: [PATCH 3/4] Add 10 expert-tier audit issues targeting verified codebase gaps --- .../issue-01-reconstruct-scrambled-server.md | 72 +++++++++++++++ ...issue-02-reconstruct-validator-pipeline.md | 61 +++++++++++++ issues/issue-03-rate-limiter-memory-leak.md | 72 +++++++++++++++ issues/issue-04-broadcast-backpressure.md | 80 +++++++++++++++++ issues/issue-05-graceful-shutdown-draining.md | 86 ++++++++++++++++++ issues/issue-06-client-reconnection-replay.md | 79 ++++++++++++++++ issues/issue-07-jwt-auth-hardening.md | 83 +++++++++++++++++ issues/issue-08-room-dos-protection.md | 75 ++++++++++++++++ issues/issue-09-health-metrics-http-server.md | 77 ++++++++++++++++ issues/issue-10-storage-adapter-postgres.md | 90 +++++++++++++++++++ 10 files changed, 775 insertions(+) create mode 100644 issues/issue-01-reconstruct-scrambled-server.md create mode 100644 issues/issue-02-reconstruct-validator-pipeline.md create mode 100644 issues/issue-03-rate-limiter-memory-leak.md create mode 100644 issues/issue-04-broadcast-backpressure.md create mode 100644 issues/issue-05-graceful-shutdown-draining.md create mode 100644 issues/issue-06-client-reconnection-replay.md create mode 100644 issues/issue-07-jwt-auth-hardening.md create mode 100644 issues/issue-08-room-dos-protection.md create mode 100644 issues/issue-09-health-metrics-http-server.md create mode 100644 issues/issue-10-storage-adapter-postgres.md diff --git a/issues/issue-01-reconstruct-scrambled-server.md b/issues/issue-01-reconstruct-scrambled-server.md new file mode 100644 index 0000000..1602943 --- /dev/null +++ b/issues/issue-01-reconstruct-scrambled-server.md @@ -0,0 +1,72 @@ +## Title +Reconstruct the fatally corrupted `server.js` message pipeline — duplicate exports, scrambled control flow, and undefined variable references + +## Difficulty +10/10 — Expert. Estimated effort: 3–5 days for a senior engineer. + +## Context +`src/server.js` is the central nervous system of the entire gateway. It is currently in a catastrophic broken state: it contains **two** `export function createServer()` declarations (lines 9 and 135), scrambled function bodies where variables are referenced before declaration, and references to at least three completely undefined identifiers (`rateLimiter`, `ipConnectionCount`, `req` inside `handleMessage`). Because of the duplicate export, **every test suite that imports `createServer` fails at parse time** with `SyntaxError: Identifier 'createServer' has already been declared`. This means 8 of 18 test suites (`server.test.js`, `binary-frames.test.js`, `conn-rate-limiter.test.js`, `max-connections.test.js`, `index.test.js`, `integration.test.js`, `message-size-limits.test.js`, `validator.test.js`) cannot even load, let alone run. + +This is not a refactor. This is a reconstruction of a broken module using the surviving test expectations and the other working modules as behavioral contracts. + +## Problem statement +Reconstruct `src/server.js` so that it exports exactly one `createServer` function which: + +1. Binds a `WebSocketServer` on the configured port with `maxPayload`. +2. On each connection: extracts the IP from `req.socket.remoteAddress`, checks the per-IP connection rate limiter (`createConnRateLimiter`), parses the auth token from the query string, and calls `verifyConnection(token)`. If auth fails, closes with code 4001. If connection rate limit is exceeded, closes with code 4029. +3. Resolves the effective `clientId` from the auth result (`authResult.clientId ?? clientId` where `clientId` is the uuid generated per connection). +4. Sets up per-message rate limiting using `createRateLimiter` (imported from `./rate-limiter.js`), checked on every incoming message before validation. +5. Validates each incoming message via `validateMessage` from `./validator.js`. On validation failure, sends `{ type: "error", payload: { message: "" } }` back to the client. +6. Routes validated messages through the room manager: `join_room`, `leave_room`, and `location_update` (fan-out to all rooms the client belongs to, excluding the sender). +7. On disconnect: calls `rooms.disconnect(actualClientId)` and decrements the per-IP connection count. +8. Sets up heartbeat ping/pong via `setupHeartbeat` that terminates zombie connections (`ws.isAlive === false`). +9. Returns `{ wss, rooms }` (not `ipConnectionCount`). + +The scrambled code currently interleaves fragments of what was clearly a `handleConnection` function and a `handleMessage` function in the wrong order, with the `handleMessage` body referencing `req` which was only available in the connection handler scope. + +## Current behavior +`src/server.js` has: +- **Line 9**: First `export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit } = {})` — creates WSS, rooms, and connRateLimiter, defines `safeSend` and a scrambled `handleMessage`, then falls through to undefined variable references. +- **Line 26**: `function handleMessage(ws, clientId, rooms, raw)` — but the body references `req` (line 29), `ip` (line 30), `connRateLimiter` (line 31), `url` (line 37), `token` (line 46), `authResult` (line 47), `rateLimiter` (line 84, never defined anywhere), and `actualClientId` (line 92, assigned after first use on line 84). +- **Line 135**: Second `export function createServer({ port, heartbeatMs, maxPayloadBytes } = {})` — a cleaner but incomplete implementation that wires up `wss.on("connection", ...)` and `setupHeartbeat` but doesn't handle auth, rate limiting, or message routing. +- **Line 151**: Returns `{ wss, rooms, ipConnectionCount }` where `ipConnectionCount` is never declared. + +Additionally, `createRateLimiter` from `./rate-limiter.js` is never imported. The variable `rateLimiter` referenced on line 84 is undefined. The variable `ipConnectionCount` referenced on line 100 and returned on line 151 is never declared. + +## Required behavior +- Exactly one `export function createServer()` that accepts `{ port, heartbeatMs, maxPayloadBytes, connRateLimit }`. +- Imports `createRateLimiter` from `./rate-limiter.js`. +- Implements the full connection lifecycle described in the problem statement. +- All 8 currently-failing test suites pass. +- No references to undefined variables. +- `safeSend` is defined and used for all outbound messages. + +## Constraints +- Do not change the public API of `createServer` beyond adding the optional `connRateLimit` parameter (already tested). +- Do not change any file other than `src/server.js`. +- Do not add new npm dependencies. +- Do not modify any test file. +- The `handleMessage` function must be properly scoped — it should receive `ws`, `clientId`, `rooms`, `raw`, and the rate limiter instance (or a closure over it), but NOT `req`. +- Auth check (`verifyConnection`) must use the token extracted from `req.url` query params, not from `raw`. + +## Acceptance criteria +- [ ] `src/server.js` exports exactly one `createServer` function +- [ ] No `SyntaxError` when importing `src/server.js` +- [ ] `npm run lint` passes with zero errors +- [ ] `npm test` passes: all 18 test suites green (166+ tests) +- [ ] `server.test.js`: connection with valid token accepted, missing token rejected with 4001, invalid JSON returns error frame, join_room returns room_joined, leave_room returns room_left, location_update broadcasts to room members, sender excluded from own broadcast, disconnect cleans up room membership +- [ ] `binary-frames.test.js`: binary Buffer frames accepted for join_room and location_update, invalid JSON in binary frame returns error +- [ ] `conn-rate-limiter.test.js`: connections exceeding per-IP rate limit rejected with 4029 +- [ ] `max-connections.test.js`: connections exceeding per-IP max rejected with 4029 + +## Out of scope +- Changes to `auth.js`, `validator.js`, `room-manager.js`, `rate-limiter.js`, `conn-rate-limiter.js`, `logger.js`, or any test file. +- Adding new message types or protocol features. +- Database integration. +- TLS or HTTP server setup. + +## Hints and references +- The test file `tests/server.test.js` defines the exact behavioral contract: `connect()` helper builds `ws://localhost:${port}/?token=${token}`, `nextMessages()` collects N messages, `closeAll()` tears down sockets. Study these helpers to understand the expected protocol. +- The variable `clientId` should be generated with `uuid.v4()` (already imported but unused in the second `createServer`). +- The `rateLimiter` (per-message) should be a module-level `createRateLimiter()` instance, not per-connection, since the rate limit is per-client-id and the same client ID could theoretically reconnect. +- The `ipConnectionCount` tracking in the close handler (lines 98-108 of current file) is for a max-connections-per-IP feature. The test `max-connections.test.js` expects a `maxConnectionsPerIp` option. You must implement this as an in-process Map tracking active connection count per IP, incremented on connection and decremented on close. diff --git a/issues/issue-02-reconstruct-validator-pipeline.md b/issues/issue-02-reconstruct-validator-pipeline.md new file mode 100644 index 0000000..b72d003 --- /dev/null +++ b/issues/issue-02-reconstruct-validator-pipeline.md @@ -0,0 +1,61 @@ +## Title +Reconstruct the corrupted `validator.js` — resolve duplicate declarations and undefined helper references to restore the dual-input validation pipeline + +## Difficulty +9/10 — Expert. Estimated effort: 2–3 days for a senior engineer. + +## Context +`src/validator.js` has two competing `export function validateMessage()` declarations (lines 33–41 and 63–84). Because ES modules do not allow duplicate exports of the same name, this module fails at parse time with `SyntaxError: Identifier 'validateMessage' has already been declared`. This breaks every test suite that imports the validator: `validator.test.js`, `integration.test.js`, `message-size-limits.test.js`, and transitively `server.test.js` (which imports `validateMessage` indirectly through `server.js`). + +The first definition (lines 33–41) is incomplete — it parses JSON but never invokes the Zod schema, never returns a result for valid input, and doesn't enforce per-type byte size limits. The second definition (lines 63–84) is more complete but references three functions that do not exist anywhere in the file: `parseJSON` (line 64), `isString` (line 67), and `formatIssue` (line 80). A standalone `buildError` helper exists (lines 49–51) but is never called — `formatIssue` is used instead. + +## Problem statement +Reconstruct `src/validator.js` into a single, correct validation pipeline that: + +1. Accepts either a raw string (from WebSocket `Buffer.toString()`) or a pre-parsed object (from tests that call `validateMessage` with an object directly). +2. Parses JSON from strings; passes objects through. +3. Enforces per-type byte size limits **only** when the input is a string (binary frame → string). The limits are defined in `MESSAGE_SIZE_LIMITS`: `location_update` ≤ 512 bytes, `join_room` ≤ 256 bytes, `leave_room` ≤ 256 bytes. +4. Validates the parsed object against `messageSchema` (the Zod discriminated union). +5. Returns `{ ok: true, data }` on success or `{ ok: false, error }` on failure. +6. Exports `validateMessage` exactly once and exports the `buildError` helper. + +## Current behavior +- **Line 33**: First `validateMessage(raw)` — calls `JSON.parse` on strings, but the function body ends at line 41 without returning a validation result or calling the Zod schema. +- **Line 63**: Second `validateMessage(raw)` — calls `parseJSON(raw)` (undefined), checks `isString` (undefined), uses `formatIssue` (undefined) instead of the existing `buildError` helper. +- **Line 49**: `buildError(issues)` is defined but never referenced. +- **Lines 1–25**: The Zod schemas (`locationPayloadSchema`, `joinRoomSchema`, `leaveRoomSchema`, `messageSchema`) and `MESSAGE_SIZE_LIMITS` are correct and should be preserved. + +## Required behavior +- `validateMessage` exported once, accepting `string | unknown`. +- Returns `{ ok: true, data }` where `data` matches the Zod discriminated union. +- Returns `{ ok: false, error: string }` for: invalid JSON, oversized messages (string input only), schema validation failures. +- `buildError` exported for external use. +- All Zod schemas and size limits preserved exactly as-is. + +## Constraints +- Do not change any file other than `src/validator.js`. +- Do not add new npm dependencies. +- Do not modify any test file. +- The byte size check must use `Buffer.byteLength(raw, "utf8")` against the original raw string, not the parsed object. +- The `MESSAGE_SIZE_LIMITS` object must remain exactly as defined (512/256/256). +- Schema validation errors should be human-readable, joined with `"; "`. + +## Acceptance criteria +- [ ] `src/validator.js` exports exactly one `validateMessage` function +- [ ] No `SyntaxError` when importing `src/validator.js` +- [ ] `npm run lint` passes +- [ ] `validator.test.js` passes: all location_update, join_room, leave_room, and malformed JSON tests green +- [ ] `message-size-limits.test.js` passes: all per-type size limit tests green +- [ ] `integration.test.js` passes: valid messages route through validator → room manager, invalid messages are rejected +- [ ] `validateMessage("not json")` returns `{ ok: false, error: "Invalid JSON" }` +- [ ] `validateMessage({ type: "location_update", payload: { latitude: 40, longitude: -74 } })` returns `{ ok: true, data: ... }` + +## Out of scope +- Changes to `server.js`, `auth.js`, `room-manager.js`, or any test file. +- Adding new message types to the schema. +- Changing the Zod schema definitions or size limits. + +## Hints and references +- The `isString` check in the second `validateMessage` was checking `typeof raw === "string"` before the JSON parse, to decide whether to apply byte size limits. After parsing, you no longer have the original string to measure — so the size check must happen **before** parsing, using the raw input directly. +- The `parseJSON` helper should attempt `JSON.parse` for strings and pass objects through, returning `{ ok: true, data }` or `{ ok: false, error: "Invalid JSON" }`. +- The `formatIssue` that was referenced was likely meant to be a single-issue formatter; `buildError` already does this for arrays. Use `buildError(result.error.issues)` for the Zod failure path. diff --git a/issues/issue-03-rate-limiter-memory-leak.md b/issues/issue-03-rate-limiter-memory-leak.md new file mode 100644 index 0000000..64d7e67 --- /dev/null +++ b/issues/issue-03-rate-limiter-memory-leak.md @@ -0,0 +1,72 @@ +## Title +Eliminate the unbounded memory leak in sliding-window rate limiters by implementing time-bucketed eviction with amortized O(1) cleanup + +## Difficulty +10/10 — Expert. Estimated effort: 3–4 days for a senior engineer. + +## Context +Both `src/rate-limiter.js` (per-message, 1-second window) and `src/conn-rate-limiter.js` (per-IP connection, 60-second window) store a `Map` of timestamps. Stale entries are only evicted during an active `check()` call for that specific key — if a client connects once and never reconnects, or sends a burst and disconnects, their entry persists in the Map forever. In a fleet-tracking deployment with 10,000+ devices each sending 1 update/second, and with device reconnections creating new client IDs, this is an unbounded O(n) memory leak measured in megabytes per hour. + +The `remove()` method on `createRateLimiter` exists but is never called from `server.js` (which itself is broken and doesn't wire up disconnect cleanup for the per-message rate limiter). Even if it were called, `createConnRateLimiter` has no `remove()` method at all. + +This issue is **not** about fixing `server.js` — it is about fixing the rate limiters themselves so they are safe by construction, regardless of whether the caller remembers to call `remove()`. + +## Problem statement +Implement bounded, self-cleaning sliding-window rate limiters for both modules that: + +1. **Guarantee memory is bounded**: No rate limiter instance should hold more than `K` entries where `K` is proportional to the number of active clients (not the historical total). +2. **Evict stale entries proactively**: Entries older than the window (1 second for per-message, 60 seconds for per-IP) must be evicted without requiring a `check()` call for that key. +3. **Maintain O(1) amortized `check()`**: The per-call cost must not degrade below O(1) amortized, even with eviction running. +4. **Expose cleanup methods**: Both limiters must expose a `cleanup()` or `evict()` method that removes all stale entries, callable from a periodic timer or disconnect handler. +5. **Expose metrics**: Both limiters must expose `size` (current entry count) for observability. + +## Current behavior +`src/rate-limiter.js` (lines 34–42): +```js +let timestamps = windows.get(clientId); +if (!timestamps) { + timestamps = []; + windows.set(clientId, timestamps); +} +while (timestamps.length > 0 && timestamps[0] <= cutoff) { + timestamps.shift(); // O(n) shift on each check +} +``` +The `shift()` is O(n) in the worst case. The Map only shrinks when `check()` is called for that specific key. `remove()` exists but is never called in production. + +`src/conn-rate-limiter.js` (lines 14–22): Identical pattern, identical problems, no `remove()` method. + +## Required behavior +- Both `createRateLimiter` and `createConnRateLimiter` must not accumulate unbounded entries. +- Each must expose `cleanup()` that iterates and removes entries where the oldest timestamp is older than the window. +- Each must expose `size` getter returning the current Map size. +- The `shift()` pattern must be replaced with a more efficient approach (e.g., ring buffer, or batch cleanup that replaces the array rather than shifting). +- Memory usage must be measurable: a test that creates 100K entries, calls `cleanup()`, and verifies the Map shrinks to ≤ the number of entries with recent timestamps. + +## Constraints +- Do not change the public API: `check(clientId)` and `remove(clientId)` on `createRateLimiter`, `check(ip)` on `createConnRateLimiter`. +- Do not add new npm dependencies. +- Do not modify `server.js` or any test file (except adding new test files if needed for the new behavior). +- The sliding window semantics must remain identical: `maxPerSecond` messages in a 1-second window for per-message; `maxPerMinute` connections in a 60-second window for per-IP. +- `check()` must still record the timestamp on success and return `false` on limit exceeded — no behavior change for callers. + +## Acceptance criteria +- [ ] `createRateLimiter(5).check("c1")` returns `true` for 5 calls and `false` on the 6th (existing behavior preserved) +- [ ] `createConnRateLimiter(3).check("1.1.1.1")` returns `true` for 3 calls and `false` on the 4th (existing behavior preserved) +- [ ] After inserting 100K entries with old timestamps, calling `cleanup()` reduces Map size to ≤ entries with timestamps within the window +- [ ] `size` getter returns current entry count +- [ ] `rate-limiter.test.js` passes (all 7 tests) +- [ ] `conn-rate-limiter.test.js` passes (all 4 createConnRateLimiter tests — skip the server tests) +- [ ] Memory test: a test that calls `check()` for 50K unique keys, waits >1 second, calls `cleanup()`, and asserts `size` dropped by ≥ 90% +- [ ] No `O(n)` `Array.shift()` in hot path (lint or code review) + +## Out of scope +- Changes to `server.js`, `auth.js`, `validator.js`, `room-manager.js`, `index.js`, or `logger.js`. +- Fixing the disconnect cleanup wiring in `server.js` (that is issue 1's responsibility). +- Implementing token-bucket or leaky-bucket algorithms — this is specifically about fixing the sliding-window implementation. + +## Hints and references +- A ring buffer (fixed-size circular array with head/tail pointers) replaces `Array.shift()` with an O(1) pointer bump, and `cleanup()` becomes a single pointer advance to the first non-expired entry. +- Alternatively, batch-replace the array: instead of `shift()` one at a time, after `check()` completes, filter the array in one pass and replace the Map value. This is O(k) where k is the number of expired entries, amortized over all `check()` calls in that window. +- The `cleanup()` method is useful for a periodic `setInterval` in `server.js` (e.g., every 10 seconds) to prevent unbounded growth even for keys that are never checked again. +- Consider: what happens if `cleanup()` runs concurrently with `check()` on the same key? In Node.js single-threaded runtime this cannot happen within a single tick, but if `cleanup()` is async-aware or yields, it could. Keep it synchronous and atomic per tick. diff --git a/issues/issue-04-broadcast-backpressure.md b/issues/issue-04-broadcast-backpressure.md new file mode 100644 index 0000000..1b2971b --- /dev/null +++ b/issues/issue-04-broadcast-backpressure.md @@ -0,0 +1,80 @@ +## Title +Implement backpressure-aware async broadcast with per-client drain queues to prevent event loop starvation in large rooms + +## Difficulty +10/10 — Expert. Estimated effort: 4–5 days for a senior engineer. + +## Context +`RoomManager.broadcast()` in `src/room-manager.js` (lines 97–115) serializes the message to JSON once, then iterates every member and calls `ws.send(data)` synchronously. In a fleet-tracking deployment, a single room (e.g., `fleet-alpha`) can contain 5,000 vehicles. Each vehicle publishes a location update every second. When vehicle X publishes, `broadcast()` must call `ws.send()` 4,999 times synchronously. At 10 updates/second across the fleet, that is 50,000 synchronous `ws.send()` calls per second — all blocking the Node.js event loop. + +The `ws` library's `ws.send(data)` is not truly synchronous for TCP — it buffers into the kernel write buffer. When the kernel buffer fills (slow consumer), `ws.send()` internally queues the message. But the `RoomManager` has no awareness of this: it will keep pushing messages into the internal buffer of a slow client, growing memory without bound until the client is OOM-killed or the server runs out of memory. + +There is no write backpressure detection, no per-client send queue with configurable high-water mark, and no mechanism to pause or drop messages for slow consumers. This is a correctness and stability issue, not just a performance optimization. + +## Problem statement +Redesign `RoomManager.broadcast()` to be backpressure-aware and non-blocking, such that: + +1. **Event loop is not blocked**: `broadcast()` must not iterate N clients synchronously. It must yield control back to the event loop periodically (e.g., every M sends). +2. **Slow consumers are detected**: When a client's internal send buffer exceeds a configurable high-water mark (e.g., 1MB), the client is flagged as slow. Subsequent broadcasts to that client are either queued with a bounded buffer or dropped (configurable policy). +3. **Slow consumers are eventually evicted**: If a slow client does not drain within a configurable timeout (e.g., 30 seconds), the connection is terminated. +4. **Message coalescing for location updates**: When a slow client is catching up, intermediate location updates can be coalesced — only the most recent location per client needs to be delivered. This requires the broadcast path to know the message type. +5. **Metrics are exposed**: The `RoomManager` must expose per-room and per-client send queue depths for observability. + +## Current behavior +`src/room-manager.js` lines 97–115: +```js +broadcast(roomId, message, excludeClientId = null) { + const room = this._rooms.get(roomId); + if (!room) return; + const data = typeof message === "string" ? message : JSON.stringify(message); + for (const [clientId, ws] of room) { + if (clientId === excludeClientId) continue; + if (ws != null && ws.readyState === WebSocket.OPEN) { + try { ws.send(data); } catch { } + } + } +} +``` +Problems: +- Synchronous loop over all members — blocks event loop for O(N) sends. +- No backpressure detection — `ws.send()` can buffer indefinitely. +- No slow consumer handling — one slow client causes memory growth for the entire server. +- No message coalescing — a client receiving 100 location updates/sec behind a slow connection gets all 100, even though only the latest matters. + +## Required behavior +- `broadcast()` must be non-blocking: it must process sends in batches (e.g., 100 per tick) using `setImmediate` or `queueMicrotask` batching. +- Each client connection must have an associated send queue with a configurable `highWaterMark` (bytes). +- When a client's send queue exceeds `highWaterMark`, the client is marked as "slow" and subsequent location_update broadcasts to that client coalesce — only the latest location is kept. +- A slow client that doesn't drain within `slowConsumerTimeout` ms is terminated via `ws.close(4000, "Slow consumer")`. +- `getRoomStats(roomId)` returns `{ memberCount, sendQueueDepths: { [clientId]: number }, slowConsumers: string[] }`. + +## Constraints +- Do not change the `RoomManager` public API for `join`, `leave`, `disconnect`, `getClientRooms`, `getRoomSize` — existing callers must work unchanged. +- The new backpressure behavior must be opt-in via constructor options (defaulting to the current synchronous behavior for backward compatibility in tests). +- Do not add new npm dependencies. +- Do not modify any existing test file. +- The `ws` library's `bufferedAmount` property is available on WebSocket instances and returns the number of bytes queued in the outgoing buffer — use it for backpressure detection. +- Must not introduce memory leaks: slow consumer queues must be bounded and cleaned up on disconnect. + +## Acceptance criteria +- [ ] `RoomManager` constructor accepts `{ backpressure: { enabled: boolean, highWaterMark: number, slowConsumerTimeout: number, batchSize: number } }` with sane defaults +- [ ] When `backpressure.enabled === false` (default), `broadcast()` behaves identically to current implementation +- [ ] When `backpressure.enabled === true`, `broadcast()` processes sends in batches of `batchSize` using `setImmediate` yielding between batches +- [ ] A client whose `ws.bufferedAmount > highWaterMark` is flagged as slow +- [ ] Slow clients receive only the latest location_update (message coalescing) — intermediate updates are dropped +- [ ] Slow clients are terminated after `slowConsumerTimeout` ms +- [ ] `getRoomStats(roomId)` returns correct member count, queue depths, and slow consumer list +- [ ] Disconnect cleanup removes slow consumer tracking state (no leak) +- [ ] Existing `room-manager.test.js` (11 tests), `room-manager-extended.test.js` (11 tests), `room-manager-additional.test.js` (8 tests) all pass unchanged +- [ ] Performance test: `broadcast()` to 10,000 mock clients completes without blocking the event loop for more than 50ms per batch (measured via `setImmediate` timing) + +## Out of scope +- Changes to `server.js`, `validator.js`, `auth.js`, `rate-limiter.js`, `index.js`. +- Implementing a full pub/sub system with topic hierarchies. +- WebSocket compression (permessage-deflate). + +## Hints and references +- `ws` WebSocket instances expose `bufferedAmount` (readonly) — the number of bytes of data queued for delivery. This is the canonical backpressure signal per the WebSocket spec (RFC 6455 §5.2). +- The `backpressureOptions` should be stored on the `RoomManager` instance, not as module-level globals, to support testing with different configurations. +- For message coalescing, maintain a `Map` of latest pending messages per slow client. On each broadcast, overwrite the entry. A drain loop (triggered by `ws.on("drain")` or periodic timer) serializes and sends the latest entry. +- Consider: `ws.send()` returns `false` when the internal buffer is full (backpressure signal). But the `ws` library v8+ doesn't consistently expose this — check `bufferedAmount` instead. diff --git a/issues/issue-05-graceful-shutdown-draining.md b/issues/issue-05-graceful-shutdown-draining.md new file mode 100644 index 0000000..9e6eb82 --- /dev/null +++ b/issues/issue-05-graceful-shutdown-draining.md @@ -0,0 +1,86 @@ +## Title +Implement connection draining with pending message flush for graceful shutdown — prevent data loss during deploys and SIGTERM + +## Difficulty +9/10 — Expert. Estimated effort: 2–3 days for a senior engineer. + +## Context +The current `shutdown()` function in `src/index.js` (lines 43–53) calls `wss.close()` and sets a 5-second force-exit timer. `wss.close()` stops accepting new connections but does **not** close existing client connections or wait for pending sends to complete. In a fleet-tracking deployment, at any given moment there may be hundreds of in-flight location broadcasts queued in WebSocket send buffers. When `wss.close()` is called, these pending bytes are silently dropped. The 5-second force exit then kills the process before clients can receive close frames. + +This means every deploy (rolling update, Docker restart, Kubernetes pod eviction) causes: +1. **Silent data loss** of all pending location updates. +2. **No close frame sent** to clients — they must wait for TCP timeout (typically 30–120 seconds) to detect the disconnect, during which they appear as "connected" but receive nothing. +3. **No room membership cleanup events** — other clients in the same rooms don't know a member disconnected until the heartbeat reaper catches it. + +For a fleet-tracking system, this means vehicles appear frozen on the map for up to 2 minutes after a deploy, and location history has gaps. + +## Problem statement +Implement a multi-phase graceful shutdown that: + +1. **Phase 1 — Stop accepting** (0–100ms): Close the HTTP upgrade listener so no new WebSocket connections are established. Existing connections remain open. +2. **Phase 2 — Notify clients** (100ms–500ms): Send a `{ type: "server_shutting_down", payload: { reconnectIn: N } }` message to all connected clients, giving them time to reconnect to another instance. +3. **Phase 3 — Drain pending sends** (500ms–4000ms): For each connected client, flush any buffered outgoing data. Use `ws.bufferedAmount === 0` as the drain signal. Clients that haven't drained within this window are terminated. +4. **Phase 4 — Close connections** (4000ms–5000ms): Send WebSocket close frames (code 1001 "Going Away") to all remaining clients. Wait for close acknowledgements. +5. **Phase 5 — Force exit** (>5000ms): If any clients are still connected, force `process.exit(1)`. + +The shutdown must also emit structured log entries at each phase transition for observability. + +## Current behavior +`src/index.js` lines 43–53: +```js +function shutdown(signal) { + logger.info("Shutting down", { signal }); + wss.close(() => { + logger.info("Server closed"); + process.exit(0); + }); + setTimeout(() => { + logger.error("Forced shutdown"); + process.exit(1); + }, 5000); +} +``` +- `wss.close()` stops accepting but doesn't close existing connections. +- No drain phase — pending sends are dropped. +- No client notification — clients don't know the server is shutting down. +- The 5-second timer fires unconditionally — even if clients have already disconnected. +- No structured log entries per phase. + +## Required behavior +- `shutdown(wss, signal)` implements the 5-phase draining protocol. +- All connected clients receive a shutdown notification before connections are closed. +- Pending send buffers are drained with a bounded timeout. +- Clients receive proper close frames (code 1001). +- The process exits 0 on clean shutdown, 1 on forced. +- Structured log entries at each phase: "shutdown: stopping accept", "shutdown: notifying N clients", "shutdown: draining N clients", "shutdown: closing N clients", "shutdown: force exit". + +## Constraints +- Do not change the `SIGTERM`/`SIGINT` signal handlers — they must still call `shutdown()`. +- Do not add new npm dependencies. +- Do not modify `server.js` — `shutdown()` receives the `wss` instance and operates on it. +- Do not modify existing test files. The existing `shutdown` tests in `index.test.js` must still pass (they test the current behavior with a mock WSS — ensure backward compatibility). +- The total shutdown timeline must not exceed 5 seconds (the existing force-exit budget). +- Must handle: clients that never respond to close frames, clients with broken TCP connections, clients that reconnect during shutdown. + +## Acceptance criteria +- [ ] `index.test.js` existing tests pass: `shutdown()` calls `wss.close()` and `process.exit(0)`, force exit after 5s if not closed +- [ ] New test: `shutdown()` sends shutdown notification to all connected clients before closing +- [ ] New test: `shutdown()` waits for `bufferedAmount === 0` before closing each client +- [ ] New test: `shutdown()` sends close frame with code 1001 +- [ ] New test: total shutdown time does not exceed 5 seconds even with unresponsive clients +- [ ] New test: structured log entries emitted at each phase transition +- [ ] `npm run lint` passes +- [ ] `npm test` passes (all suites) + +## Out of scope +- Changes to `server.js`, `room-manager.js`, `auth.js`, `validator.js`, `rate-limiter.js`, `logger.js`. +- Implementing a health check endpoint (that is a separate concern). +- Zero-downtime deploy orchestration (that requires a load balancer and is outside this service's scope). +- Persisting pending messages to disk or a queue during shutdown. + +## Hints and references +- `wss.clients` is a `Set` of all connected clients — iterate it for the drain phase. +- `ws.bufferedAmount` (RFC 6455 §5.2) returns bytes queued. Poll it with `setInterval(100)` during the drain phase. +- WebSocket close code 1001 ("Going Away") is the standard code for server-initiated shutdown (RFC 6455 §7.4.1). +- The notification message `{ type: "server_shutting_down" }` is not part of the current protocol — you'll need to add it to the `messageSchema` in `validator.js`... **except** issue constraints say not to modify `validator.js`. Solution: send the notification as a raw JSON string bypassing validation (use `ws.send(JSON.stringify(...))` directly, not through the validated message path). +- Consider: what if `wss.close()` is called while a drain is in progress? The existing test expects `wss.close` to be called. Ensure both the drain AND the `wss.close()` happen. diff --git a/issues/issue-06-client-reconnection-replay.md b/issues/issue-06-client-reconnection-replay.md new file mode 100644 index 0000000..86e4d6f --- /dev/null +++ b/issues/issue-06-client-reconnection-replay.md @@ -0,0 +1,79 @@ +## Title +Implement server-side message sequencing, bounded ring-buffer replay, and reconnection handshake for gap-free location delivery + +## Difficulty +10/10 — Expert. Estimated effort: 5–7 days for a senior engineer. + +## Context +The current protocol has no message ordering, no sequence numbers, and no replay mechanism. When a client disconnects (network handoff, app backgrounding, TCP reset) and reconnects, it has no way to know what messages it missed. In a fleet-tracking deployment, this means: + +1. A vehicle driving through a cell tower handoff loses 5–30 seconds of location history. +2. A dispatcher viewing the fleet map sees the vehicle "teleport" from the last known position to the current one, with no intermediate points. +3. Geofence enforcement is violated silently — the vehicle may have crossed a boundary during the gap, but the server never delivered the crossing event. +4. There is no deduplication — if a client sends the same location_update twice (e.g., retry after timeout), it's broadcast twice to all subscribers. + +The `project.md` describes the system as targeting "geofencing enforcement" and "fleet tracking" — both require reliable, ordered, gap-free delivery as a core correctness property. + +## Problem statement +Implement a message sequencing and replay subsystem that: + +1. **Sequence numbering**: Every broadcast message gets a monotonically increasing sequence number per room. The sequence is a 64-bit integer (safe for `Number.MAX_SAFE_INTEGER` at 9,007,199,254,740,991 — more than enough for centuries of updates). +2. **Bounded ring buffer per room**: Each room maintains a ring buffer of the last N broadcast messages (configurable, default 1,000). The buffer stores `{ seq, payload, timestamp }`. +3. **Reconnection handshake**: When a client reconnects, it sends `{ type: "reconnect", roomId: string, lastSeq: number }`. The server responds with: + - If `lastSeq` is within the buffer range: `{ type: "replay", roomId: string, messages: [...], currentSeq: number }` containing all missed messages. + - If `lastSeq` is too old (buffer has rotated past it): `{ type: "replay_gap", roomId: string, fromSeq: number, currentSeq: number }` — the client knows it has an unrecoverable gap. + - If `lastSeq === currentSeq`: `{ type: "replay_complete", roomId: string }` — no messages missed. +4. **Deduplication**: Broadcast deduplicates messages by `(roomId, clientId, timestamp)` within a 5-second TTL window. Duplicate location_updates from the same client with the same timestamp are silently dropped. +5. **Client-side tracking**: Each client tracks the last sequence number they received per room. This is maintained server-side (not client-reported for the initial implementation) by recording the last seq delivered to each `(clientId, roomId)` pair. + +## Current behavior +- No sequence numbers on any message. `broadcast()` in `room-manager.js` sends raw payloads. +- No ring buffer. Messages are fire-and-forget. +- No reconnection protocol. The only message types are `join_room`, `leave_room`, `location_update`. +- No deduplication. Identical messages are broadcast multiple times. +- The `messageSchema` in `validator.js` is a discriminated union of three types. Adding new types requires modifying the schema. + +## Required behavior +- Every `broadcast()` call increments a per-room sequence counter and stores the message in a bounded ring buffer. +- The reconnection handshake types (`reconnect`, `replay`, `replay_gap`, `replay_complete`) are added to the Zod schema in `validator.js`. +- Ring buffer size is configurable per `RoomManager` instance. +- Deduplication window is configurable (default 5 seconds). +- Existing `location_update`, `join_room`, `leave_room` behavior is unchanged. +- `getRoomSeq(roomId)` returns the current sequence number for a room. +- `getRingBuffer(roomId)` returns the contents of the ring buffer (for testing). + +## Constraints +- Ring buffer memory per room must be bounded: `bufferSize × averageMessageSize` must not exceed a configurable `maxBufferBytes` (default 5MB). +- Sequence numbers must be gap-free within the buffer — if the buffer is full, the oldest message is evicted and the gap is detectable via `replay_gap`. +- Deduplication must not add per-client state that grows unboundedly — use a time-bounded LRU or similar structure with a hard size cap. +- Do not add new npm dependencies. +- The existing 30 passing test suites must continue to pass unchanged (the new features are additive). +- New message types must be added to `validator.js`'s `messageSchema` discriminated union. + +## Acceptance criteria +- [ ] `RoomManager` constructor accepts `{ ringBufferSize: number, deduplicationWindowMs: number, maxBufferBytes: number }` with sane defaults +- [ ] `broadcast()` stores each message in the ring buffer with a monotonically increasing sequence number +- [ ] `getRoomSeq(roomId)` returns the current sequence number (0 if room has no broadcasts) +- [ ] `getRingBuffer(roomId)` returns an array of `{ seq, payload, timestamp }` in order +- [ ] Ring buffer is bounded: when full, oldest messages are evicted and `getRingBuffer()` length ≤ `ringBufferSize` +- [ ] `reconnect` message type is accepted by `validateMessage()` in `validator.js` +- [ ] Server responds to `reconnect` with correct `replay`, `replay_gap`, or `replay_complete` +- [ ] Deduplication: sending the same `location_update` twice within the dedup window results in only one broadcast +- [ ] Deduplication state does not grow beyond `maxDedupEntries` (configurable, default 10,000) +- [ ] `maxBufferBytes` is enforced: ring buffer memory ≤ configured limit +- [ ] All existing tests pass (room-manager, validator, server, integration, etc.) +- [ ] New tests: replay delivers correct missed messages, gap detection works, dedup drops duplicates, ring buffer eviction works + +## Out of scope +- Client-side implementation (sequence tracking, replay requests) — this is server-only. +- Persistent storage of ring buffers (in-memory only for this issue). +- Exactly-once delivery semantics across network partitions (this requires distributed consensus, which is out of scope). +- Historical trail / breadcrumb storage in a database. + +## Hints and references +- A ring buffer can be implemented as a fixed-size `Array` with a head pointer and modular arithmetic. `buffer[seq % bufferSize] = entry`. Eviction is automatic — old entries are overwritten. +- For the sequence number, use a `Map` of `roomId → currentSeq`. Increment on each `broadcast()`. This is a simple counter, not a complex data structure. +- The deduplication key `(roomId, clientId, timestamp)` can be stored in a `Map` where the key is `${roomId}:${clientId}:${timestamp}` and the value is `Date.now()`. Periodic cleanup removes entries older than `deduplicationWindowMs`. Use a `Set` with TTL-based expiry if you want O(1) lookup. +- For the `replay` response, serialize the ring buffer entries from `lastSeq + 1` to `currentSeq`. If `lastSeq < currentSeq - bufferSize`, the gap is unrecoverable. +- Consider: what if a client sends `reconnect` before `join_room`? The server should reject it with an error frame — the client must join a room first. +- The Zod schema addition for `reconnect` is straightforward: add a new variant to the discriminated union in `validator.js`. The response types (`replay`, `replay_gap`, `replay_complete`) are server-to-client only and don't need schema validation (they're never received from clients). diff --git a/issues/issue-07-jwt-auth-hardening.md b/issues/issue-07-jwt-auth-hardening.md new file mode 100644 index 0000000..09c0089 --- /dev/null +++ b/issues/issue-07-jwt-auth-hardening.md @@ -0,0 +1,83 @@ +## Title +Harden JWT authentication with claim validation, JWKS endpoint support, clock-skew tolerance, and token-refresh protocol + +## Difficulty +10/10 — Expert. Estimated effort: 4–5 days for a senior engineer. + +## Context +`src/auth.js` (lines 27–47) performs JWT verification but only checks signature and expiry. It does not validate the `iss` (issuer), `aud` (audience), or `nbf` (not-before) claims beyond the library defaults. In a microservices environment, this means a JWT minted by an unrelated service (e.g., an internal SSO that shares the same symmetric secret) would be accepted by the gateway. The `notBefore` check is delegated to the library, but there is no clock-skew tolerance — clients whose clocks are even 1 second ahead of the server receive `Token has expired` on tokens that are nominally valid. + +Additionally, there is no token-refresh mechanism. When a JWT expires, the client is disconnected with close code 4001 and must obtain a new token out-of-band. For mobile fleet-tracking clients that maintain persistent connections through cell handoffs, this means connection drops every `expiresIn` interval (commonly 1 hour), causing the gaps that issue 6 aims to fix. A proper refresh protocol allows the client to send a new token on the existing connection without disconnecting. + +Finally, the auth module has no support for asymmetric key verification (RSA/ES256 via JWKS endpoint). The current implementation only supports HMAC symmetric secrets. Production deployments behind an API gateway or identity provider (Auth0, Keycloak, AWS Cognito) require JWKS-based key rotation. + +## Problem statement +Extend `src/auth.js` to support: + +1. **Claim validation**: Validate `iss` (must match `AUTH_ISSUER` env var if set), `aud` (must match `AUTH_AUDIENCE` env var if set), and `nbf` (reject tokens not yet valid with a configurable clock-skew tolerance of ±30 seconds). +2. **JWKS endpoint support**: When `AUTH_JWKS_URI` is set, fetch the JWKS document (with caching and automatic refresh) and verify tokens using the matching `kid` header against the public key. Support both RSA (RS256) and EC (ES256) algorithms. +3. **Clock-skew tolerance**: Allow ±30 seconds of clock drift between client and server for `exp` and `nbf` checks. +4. **Token refresh protocol**: Add a new message type `token_refresh` that allows a connected client to send a new JWT on the existing connection. The server verifies the new token, updates the client's identity, and responds with `{ type: "token_refresh_ok" }` or `{ type: "error", payload: { message: "..." } }`. +5. **Algorithm restriction**: Reject tokens using the `none` algorithm and restrict to explicitly allowed algorithms (`HS256`, `RS256`, `ES256`). + +## Current behavior +`src/auth.js` lines 27–47: +```js +export function verifyConnection(token) { + const secret = process.env.AUTH_SECRET; + if (!secret) { + return { ok: false, error: "Server misconfiguration: AUTH_SECRET is missing" }; + } + try { + const payload = jwt.verify(token, secret); + return { ok: true, clientId: payload.sub ?? payload.clientId ?? "anonymous" }; + } catch (err) { + // ... error handling + } +} +``` +- No `iss` or `aud` claim validation. +- No JWKS support — only symmetric HMAC. +- No clock-skew tolerance — `jwt.verify` uses default options. +- No token refresh — client must disconnect and reconnect with a new token. +- No algorithm restriction — accepts whatever algorithm the JWT header specifies (algorithm confusion attack vector). + +## Required behavior +- `verifyConnection(token)` validates `iss`, `aud`, `nbf` with clock-skew tolerance when env vars are set. +- When `AUTH_JWKS_URI` is set, verification uses the public key from the JWKS document. +- `none` algorithm is explicitly rejected. +- Token refresh message type is accepted by the validator and processed by the server. +- All existing `auth.test.js` and `auth-extended.test.js` tests pass. +- New tests for: claim validation, JWKS verification, clock skew, token refresh, algorithm restriction. + +## Constraints +- Do not add new npm dependencies beyond the existing `jsonwebtoken`. For JWKS, implement the HTTP fetch and JWK-to-PEM conversion using Node.js built-in `crypto.createPublicKey` (available since Node 12) — do NOT add `jose` or `node-jose`. +- Do not break the existing `AUTH_SECRET` HMAC flow — it must remain the default when `AUTH_JWKS_URI` is not set. +- JWKS document must be cached with a configurable TTL (default 10 minutes) and refreshed in the background. +- Clock-skew tolerance must be ±30 seconds, configurable via `AUTH_CLOCK_SKEW_MS` env var. +- Do not modify `server.js` beyond adding the `token_refresh` message handling (or note that server.js is broken and this depends on issue 1). + +## Acceptance criteria +- [ ] `auth.test.js` passes (after accounting for the stale test — see below) +- [ ] `auth-extended.test.js` passes +- [ ] Token with invalid `iss` claim is rejected when `AUTH_ISSUER` is set +- [ ] Token with invalid `aud` claim is rejected when `AUTH_AUDIENCE` is set +- [ ] Token with `nbf` 31 seconds in the future is rejected +- [ ] Token with `exp` 29 seconds in the future (but past nominal expiry) is accepted with ±30s skew +- [ ] Token with `alg: "none"` is rejected +- [ ] JWKS verification works when `AUTH_JWKS_URI` is set (mock the HTTP endpoint in tests) +- [ ] Token refresh: connected client sends `token_refresh` with new JWT, receives `token_refresh_ok` +- [ ] Token refresh: invalid new token returns error frame, connection remains open +- [ ] `npm run lint` passes + +## Out of scope +- Changes to `room-manager.js`, `rate-limiter.js`, `conn-rate-limiter.js`, `logger.js`. +- Full OAuth2/OIDC integration (that is an identity provider concern). +- Token introspection endpoint (RFC 7662). +- Refresh token rotation (this is about refreshing the WS session token, not OAuth refresh tokens). + +## Hints and references +- The `jsonwebtoken` library's `jwt.verify` accepts an `algorithms` option: `jwt.verify(token, secret, { algorithms: ["HS256", "RS256", "ES256"] })`. Always pass this explicitly to prevent algorithm confusion attacks. +- For JWKS, the standard endpoint returns `{ keys: [{ kid, kty, n, e, ... }] }`. Use `crypto.createPublicKey({ format: "jwk", key: jwk })` (Node 16+) to convert JWK to a Node.js KeyObject. +- Clock-skew tolerance: subtract `clockTolerance` from `currentTimestamp` before comparing with `exp`, and add it before comparing with `nbf`. The `jsonwebtoken` library does not have a built-in clock-skew option — you must adjust the `clockTolerance` parameter manually or use the `clockTimestamp` option. +- The stale `auth.test.js` test (line 4–9) expects `verifyConnection(null)` to return `{ ok: true }` when `AUTH_SECRET` is empty string. But `auth.js` was hardened to fail-closed. This test needs to be updated to expect `{ ok: false }`. However, the issue constraint says not to modify test files — so ensure the test passes by making `AUTH_SECRET=""` behave as expected (empty string is falsy in JS, so the existing `if (!secret)` check already rejects it). The test will need updating — note this dependency. diff --git a/issues/issue-08-room-dos-protection.md b/issues/issue-08-room-dos-protection.md new file mode 100644 index 0000000..c7708a8 --- /dev/null +++ b/issues/issue-08-room-dos-protection.md @@ -0,0 +1,75 @@ +## Title +Implement room membership DoS protection with per-client room limits, per-room capacity caps, total room count ceiling, and resource-pressure circuit breaker + +## Difficulty +10/10 — Expert. Estimated effort: 3–4 days for a senior engineer. + +## Context +`RoomManager.join()` in `src/room-manager.js` (lines 51–58) accepts any `clientId` into any `roomId` with no limits whatsoever. A single misbehaving client can call `join_room` for 100,000 different room IDs, creating 100,000 entries in `_rooms` and 100,000 entries in `_clientRooms`. Each room entry is a `Map` (minimum ~100 bytes overhead per Map), and each client-room entry is a `Set` that grows by one per join. At 100K rooms, this is ~10MB of pure overhead — and that's from a single client. + +Conversely, a single room can accumulate unlimited members. A public "region-global" room could attract 50,000 clients, making every `broadcast()` call iterate 50,000 WebSocket objects — an O(N) event-loop-blocking operation that takes ~50ms per broadcast at 50K members (issue 4 addresses broadcast backpressure, but the root cause is the lack of room capacity limits). + +The `project.md` describes the system as supporting "fleet tracking, asset monitoring, geofencing enforcement, and live mapping." In a real deployment, rooms correspond to fleet IDs, regions, or user groups — all of which have natural cardinality and membership bounds. The absence of these bounds makes the system trivially vulnerable to resource exhaustion. + +## Problem statement +Implement a multi-layered resource accounting and DoS protection system for `RoomManager`: + +1. **Per-client room limit**: A single client cannot join more than `maxRoomsPerClient` rooms (configurable, default 50). Excess `join_room` attempts return `{ type: "error", payload: { code: "ROOM_LIMIT_EXCEEDED", message: "..." } }`. +2. **Per-room member limit**: A single room cannot have more than `maxMembersPerRoom` clients (configurable, default 10,000). Excess joins return `{ type: "error", payload: { code: "ROOM_FULL", message: "..." } }`. +3. **Total room count ceiling**: The `RoomManager` cannot hold more than `maxRooms` rooms (configurable, default 10,000). When the ceiling is hit, new room creation (first `join` to a non-existent room) is rejected with `{ type: "error", payload: { code: "MAX_ROOMS_REACHED", message: "..." } }`. +4. **Resource-pressure circuit breaker**: When total memory usage of the `RoomManager` (estimated via `roomCount × avgRoomSize + clientCount × avgClientRoomCount`) exceeds a configurable threshold, the circuit breaker opens. While open, all `join_room` operations are rejected with `{ type: "error", payload: { code: "CIRCUIT_BREAKER_OPEN", message: "..." } }`. The breaker closes automatically when memory pressure subsides (configurable recovery threshold). +5. **Metrics exposure**: `RoomManager` exposes `stats` getter returning `{ roomCount, clientCount, totalMembers, circuitBreakerState }`. + +## Current behavior +`src/room-manager.js` `join()` (lines 51–58): +```js +join(clientId, roomId, ws) { + if (clientId == null) throw new TypeError("clientId is required"); + if (roomId == null) throw new TypeError("roomId is required"); + if (ws == null) throw new TypeError("ws is required"); + this._ensureRoom(roomId).set(clientId, ws); + this._ensureClientRooms(clientId).add(roomId); +} +``` +No limits. No resource accounting. No circuit breaker. A single client can join unlimited rooms, and a single room can hold unlimited members. + +## Required behavior +- `RoomManager` constructor accepts `{ maxRoomsPerClient, maxMembersPerRoom, maxRooms, circuitBreaker: { enabled, memoryThresholdBytes, recoveryThresholdBytes } }` with sane defaults. +- `join()` checks all limits before adding membership and returns/throws structured error objects (not just `TypeError`). +- The circuit breaker uses `process.memoryUsage().heapUsed` to estimate memory pressure. +- All existing `room-manager.test.js`, `room-manager-extended.test.js`, `room-manager-additional.test.js` tests pass (they don't hit any limits with default settings). +- New tests for: limit enforcement, circuit breaker open/close transitions, metrics exposure. + +## Constraints +- Do not change the `join(clientId, roomId, ws)` method signature — callers pass the same arguments. +- `join()` must throw `TypeError` for null/undefined required args (existing behavior) AND return/throw structured errors for limit violations. +- Circuit breaker must not add per-request overhead beyond a single `process.memoryUsage()` call (which is ~1μs in Node.js). +- Do not add new npm dependencies. +- Do not modify `server.js` or any test file (except adding new test files). +- The `maxMembersPerRoom` limit must be enforced atomically — no TOCTOU race between checking the count and adding the member (Node.js single-threaded, so this is naturally satisfied, but the code must not yield between check and insert). + +## Acceptance criteria +- [ ] `join("c1", "r1", ws)` succeeds when client has < `maxRoomsPerClient` rooms +- [ ] `join("c1", "r2", ws)` throws/returns error when client already has `maxRoomsPerClient` rooms +- [ ] `join("c1", "r1", ws)` succeeds when room has < `maxMembersPerRoom` members +- [ ] `join("c2", "r1", ws)` throws/returns error when room already has `maxMembersPerRoom` members +- [ ] First join to a new room succeeds when `roomCount < maxRooms` +- [ ] First join to a new room fails when `roomCount === maxRooms` (existing room joins still work) +- [ ] Circuit breaker opens when `heapUsed > memoryThresholdBytes` +- [ ] Circuit breaker closes when `heapUsed < recoveryThresholdBytes` +- [ ] When circuit breaker is open, all `join()` calls are rejected +- [ ] `stats` getter returns `{ roomCount, clientCount, totalMembers, circuitBreakerState }` +- [ ] All 30 existing RoomManager tests pass unchanged +- [ ] `npm run lint` passes + +## Out of scope +- Changes to `server.js`, `auth.js`, `validator.js`, `rate-limiter.js`, `index.js`, `logger.js`. +- Implementing per-room message rate limiting (that is a broadcast concern). +- Dynamic limit adjustment based on system load (static limits are sufficient for this issue). + +## Hints and references +- `process.memoryUsage().heapUsed` returns the V8 heap usage in bytes. It's fast (~1μs) but not free — call it only on `join()`, not on every `broadcast()`. +- The circuit breaker pattern has three states: `CLOSED` (normal), `OPEN` (rejecting), `HALF_OPEN` (testing recovery). For this implementation, `CLOSED` and `OPEN` are sufficient — the breaker transitions directly from `OPEN` to `CLOSED` when memory drops below the recovery threshold. +- For `maxMembersPerRoom`, the check is simply `room.size >= maxMembersPerRoom` after `_ensureRoom(roomId)` — the room is guaranteed to exist at that point. +- For `maxRoomsPerClient`, check `_clientRooms.get(clientId)?.size ?? 0` before `_ensureClientRooms(clientId).add(roomId)`. +- For `maxRooms`, check `_rooms.size` before `_ensureRoom(roomId)` — but only when the room doesn't already exist: `!this._rooms.has(roomId) && this._rooms.size >= maxRooms`. diff --git a/issues/issue-09-health-metrics-http-server.md b/issues/issue-09-health-metrics-http-server.md new file mode 100644 index 0000000..cdbc8a7 --- /dev/null +++ b/issues/issue-09-health-metrics-http-server.md @@ -0,0 +1,77 @@ +## Title +Implement a co-located HTTP server with health check, readiness probe, Prometheus metrics exposition, and WebSocket upgrade protocol detection on a shared port + +## Difficulty +10/10 — Expert. Estimated effort: 4–5 days for a senior engineer. + +## Context +The gateway currently only binds a raw `WebSocketServer` on a port. There is no HTTP listener, no health check endpoint, and no metrics endpoint. This makes the service undeployable in any container orchestrator (Kubernetes, ECS, Nomad) that requires: + +- **Liveness probe**: An HTTP endpoint that returns 200 when the process is alive and not deadlocked. Without it, the orchestrator cannot detect a stuck process and will not restart it. +- **Readiness probe**: An HTTP endpoint that returns 200 only when the service is ready to accept traffic (all subsystems initialized, auth configured, rate limiters functional). Without it, traffic is routed to the instance before it's ready, causing connection failures. +- **Prometheus metrics**: A `/metrics` endpoint exposing connection count, message rate, room sizes, rate limit rejections, memory usage, and event loop lag. Without it, there is no observability into the running system. + +The `docker-compose.yml` includes a Postgres service and the `Dockerfile` exposes port 8080, but the application has no HTTP server. The README architecture diagram shows an "Observability" layer with "Structured logging" but no metrics endpoint. + +The hard part is **protocol detection on a shared port**: WebSocket connections start with an HTTP GET request with `Upgrade: websocket` header. A naive HTTP server would intercept these upgrade requests and break WebSocket connectivity. The solution requires inspecting the incoming request to determine if it's a WebSocket upgrade (pass through to WSS) or a plain HTTP request (handle as health/metrics). This is the core algorithmic challenge. + +## Problem statement +Implement an HTTP server that co-exists with the WebSocket server on the same port using protocol detection: + +1. **Protocol detection**: Create an `http.createServer` as the primary listener. On each incoming request, check for the `Upgrade: websocket` header. If present, pass the socket to the `WebSocketServer` for upgrade. If absent, handle as an HTTP request. +2. **Health endpoint** (`GET /healthz`): Returns `200 { "status": "ok", "uptime": }` when the process is alive. Always returns 200 unless the process is shutting down. +3. **Readiness endpoint** (`GET /readyz`): Returns `200 { "status": "ready", "connections": , "rooms": }` when all subsystems are initialized and the circuit breaker (if implemented) is closed. Returns `503 { "status": "not ready", "reason": "..." }` when not ready. +4. **Prometheus metrics** (`GET /metrics`): Returns Prometheus exposition format (text/plain) with: + - `gateway_connections_active` (gauge) + - `gateway_rooms_active` (gauge) + - `gateway_messages_total` (counter, labeled by `type`) + - `gateway_rate_limit_rejections_total` (counter, labeled by `kind`) + - `gateway_auth_failures_total` (counter) + - `gateway_heap_used_bytes` (gauge) + - `gateway_event_loop_lag_ms` (gauge) +5. **Graceful shutdown integration**: When `shutdown()` is called, the HTTP server stops accepting new connections (both HTTP and WebSocket) simultaneously. + +## Current behavior +`src/server.js` creates a `WebSocketServer` with `new WebSocketServer({ port })`. This directly binds a TCP listener. There is no HTTP server, no health endpoint, no metrics. The `index.js` shutdown function calls `wss.close()` which closes the underlying TCP listener. + +`src/index.js` creates the server with `createServer({ port, heartbeatMs, maxPayloadBytes })` and the `wss` is the only listener. + +## Required behavior +- `createServer()` returns `{ wss, httpServer, rooms, metrics }` where `httpServer` is the `http.Server` instance. +- WebSocket upgrade requests are correctly routed to the WSS. +- `/healthz`, `/readyz`, `/metrics` endpoints respond correctly. +- Prometheus metrics are exposed in standard exposition format. +- All existing tests pass (they create `WebSocketServer` on port 0 — you must ensure the HTTP server doesn't interfere with direct WSS creation in tests). +- Event loop lag is measured using a periodic `setTimeout` drift detector. + +## Constraints +- Do not add new npm dependencies. Use Node.js built-in `http` module. +- The HTTP server must be the primary listener — the `WebSocketServer` must be attached via `noServer: true` mode and the upgrade handled manually. +- Metrics counters must be atomically incremented (no race conditions — Node.js single-threaded, but ensure no counter corruption during async operations). +- The event loop lag detector must not add more than 1ms of overhead per measurement cycle. +- Do not modify existing test files. New test files for HTTP endpoints are allowed. +- `/metrics` must return valid Prometheus exposition format (no library — format manually). + +## Acceptance criteria +- [ ] `GET /healthz` returns `200` with `{ status: "ok", uptime: }` +- [ ] `GET /readyz` returns `200` when server is initialized +- [ ] `GET /readyz` returns `503` during shutdown +- [ ] `GET /metrics` returns `200` with `Content-Type: text/plain; version=0.0.4; charset=utf-8` +- [ ] `/metrics` output contains `gateway_connections_active`, `gateway_rooms_active`, `gateway_messages_total`, `gateway_heap_used_bytes`, `gateway_event_loop_lag_ms` +- [ ] WebSocket connections still work correctly on the same port (all server.test.js tests pass) +- [ ] HTTP requests to non-upgrade paths do not interfere with WebSocket upgrade +- [ ] `npm run lint` passes +- [ ] New test file: HTTP endpoint tests for health, readiness, metrics + +## Out of scope +- Changes to `room-manager.js`, `auth.js`, `validator.js`, `rate-limiter.js`, `logger.js`. +- Alerting or dashboard configuration (that's a Grafana/Datadog concern). +- Distributed tracing (OpenTelemetry integration). +- Admin API for runtime configuration changes. + +## Hints and references +- The `ws` library supports `noServer: true` mode: `new WebSocketServer({ noServer: true })`. Then handle the `upgrade` event on the HTTP server: `httpServer.on("upgrade", (req, socket, head) => { wss.handleUpgrade(req, socket, head, (ws) => { wss.emit("connection", ws, req); }); })`. +- For event loop lag measurement: record `Date.now()` before and after a `setTimeout(0)`. The difference minus 0ms is the lag. Do this every 5 seconds. Store the result in a module-level variable. +- Prometheus format for a gauge: `# TYPE gateway_connections_active gauge\ngateway_connections_active 42\n`. For a counter: `# TYPE gateway_messages_total counter\ngateway_messages_total{type="location_update"} 12345\n`. +- The `uptime` in `/healthz` is `process.uptime()` — built into Node.js. +- Consider: the existing tests use `createServer({ port: 0 })` which creates a `WebSocketServer` directly. If you change `createServer` to return an HTTP server, the tests that do `server.wss.address().port` will break unless the HTTP server's port is used instead. Ensure the returned port is consistent. diff --git a/issues/issue-10-storage-adapter-postgres.md b/issues/issue-10-storage-adapter-postgres.md new file mode 100644 index 0000000..44d23f0 --- /dev/null +++ b/issues/issue-10-storage-adapter-postgres.md @@ -0,0 +1,90 @@ +## Title +Implement the pluggable Storage Adapter interface with PostgreSQL backend, write batching, spatial indexing, and backpressure-aware integration into the broadcast pipeline + +## Difficulty +10/10 — Expert. Estimated effort: 5–7 days for a senior engineer. + +## Context +The `README.md` architecture diagram (lines 43–62) explicitly shows a **Storage Adapter** layer between the Room Manager and "Postgres / Mongo / etc." The `Core Features` section (line 37) lists "Extensible Storage Adapter — Plug in your preferred database (PostgreSQL, MongoDB, InfluxDB, etc.) for persisting historical tracks." The `docker-compose.yml` provisions a PostgreSQL 16 instance with database `spatial_tracking` and user `tracker`. The `.env.example` includes `DATABASE_URL=`. + +**None of this exists.** There is zero database code anywhere in the codebase. No schema, no queries, no connection pooling, no adapter interface. The `docker-compose.yml` Postgres service sits unused. The `DATABASE_URL` env var is never read. The promised storage layer is entirely unimplemented. + +For a fleet-tracking system, this gap means: location updates are broadcast in real-time but never persisted. When a client reconnects (issue 6), there is no historical data to replay from. When a dispatcher wants to view a vehicle's route history, there is nothing to query. Geofence violation events have no audit trail. The system is a pure relay with amnesia — useful for live viewing, but useless for any after-the-fact analysis. + +## Problem statement +Implement a pluggable Storage Adapter with: + +1. **Adapter interface** (`StorageAdapter`): Define an abstract interface with methods: + - `writeBatch(events: LocationEvent[]): Promise` — persist a batch of location events. + - `queryRoom(roomId: string, options: { from?: Date, to?: Date, limit?: number }): Promise` — retrieve historical locations for a room. + - `querySpatial(bounds: { minLat, maxLat, minLon, maxLon }, options?: { limit?: number }): Promise` — spatial range query. + - `getLatest(roomId: string): Promise` — get the most recent location for a room. + - `close(): Promise` — clean shutdown. + +2. **PostgreSQL backend** (`PostgresAdapter`): Implement the adapter using `pg` (node-postgres) with: + - Connection pooling (configurable `poolSize`, default 10). + - Schema migration: auto-create the `location_events` table with columns `(id, client_id, room_id, latitude, longitude, altitude, accuracy, speed, timestamp, created_at)` and appropriate indexes. + - Spatial index: use PostGIS `geography` type or a GiST index on `(latitude, longitude)` for efficient range queries. + - Write batching: accumulate events in a buffer (configurable `batchSize`, default 100, `flushIntervalMs`, default 1000) and flush via `INSERT ... SELECT unnest(...)` for amortized round-trip cost. + +3. **In-memory backend** (`MemoryAdapter`): A trivial in-memory implementation for testing, implementing the same interface with `Array` storage and linear scan for spatial queries. + +4. **Backpressure integration**: The adapter's `writeBatch` must not block the event loop. If the database is slow (connection pool exhausted, network latency), the write buffer must grow to a configurable `maxBufferSize` and then start dropping oldest events (with a counter metric). The adapter must never cause the WebSocket broadcast pipeline to stall. + +5. **Configuration**: `STORAGE_ADAPTER` env var selects the backend (`"postgres"` | `"memory"` | `"none"`). `DATABASE_URL` configures the Postgres connection. `STORAGE_BATCH_SIZE`, `STORAGE_FLUSH_INTERVAL_MS`, `STORAGE_MAX_BUFFER_SIZE` tune the batching. + +## Current behavior +- `docker-compose.yml` lines 15–24: Postgres service exists but is unused. +- `.env.example` line 4: `DATABASE_URL=` is empty. +- No `package.json` dependency for `pg` or any database driver. +- No file in `src/` references `DATABASE_URL` or imports any database module. +- The `RoomManager.broadcast()` sends location updates to WebSocket clients but never persists them. + +## Required behavior +- `StorageAdapter` interface defined and exported from a new `src/storage/adapter.js`. +- `PostgresAdapter` implemented in `src/storage/postgres.js`. +- `MemoryAdapter` implemented in `src/storage/memory.js`. +- Factory function `createStorageAdapter(config)` in `src/storage/index.js` selects the backend based on config. +- Write batching accumulates events and flushes periodically or when `batchSize` is reached. +- Spatial query uses database-level indexing (not in-memory linear scan for Postgres). +- All writes are async and non-blocking — `writeBatch` returns a Promise that resolves when the batch is persisted, but the caller is not blocked from continuing broadcasts. +- `npm test` passes with the `MemoryAdapter` (used in tests). +- `npm run lint` passes. + +## Constraints +- Do not add `pg` to `package.json` for the initial implementation — use dynamic `import("pg")` only when `STORAGE_ADAPTER=postgres`. This keeps the dev/test dependency light. +- Do not change `server.js`, `room-manager.js`, `auth.js`, `validator.js`, `rate-limiter.js`, `conn-rate-limiter.js`, `logger.js`, or any existing test file. +- New test files for the storage adapters are allowed. +- The `MemoryAdapter` must implement the exact same interface as `PostgresAdapter` — tests must be runnable against both. +- The write buffer must be bounded: if `maxBufferSize` events are queued and the database is unreachable, the oldest events are dropped and a `storage_dropped_events_total` counter is incremented. +- The Postgres schema must use `TIMESTAMPTZ` for temporal columns and `DOUBLE PRECISION` for coordinates. + +## Acceptance criteria +- [ ] `src/storage/adapter.js` exports the `StorageAdapter` interface (as a documented JSDoc typedef or abstract class) +- [ ] `src/storage/postgres.js` exports `PostgresAdapter` that implements all 5 interface methods +- [ ] `src/storage/memory.js` exports `MemoryAdapter` that implements all 5 interface methods +- [ ] `src/storage/index.js` exports `createStorageAdapter(config)` factory +- [ ] `PostgresAdapter` auto-creates the `location_events` table with correct schema on first use +- [ ] `PostgresAdapter.writeBatch` batches inserts (uses `unnest` or similar bulk insert, not individual INSERT statements) +- [ ] `MemoryAdapter.writeBatch` stores events in an array, `queryRoom` returns filtered results, `querySpatial` does linear scan +- [ ] `getLatest(roomId)` returns the most recent event by timestamp +- [ ] Write buffer is bounded: after `maxBufferSize` queued events, oldest are dropped +- [ ] `storage_dropped_events_total` counter increments when events are dropped +- [ ] All existing tests pass unchanged +- [ ] New test file: `tests/storage-adapter.test.js` tests both adapters against the same interface contract +- [ ] New test file: `tests/storage-postgres.test.js` integration tests (skipped when `DATABASE_URL` is not set) + +## Out of scope +- Changes to existing source files (`server.js`, `room-manager.js`, etc.) — the storage integration into the broadcast pipeline is a separate issue. +- MongoDB, InfluxDB, or other backend implementations (only Postgres and in-memory). +- Schema migration tooling (the adapter creates the table if it doesn't exist — no Flyway/Alembic). +- Real-time CDC (Change Data Capture) from the database. +- Geo-fencing event detection (that's a separate service concern). + +## Hints and references +- The `pg` library's `Pool` class handles connection pooling: `new Pool({ connectionString: process.env.DATABASE_URL, max: 10 })`. +- For batched inserts in Postgres, use `INSERT INTO location_events (client_id, room_id, ...) SELECT * FROM unnest($1::text[], $2::text[], ...)` where the arrays are built from the batch. This is a single round-trip for up to thousands of rows. +- For spatial indexing without PostGIS, a composite index on `(latitude, longitude)` with a bounding-box query `(lat BETWEEN minLat AND maxLat AND lon BETWEEN minLon AND maxLon)` is sufficient for moderate scale. With PostGIS, use `ST_MakePoint(lon, lat)::geography` and `ST_Intersects`. +- The `MemoryAdapter` spatial query is simply `events.filter(e => e.latitude >= minLat && e.latitude <= maxLat && e.longitude >= minLon && e.longitude <= maxLon)`. This is O(n) but acceptable for testing. +- For the write buffer: use a `Array` as a FIFO queue. `push()` to enqueue, `splice(0, batchSize)` to dequeue a batch. When `length > maxBufferSize`, `splice(0, length - maxBufferSize)` to drop oldest. +- Dynamic import for `pg`: `const { Pool } = await import("pg")` inside the `PostgresAdapter` constructor. This allows the module to be loaded without `pg` installed when using `memory` or `none` adapters. From 7fda4ab1ca22833b0b3616dded7f2b0f6308f1c6 Mon Sep 17 00:00:00 2001 From: levoski1 Date: Tue, 21 Jul 2026 18:04:06 +0100 Subject: [PATCH 4/4] fix: correct indentation and move error handler outside close callback in server.js --- src/server.js | 29 +++++++++++++++-------------- 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/src/server.js b/src/server.js index d5fb02b..6f0b84f 100644 --- a/src/server.js +++ b/src/server.js @@ -102,21 +102,22 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit } }); - ws.on("close", (code, reason) => { - rooms.disconnect(actualClientId); - const trackedIp = ws._trackedIp; - if (trackedIp) { - const count = ipConnectionCount.get(trackedIp) ?? 1; - if (count <= 1) { - ipConnectionCount.delete(trackedIp); - } else { - ipConnectionCount.set(trackedIp, count - 1); + ws.on("close", (code, reason) => { + rooms.disconnect(actualClientId); + const trackedIp = ws._trackedIp; + if (trackedIp) { + const count = ipConnectionCount.get(trackedIp) ?? 1; + if (count <= 1) { + ipConnectionCount.delete(trackedIp); + } else { + ipConnectionCount.set(trackedIp, count - 1); + } } - } - logger.info("Client disconnected", { - clientId: actualClientId, - code, - reason: reason?.toString() ?? "unknown", + logger.info("Client disconnected", { + clientId: actualClientId, + code, + reason: reason?.toString() ?? "unknown", + }); }); ws.on("error", (err) => {