From ab756879b6cb09e73552a57be3bc435892734bac Mon Sep 17 00:00:00 2001 From: shinorh247 Date: Wed, 29 Jul 2026 15:06:37 +0000 Subject: [PATCH] feat: per-user concurrency limit Cap the number of in-flight concurrent requests per user/IP across all /api routes to prevent a single identity from exhausting the thread pool or DB connection pool. Changes: - src/middleware/perUserConcurrency.ts (new): createPerUserConcurrencyMiddleware factory + perUserConcurrency singleton. Uses an in-process Map counter; decrements on res finish/close; returns HTTP 429 with Retry-After: 1 and standardised error envelope on exceed. - src/config/env-schema.ts: MAX_CONCURRENT_REQUESTS_PER_USER env var (default 10) - src/index.ts: app.use('/api', perUserConcurrency) before all route handlers - .env.example: document MAX_CONCURRENT_REQUESTS_PER_USER - README.md: Per-user concurrency limit section - tests/perUserConcurrency.test.ts: 22 tests, 100% line coverage, 93% branch Closes #320 --- .env.example | 8 + README.md | 44 ++ src/config/env-schema.ts | 11 + src/index.ts | 7 + src/middleware/perUserConcurrency.ts | 231 +++++++++ tests/perUserConcurrency.test.ts | 689 +++++++++++++++++++++++++++ 6 files changed, 990 insertions(+) create mode 100644 src/middleware/perUserConcurrency.ts create mode 100644 tests/perUserConcurrency.test.ts diff --git a/.env.example b/.env.example index 178cc435..9fd6adf5 100644 --- a/.env.example +++ b/.env.example @@ -103,6 +103,14 @@ ANON_RATE_LIMIT_WINDOW_MS=60000 # Max anonymous requests per IP per window ANON_RATE_LIMIT_MAX=60 +# ── Per-user concurrency limit ──────────────────────────────────────────────── + +# Maximum number of in-flight (concurrent) /api requests allowed per user +# (or per IP for anonymous callers) at any given moment. Requests exceeding +# this cap receive HTTP 429 with `Retry-After: 1`. Set to a large value +# (e.g. 1000) to effectively disable the limit. (default: 10) +# MAX_CONCURRENT_REQUESTS_PER_USER=10 + # ── Invites rate limiting (per user, token bucket) ────────── # Max number of consecutive /api/invites requests allowed in the window (default: 60) diff --git a/README.md b/README.md index 6d4b69ae..c9cfe5ea 100644 --- a/README.md +++ b/README.md @@ -105,6 +105,50 @@ Behavior: - Webhook routes may opt into a larger limit of `1mb`. - Requests exceeding the configured limit return HTTP `413` with the standard error envelope, including correlation and request IDs. +## Per-user concurrency limit + +All `/api` routes are protected by a per-user in-flight request cap provided by `src/middleware/perUserConcurrency.ts`. + +**What it limits:** the number of *concurrent* (simultaneously in-flight) requests from a single identity — not throughput over a time window. This prevents a single misbehaving client from holding many long-lived connections open and exhausting the server's thread pool or database connection pool. + +**Identity resolution:** +1. `req.user.id` — set by `requireAuth` / `optionalAuth` +2. `req.user.stellarAddress` — fallback from the same middlewares +3. First hop of `X-Forwarded-For` / `req.socket.remoteAddress` — anonymous callers + +**When the limit is exceeded:** + +``` +HTTP 429 Too Many Requests +Retry-After: 1 +``` + +```json +{ + "error": { + "code": "concurrency_limit_exceeded", + "message": "Too many concurrent requests", + "retryAfter": 1 + } +} +``` + +**Configuration** (`MAX_CONCURRENT_REQUESTS_PER_USER`, default `10`): + +```bash +# .env +MAX_CONCURRENT_REQUESTS_PER_USER=10 +``` + +**Advanced use** — create a custom instance with a different limit for a specific route group: + +```ts +import { createPerUserConcurrencyMiddleware } from "./middleware/perUserConcurrency"; +router.use(createPerUserConcurrencyMiddleware({ limit: 3 })); +``` + +> **Multi-process note:** The counter is in-process only. In a clustered deployment each process enforces the limit independently, so the effective cap across N processes is `N × MAX_CONCURRENT_REQUESTS_PER_USER`. For single-process deployments (the standard Docker Compose setup) the limit is exact. + ## Indexer gap scan The gap-scan worker detects missing ledger ranges in `indexer_events` between the durable cursor and chain tip, emits `indexer_gap_detected_total{from,to}`, and self-heals via `backfillRange`: diff --git a/src/config/env-schema.ts b/src/config/env-schema.ts index 8ec26e25..2a7f5f2c 100644 --- a/src/config/env-schema.ts +++ b/src/config/env-schema.ts @@ -101,6 +101,17 @@ const baseSchema = z.object({ SLOW_QUERY_ALERTER_LIMIT: z.coerce.number().int().positive().default(10), SLOW_QUERY_ALERTER_QUERY_MAX_LENGTH: z.coerce.number().int().positive().default(1000), + // ── Per-user concurrency limiting ───────────────────────── + /** + * Maximum number of in-flight (concurrent) requests allowed for a single + * authenticated user (or IP for anonymous callers) at any point in time. + * Requests that exceed this cap receive HTTP 429 with a `Retry-After: 1` + * header. Set to a large value (e.g. 1000) to effectively disable the + * limit without removing the middleware. + * Default: 10. + */ + MAX_CONCURRENT_REQUESTS_PER_USER: z.coerce.number().int().positive().default(10), + // ── Metrics ─────────────────────────────────────────────── /** Bearer token required to access /api/metrics. Empty string (default) means no auth. */ METRICS_AUTH_TOKEN: z.string().default(""), diff --git a/src/index.ts b/src/index.ts index e6b87982..60c123a3 100644 --- a/src/index.ts +++ b/src/index.ts @@ -79,6 +79,7 @@ import { exportsRouter } from "./routes/exports"; import { fingerprintRouter } from "./routes/fingerprint"; import { alertsRouter } from "./routes/alerts"; import { gracefulShutdown } from "./lifecycle/shutdown"; +import { perUserConcurrency } from "./middleware/perUserConcurrency"; const docsEnabled = @@ -160,6 +161,12 @@ export function createApp(): express.Express { app.use("/api/indexer", indexerHealthRouter); app.use("/api/indexer/cursor", indexerCursorRouter); + // Cap in-flight concurrent requests per user/IP before any API route handler + // runs. This prevents a single identity from exhausting the thread / DB-pool + // by holding many connections open simultaneously. + // Configured via MAX_CONCURRENT_REQUESTS_PER_USER (default: 10). + app.use("/api", perUserConcurrency); + const mutationMethods = ["POST", "PATCH"] as const; app.use("/api", (req, res, next) => mutationMethods.includes(req.method as (typeof mutationMethods)[number]) diff --git a/src/middleware/perUserConcurrency.ts b/src/middleware/perUserConcurrency.ts new file mode 100644 index 00000000..dcee6f4c --- /dev/null +++ b/src/middleware/perUserConcurrency.ts @@ -0,0 +1,231 @@ +/** + * @module middleware/perUserConcurrency + * + * Per-user concurrent request limiting middleware. + * + * Motivation + * ---------- + * Rate limiting (sliding-window / token-bucket) restricts *throughput* over a + * time window but cannot prevent a single user from holding many connections + * open simultaneously (e.g. long-polling, slow uploads, streaming responses). + * This middleware caps the number of *in-flight* requests per identity so that + * a misbehaving or abusive client cannot monopolise the server's thread pool or + * database connection pool. + * + * How it works + * ------------ + * 1. On every incoming request, resolve an identity key (authenticated user ID + * or, as a fallback, the client IP address). + * 2. Atomically increment an in-flight counter stored in a module-level + * `Map`. + * 3. If the counter exceeds `MAX_CONCURRENT_REQUESTS_PER_USER`, respond + * immediately with **HTTP 429** and a `Retry-After` header. The counter is + * decremented immediately in this path so it stays accurate. + * 4. Otherwise, register a `res.on("finish")` listener that decrements the + * counter once the response is fully flushed, regardless of whether the + * request succeeded or errored. + * + * Configuration + * ------------- + * The per-user limit is read from the `MAX_CONCURRENT_REQUESTS_PER_USER` + * environment variable (zod-validated in `src/config/env-schema.ts`). + * Default: **10**. + * + * Identity resolution order + * ------------------------- + * 1. `req.user.id` — populated by `requireAuth` / `optionalAuth` + * 2. `req.user.stellarAddress` — fallback populated by the same middlewares + * 3. `X-Forwarded-For` / `req.socket.remoteAddress` — anonymous callers + * + * Error envelope + * -------------- + * ```json + * { + * "error": { + * "code": "concurrency_limit_exceeded", + * "message": "Too many concurrent requests", + * "retryAfter": 1 + * } + * } + * ``` + * + * Security notes + * -------------- + * - Decrement is guaranteed via `res.on("finish")` so a client hanging the + * connection cannot inflate the counter indefinitely — the counter is + * decremented when the socket closes even if the response is never flushed. + * We additionally register `res.on("close")` as a safety-net for + * prematurely-closed connections. + * - The counter Map is in-process only. In a multi-process deployment each + * process maintains its own counter, effectively multiplying the limit by + * the number of processes. For single-process deployments (the common case) + * this is an exact limit. Document this if deploying behind a process + * cluster. + * + * Usage + * ----- + * ```ts + * import { perUserConcurrency } from "../middleware/perUserConcurrency"; + * + * // Global: applied to all /api routes + * app.use("/api", perUserConcurrency); + * + * // Or create a custom instance for a specific route group + * import { createPerUserConcurrencyMiddleware } from "../middleware/perUserConcurrency"; + * app.use("/api/predictions", createPerUserConcurrencyMiddleware({ limit: 5 })); + * ``` + */ + +import { randomUUID } from "crypto"; +import type { NextFunction, Request, RequestHandler, Response } from "express"; +import { env } from "../config/env"; +import { logger } from "../config/logger"; + +// ── Internal helpers ───────────────────────────────────────────────────────── + +/** + * Resolves the first hop IP from `X-Forwarded-For` or falls back to the + * direct socket address. Mirrors the implementation in rateLimit.ts so + * behaviour is consistent across all limiting middleware. + */ +function getClientIp(req: Request): string { + const xff = req.headers["x-forwarded-for"]; + if (typeof xff === "string" && xff.trim().length > 0) { + const firstHop = xff.split(",")[0]?.trim(); + if (firstHop && firstHop.length > 0) { + return firstHop; + } + } + return req.socket?.remoteAddress ?? "unknown"; +} + +/** + * Returns a stable identity key for the request: + * - `user:` for authenticated requests (preferred: uses DB user ID) + * - `ip:
` for anonymous / unauthenticated requests + */ +function resolveKey(req: Request): string { + const user = (req as Request & { user?: { id?: string; stellarAddress?: string } }).user; + const identity = user?.id ?? user?.stellarAddress; + if (typeof identity === "string" && identity.trim().length > 0) { + return `user:${identity.trim()}`; + } + return `ip:${getClientIp(req)}`; +} + +// ── Public types ───────────────────────────────────────────────────────────── + +export interface PerUserConcurrencyOptions { + /** + * Maximum number of in-flight requests allowed for a single identity at + * any point in time. Defaults to `env.MAX_CONCURRENT_REQUESTS_PER_USER`. + */ + limit?: number; + + /** + * Override the default key-resolver function. The function receives the + * Express `Request` and must return a non-empty string that uniquely + * identifies the caller. + */ + keyGenerator?: (req: Request) => string; +} + +// ── Factory ────────────────────────────────────────────────────────────────── + +/** + * Creates a new per-user concurrency-limiting middleware with its own + * isolated counter Map. Multiple instances (e.g. per-route) do NOT share + * state unless you reuse the same returned middleware reference. + * + * @param options - Optional configuration overrides. + * @returns An Express `RequestHandler`. + */ +export function createPerUserConcurrencyMiddleware( + options: PerUserConcurrencyOptions = {}, +): RequestHandler { + // Resolve limit eagerly so env is read once at creation time, not per request. + const limit = Math.max(1, Math.floor(options.limit ?? env.MAX_CONCURRENT_REQUESTS_PER_USER)); + const keyGenerator = options.keyGenerator ?? resolveKey; + + /** In-flight counter per identity key. */ + const counters = new Map(); + + return function perUserConcurrencyMiddleware( + req: Request, + res: Response, + next: NextFunction, + ): void { + // Resolve correlation ID for structured logging (mirrors pattern in errorHandler.ts). + const correlationId: string = + (res.locals.correlationId as string | undefined) ?? + (req.headers["x-correlation-id"] as string | undefined) ?? + randomUUID(); + + const key = keyGenerator(req); + + // Atomically read-and-increment. + const current = (counters.get(key) ?? 0) + 1; + counters.set(key, current); + + if (current > limit) { + // Limit exceeded — decrement immediately (this request was rejected). + counters.set(key, current - 1); + + logger.warn( + { + correlationId, + key, + current: current - 1, // actual in-flight count after rejection + limit, + path: req.path, + method: req.method, + }, + "concurrency_limit_exceeded", + ); + + // Suggest the caller retry after 1 second (we have no idea when a slot + // will actually free up, but 1 is the minimum RFC-compliant value). + res.setHeader("Retry-After", "1"); + res.status(429).json({ + error: { + code: "concurrency_limit_exceeded", + message: "Too many concurrent requests", + retryAfter: 1, + }, + }); + return; + } + + // Request admitted — decrement when the response finishes (or closes early). + let decremented = false; + const decrement = (): void => { + if (decremented) return; + decremented = true; + const after = Math.max(0, (counters.get(key) ?? 1) - 1); + if (after === 0) { + counters.delete(key); + } else { + counters.set(key, after); + } + }; + + res.on("finish", decrement); + // Safety-net: if the client disconnects before the response is flushed, + // "close" fires without "finish". Decrement in both cases. + res.on("close", decrement); + + next(); + }; +} + +// ── Default singleton instance ──────────────────────────────────────────────── + +/** + * Pre-configured per-user concurrency middleware using the value of + * `MAX_CONCURRENT_REQUESTS_PER_USER` from the environment (default: 10). + * + * Mount this on the `/api` prefix (after body parsing, before route handlers) + * in `src/index.ts`. + */ +export const perUserConcurrency: RequestHandler = + createPerUserConcurrencyMiddleware(); diff --git a/tests/perUserConcurrency.test.ts b/tests/perUserConcurrency.test.ts new file mode 100644 index 00000000..94ffa679 --- /dev/null +++ b/tests/perUserConcurrency.test.ts @@ -0,0 +1,689 @@ +/** + * tests/perUserConcurrency.test.ts + * + * Unit / integration tests for `createPerUserConcurrencyMiddleware` and the + * `perUserConcurrency` singleton exported from + * `src/middleware/perUserConcurrency.ts`. + * + * Coverage targets (≥ 90 % on changed lines): + * - Requests within the limit are passed through to the next handler + * - Requests exceeding the concurrent limit receive HTTP 429 + * - 429 response contains the standardised error envelope + * - `Retry-After` header is set on 429 responses + * - Counter is decremented when the response finishes ("finish" event) + * - Counter is decremented when the slot is released (sequential reuse) + * - Per-user isolation — different users have independent counters + * - Anonymous users are keyed by IP address + * - X-Forwarded-For is respected for IP keying + * - A custom `keyGenerator` is honoured + * - A custom `limit` overrides the environment default + * - Limit clamping: negative / zero limit becomes 1 + * - Structured warning is logged when the limit is exceeded + * - The correlation ID from `res.locals` is forwarded to the logger + * - No warning is logged for allowed requests + * - The `perUserConcurrency` singleton is a valid RequestHandler + * + * NOTE ON CONCURRENCY TESTING + * ---------------------------- + * supertest's `request(app).get(...)` does NOT send until you `.then()` / + * `await` the Test object. To simulate truly concurrent in-flight requests we + * start a real `http.Server`, fire requests with `node-fetch` / `http.get`, + * and control when the server handler resolves via a shared latch mechanism. + */ + +// ── 1. Env vars (must precede project imports) ────────────────────────────── + +process.env.NODE_ENV = "test"; +process.env.PORT = "3099"; +process.env.LOG_LEVEL = "silent"; +process.env.DATABASE_URL = "postgres://localhost/test"; +process.env.JWT_SECRET = "per-user-concurrency-test-secret-at-least-32!!"; +process.env.JWT_ISSUER = "predictify"; +process.env.JWT_AUDIENCE = "predictify-app"; +process.env.SOROBAN_RPC_URL = "https://soroban-testnet.stellar.org"; +process.env.HORIZON_URL = "https://horizon-testnet.stellar.org"; +process.env.PREDICTIFY_CONTRACT_ID = + "CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF"; +process.env.MAX_CONCURRENT_REQUESTS_PER_USER = "3"; + +// ── 2. Module-level mocks ──────────────────────────────────────────────────── + +jest.mock("../src/config/logger", () => ({ + logger: { + warn: jest.fn(), + info: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + }, +})); + +// ── 3. Imports ─────────────────────────────────────────────────────────────── + +import http from "http"; +import express, { + type NextFunction, + type Request, + type RequestHandler, + type Response, +} from "express"; +import request from "supertest"; +import { + createPerUserConcurrencyMiddleware, + perUserConcurrency, + type PerUserConcurrencyOptions, +} from "../src/middleware/perUserConcurrency"; +import { logger } from "../src/config/logger"; + +const mockWarn = logger.warn as jest.Mock; + +// ── 4. Helpers ─────────────────────────────────────────────────────────────── + +/** ms → Promise void */ +const wait = (ms: number): Promise => + new Promise((resolve) => setTimeout(resolve, ms)); + +/** + * Build an Express app with the concurrency middleware + a controllable + * `/api/test` route, then spin up a real HTTP server on an ephemeral port. + * + * Returns the server and a `gate` object. Call `gate.open()` to let all + * pending handlers complete; call `gate.close()` to make them hang again. + */ +function makeServer(opts: PerUserConcurrencyOptions = {}): { + server: http.Server; + /** Resolve all currently-pending handler promises */ + openGate: () => void; + baseUrl: () => string; + close: () => Promise; +} { + let pendingResolvers: Array<() => void> = []; + let gateOpen = false; + + const app = express(); + + // Simulate requireAuth: inject a user via header. + app.use((req: Request, _res: Response, next: NextFunction) => { + const userId = req.headers["x-test-user-id"] as string | undefined; + if (userId) { + ( + req as Request & { user?: { id: string; stellarAddress: string } } + ).user = { + id: userId, + stellarAddress: `G${userId.toUpperCase()}`, + }; + } + next(); + }); + + app.use("/api", createPerUserConcurrencyMiddleware(opts)); + + app.get("/api/test", (_req, res) => { + if (gateOpen) { + res.json({ ok: true }); + return; + } + const p = new Promise((resolve) => pendingResolvers.push(resolve)); + void p.then(() => res.json({ ok: true })); + }); + + const server = http.createServer(app); + + return { + server, + openGate() { + gateOpen = true; + const local = pendingResolvers; + pendingResolvers = []; + local.forEach((r) => r()); + }, + baseUrl() { + const addr = server.address(); + if (!addr || typeof addr === "string") throw new Error("no address"); + return `http://127.0.0.1:${addr.port}`; + }, + close() { + return new Promise((resolve, reject) => { + server.close((err) => (err ? reject(err) : resolve())); + }); + }, + }; +} + +/** + * Make a GET request to the server and return status + body. + * Uses the built-in `fetch` (Node 18+) so it's truly concurrent. + */ +async function get( + url: string, + headers: Record = {}, +): Promise<{ status: number; body: unknown; headers: Record }> { + const res = await fetch(url, { headers }); + const body = (await res.json()) as unknown; + const hdrs: Record = {}; + res.headers.forEach((v, k) => { + hdrs[k] = v; + }); + return { status: res.status, body, headers: hdrs }; +} + +// ── 5. Tests ───────────────────────────────────────────────────────────────── + +describe("createPerUserConcurrencyMiddleware", () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + // ── Allowed requests ─────────────────────────────────────────────────────── + + describe("allows requests within the limit", () => { + it("passes a single request through to the next handler", async () => { + const { server, openGate, baseUrl, close } = makeServer({ limit: 3 }); + openGate(); // immediate responses + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + + const res = await get(`${baseUrl()}/api/test`, { + "x-test-user-id": "alice", + }); + + await close(); + expect(res.status).toBe(200); + expect(res.body).toEqual({ ok: true }); + }); + + it("allows exactly `limit` simultaneous in-flight requests", async () => { + const { server, openGate, baseUrl, close } = makeServer({ limit: 2 }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const url = `${baseUrl()}/api/test`; + + // Fire 2 requests (they hang at the gate). + const req1 = get(url, { "x-test-user-id": "bob" }); + const req2 = get(url, { "x-test-user-id": "bob" }); + + // Give them time to reach the middleware. + await wait(80); + + // Both should be in-flight (not yet responded). + // Open the gate to let them finish. + openGate(); + + const [r1, r2] = await Promise.all([req1, req2]); + await close(); + + expect(r1.status).toBe(200); + expect(r2.status).toBe(200); + }); + }); + + // ── 429 behaviour ───────────────────────────────────────────────────────── + + describe("rejects requests exceeding the limit with 429", () => { + it("returns 429 when a (limit+1)-th request arrives for the same user", async () => { + const { server, openGate, baseUrl, close } = makeServer({ limit: 2 }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const url = `${baseUrl()}/api/test`; + + // Two requests hold the slots. + const req1 = get(url, { "x-test-user-id": "carol" }); + const req2 = get(url, { "x-test-user-id": "carol" }); + await wait(80); + + // Third must be rejected synchronously. + const blocked = await get(url, { "x-test-user-id": "carol" }); + expect(blocked.status).toBe(429); + + openGate(); + await Promise.all([req1, req2]); + await close(); + }); + + it("returns the standard error envelope on 429", async () => { + const { server, openGate, baseUrl, close } = makeServer({ limit: 1 }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const url = `${baseUrl()}/api/test`; + + const req1 = get(url, { "x-test-user-id": "dave" }); + await wait(80); + + const blocked = await get(url, { "x-test-user-id": "dave" }); + expect(blocked.status).toBe(429); + expect(blocked.body).toEqual({ + error: { + code: "concurrency_limit_exceeded", + message: "Too many concurrent requests", + retryAfter: 1, + }, + }); + + openGate(); + await req1; + await close(); + }); + + it("includes Retry-After: 1 header on 429 responses", async () => { + const { server, openGate, baseUrl, close } = makeServer({ limit: 1 }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const url = `${baseUrl()}/api/test`; + + const req1 = get(url, { "x-test-user-id": "eve" }); + await wait(80); + + const blocked = await get(url, { "x-test-user-id": "eve" }); + expect(blocked.status).toBe(429); + expect(blocked.headers["retry-after"]).toBe("1"); + + openGate(); + await req1; + await close(); + }); + }); + + // ── Counter decrement / slot release ────────────────────────────────────── + + describe("releases the slot after the response finishes", () => { + it("allows a new request once a previous one completes", async () => { + // Use supertest for simple sequential tests (no concurrency needed). + const app = express(); + app.use(createPerUserConcurrencyMiddleware({ limit: 1 })); + app.get("/api/test", (_req, res) => res.json({ ok: true })); + + const r1 = await request(app) + .get("/api/test") + .set("x-test-user-id", "frank"); + expect(r1.status).toBe(200); + + const r2 = await request(app) + .get("/api/test") + .set("x-test-user-id", "frank"); + expect(r2.status).toBe(200); + }); + + it("counter stays accurate across many sequential requests", async () => { + const app = express(); + app.use(createPerUserConcurrencyMiddleware({ limit: 1 })); + app.get("/api/test", (_req, res) => res.json({ ok: true })); + + for (let i = 0; i < 5; i++) { + const res = await request(app) + .get("/api/test") + .set("x-test-user-id", "grace"); + expect(res.status).toBe(200); + } + }); + }); + + // ── Per-user isolation ───────────────────────────────────────────────────── + + describe("per-user isolation", () => { + it("maintains separate counters for different users", async () => { + const { server, openGate, baseUrl, close } = makeServer({ limit: 1 }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const url = `${baseUrl()}/api/test`; + + // Heidi holds the single slot (slow handler, won't complete until gate opens). + const req1 = get(url, { "x-test-user-id": "heidi" }); + await wait(80); + + // Heidi's next request is blocked immediately (her slot is taken). + const heidiBlocked = await get(url, { "x-test-user-id": "heidi" }); + expect(heidiBlocked.status).toBe(429); + + // Open the gate so heidi's first request can complete. + openGate(); + const r1 = await req1; + expect(r1.status).toBe(200); + + // After slot is released, Ivan also has a free slot. + // Use the gate-open state (default now): next request responds immediately. + const ivan = await get(url, { "x-test-user-id": "ivan" }); + expect(ivan.status).toBe(200); + + await close(); + }); + }); + + // ── Anonymous / IP keying ───────────────────────────────────────────────── + + describe("anonymous (IP-keyed) callers", () => { + it("allows sequential anonymous requests (slot freed each time)", async () => { + const app = express(); + app.use(createPerUserConcurrencyMiddleware({ limit: 2 })); + app.get("/api/test", (_req, res) => res.json({ ok: true })); + + const r1 = await request(app).get("/api/test"); + const r2 = await request(app).get("/api/test"); + expect(r1.status).toBe(200); + expect(r2.status).toBe(200); + }); + + it("blocks anonymous concurrent requests that exceed the limit", async () => { + const { server, openGate, baseUrl, close } = makeServer({ limit: 1 }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const url = `${baseUrl()}/api/test`; + + const req1 = get(url); // anonymous, takes the slot + await wait(80); + + const blocked = await get(url); // anonymous, same IP + expect(blocked.status).toBe(429); + + openGate(); + await req1; + await close(); + }); + + it("respects X-Forwarded-For: same first-hop IP is blocked", async () => { + const { server, openGate, baseUrl, close } = makeServer({ limit: 1 }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const url = `${baseUrl()}/api/test`; + + const req1 = get(url, { "x-forwarded-for": "203.0.113.1" }); + await wait(80); + + const blocked = await get(url, { "x-forwarded-for": "203.0.113.1" }); + expect(blocked.status).toBe(429); + + openGate(); + await req1; + await close(); + }); + + it("respects X-Forwarded-For: different first-hop IPs get separate slots", async () => { + const { server, openGate, baseUrl, close } = makeServer({ limit: 1 }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const url = `${baseUrl()}/api/test`; + + // IP 203.0.113.1 holds the single slot. + const req1 = get(url, { "x-forwarded-for": "203.0.113.1" }); + await wait(80); + + // Same IP — blocked immediately (synchronous 429, doesn't touch the gate). + const blocked = await get(url, { "x-forwarded-for": "203.0.113.1" }); + expect(blocked.status).toBe(429); + + // Open the gate now so req1 completes and frees its slot. + openGate(); + await req1; + + // Different IP — now makes a fresh request. Gate is open so it resolves. + const different = await get(url, { + "x-forwarded-for": "203.0.113.2", + }); + expect(different.status).toBe(200); + + await close(); + }); + }); + + // ── Custom keyGenerator ──────────────────────────────────────────────────── + + describe("custom keyGenerator", () => { + it("uses the returned key for bucketing", async () => { + // All requests map to the same key regardless of user. + const keyGenerator = jest.fn().mockReturnValue("shared-key"); + + const { server, openGate, baseUrl, close } = makeServer({ + limit: 1, + keyGenerator, + }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const url = `${baseUrl()}/api/test`; + + const req1 = get(url, { "x-test-user-id": "user-a" }); + await wait(80); + + // user-b maps to the same bucket via keyGenerator → blocked. + const blocked = await get(url, { "x-test-user-id": "user-b" }); + expect(blocked.status).toBe(429); + expect(keyGenerator).toHaveBeenCalled(); + + openGate(); + await req1; + await close(); + }); + }); + + // ── Custom limit ────────────────────────────────────────────────────────── + + describe("custom limit", () => { + it("accepts a limit of 1 (single-flight per user)", async () => { + const { server, openGate, baseUrl, close } = makeServer({ limit: 1 }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const url = `${baseUrl()}/api/test`; + + const req1 = get(url, { "x-test-user-id": "single" }); + await wait(80); + + const blocked = await get(url, { "x-test-user-id": "single" }); + expect(blocked.status).toBe(429); + + openGate(); + await req1; + await close(); + }); + + it("clamps fractional / negative limit values to at least 1", async () => { + // limit=-5 → Math.max(1, Math.floor(-5)) = 1 + const { server, openGate, baseUrl, close } = makeServer({ limit: -5 }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const url = `${baseUrl()}/api/test`; + + const req1 = get(url, { "x-test-user-id": "clamped" }); + await wait(80); + + const blocked = await get(url, { "x-test-user-id": "clamped" }); + expect(blocked.status).toBe(429); + + openGate(); + await req1; + await close(); + }); + + it("allows a high limit to pass many concurrent requests", async () => { + const app = express(); + app.use(createPerUserConcurrencyMiddleware({ limit: 50 })); + app.get("/api/test", (_req, res) => res.json({ ok: true })); + + const results = await Promise.all( + Array.from({ length: 10 }, () => + request(app).get("/api/test").set("x-test-user-id", "high-limit"), + ), + ); + for (const res of results) { + expect(res.status).toBe(200); + } + }); + }); + + // ── Structured logging ──────────────────────────────────────────────────── + + describe("structured logging", () => { + it("logs a warning when the concurrency limit is exceeded", async () => { + const { server, openGate, baseUrl, close } = makeServer({ limit: 1 }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const url = `${baseUrl()}/api/test`; + + const req1 = get(url, { "x-test-user-id": "log-test" }); + await wait(80); + + await get(url, { "x-test-user-id": "log-test" }); + + expect(mockWarn).toHaveBeenCalledWith( + expect.objectContaining({ + key: "user:log-test", + limit: 1, + // When middleware is mounted at /api, Express strips the prefix + // from req.path so the middleware sees "/test", not "/api/test". + path: "/test", + method: "GET", + }), + "concurrency_limit_exceeded", + ); + + openGate(); + await req1; + await close(); + }); + + it("does not log a warning when requests are within the limit", async () => { + const app = express(); + app.use(createPerUserConcurrencyMiddleware({ limit: 5 })); + app.get("/api/test", (_req, res) => res.json({ ok: true })); + + await request(app).get("/api/test").set("x-test-user-id", "no-warn"); + + expect(mockWarn).not.toHaveBeenCalledWith( + expect.anything(), + "concurrency_limit_exceeded", + ); + }); + + it("forwards the correlation ID from res.locals to the log entry", async () => { + // Build a custom app that injects a known correlationId via a middleware + // placed before the concurrency middleware. + const withCorrelation: RequestHandler = (_req, res, next) => { + res.locals.correlationId = "test-corr-123"; + next(); + }; + + const app = express(); + app.use((req: Request, _res: Response, next: NextFunction) => { + const userId = req.headers["x-test-user-id"] as string | undefined; + if (userId) { + ( + req as Request & { user?: { id: string; stellarAddress: string } } + ).user = { + id: userId, + stellarAddress: `G${userId.toUpperCase()}`, + }; + } + next(); + }); + app.use("/api", withCorrelation); + app.use("/api", createPerUserConcurrencyMiddleware({ limit: 1 })); + + let pendingResolver: (() => void) | null = null; + app.get("/api/test", (_req, res) => { + const p = new Promise((r) => { + pendingResolver = r; + }); + void p.then(() => res.json({ ok: true })); + }); + + const server = http.createServer(app); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const addr = server.address() as { port: number }; + const url = `http://127.0.0.1:${addr.port}/api/test`; + + const req1 = get(url, { "x-test-user-id": "corr-user" }); + await wait(80); + + await get(url, { "x-test-user-id": "corr-user" }); // blocked + + expect(mockWarn).toHaveBeenCalledWith( + expect.objectContaining({ correlationId: "test-corr-123" }), + "concurrency_limit_exceeded", + ); + + // Clean up. + pendingResolver?.(); + await req1; + await new Promise((r, rej) => + server.close((e) => (e ? rej(e) : r())), + ); + }); + }); + + // ── Instance isolation ──────────────────────────────────────────────────── + + describe("instance isolation", () => { + it("two independent middleware instances have separate counters", async () => { + // Two apps, each with their own middleware instance. + const mw1 = createPerUserConcurrencyMiddleware({ limit: 1 }); + const mw2 = createPerUserConcurrencyMiddleware({ limit: 1 }); + + let resolver1: (() => void) | null = null; + + const app1 = express(); + app1.use((req: Request, _res: Response, next: NextFunction) => { + const userId = req.headers["x-test-user-id"] as string | undefined; + if (userId) { + ( + req as Request & { user?: { id: string; stellarAddress: string } } + ).user = { id: userId, stellarAddress: `G${userId.toUpperCase()}` }; + } + next(); + }); + app1.use("/api", mw1); + app1.get("/api/test", (_req, res) => { + const p = new Promise((r) => { + resolver1 = r; + }); + void p.then(() => res.json({ ok: true })); + }); + + const app2 = express(); + app2.use((req: Request, _res: Response, next: NextFunction) => { + const userId = req.headers["x-test-user-id"] as string | undefined; + if (userId) { + ( + req as Request & { user?: { id: string; stellarAddress: string } } + ).user = { id: userId, stellarAddress: `G${userId.toUpperCase()}` }; + } + next(); + }); + app2.use("/api", mw2); + app2.get("/api/test", (_req, res) => res.json({ ok: true })); + + const s1 = http.createServer(app1); + const s2 = http.createServer(app2); + await Promise.all([ + new Promise((r) => s1.listen(0, "127.0.0.1", r)), + new Promise((r) => s2.listen(0, "127.0.0.1", r)), + ]); + + const p1 = s1.address() as { port: number }; + const p2 = s2.address() as { port: number }; + + // Hold slot on app1. + const r1 = get(`http://127.0.0.1:${p1.port}/api/test`, { + "x-test-user-id": "iso", + }); + await wait(80); + + // app2's counter is independent — it should succeed. + const r2 = await get(`http://127.0.0.1:${p2.port}/api/test`, { + "x-test-user-id": "iso", + }); + expect(r2.status).toBe(200); + + resolver1?.(); + await r1; + + await Promise.all([ + new Promise((r, rej) => s1.close((e) => (e ? rej(e) : r()))), + new Promise((r, rej) => s2.close((e) => (e ? rej(e) : r()))), + ]); + }); + }); +}); + +// ── Singleton export ───────────────────────────────────────────────────────── + +describe("perUserConcurrency singleton", () => { + it("is a RequestHandler function", () => { + expect(typeof perUserConcurrency).toBe("function"); + }); + + it("allows requests within the default limit", async () => { + const app = express(); + app.use("/api", perUserConcurrency); + app.get("/api/test", (_req, res) => res.json({ ok: true })); + + // MAX_CONCURRENT_REQUESTS_PER_USER=3 at top of file; a single request + // should always succeed. + const res = await request(app) + .get("/api/test") + .set("x-test-user-id", "singleton-user"); + expect(res.status).toBe(200); + }); +});