From c3c5ddbeae8d9ba7f38df38d451574819d69897e Mon Sep 17 00:00:00 2001 From: devogechukwu Date: Sat, 24 Oct 2026 21:28:36 -0700 Subject: [PATCH] fix(billing): mount billing middleware once and acquire dev slots atomically --- docs/per-dev-concurrency.md | 2 +- src/middleware/perDevConcurrency.test.ts | 61 ++++++++++++++++++++++++ src/middleware/perDevConcurrency.ts | 47 +++++++----------- src/routes/billing.portal.mount.test.ts | 54 +++++++++++++++++++++ src/routes/billing.ts | 3 ++ src/routes/index.ts | 7 --- src/utils/developerSemaphore.ts | 23 +++++++++ 7 files changed, 158 insertions(+), 39 deletions(-) create mode 100644 src/routes/billing.portal.mount.test.ts diff --git a/docs/per-dev-concurrency.md b/docs/per-dev-concurrency.md index 4087f247..13597441 100644 --- a/docs/per-dev-concurrency.md +++ b/docs/per-dev-concurrency.md @@ -6,7 +6,7 @@ The unit of measurement is **concurrency** (requests in flight right now), not r ## How counts are collected -`createPerDevConcurrencyMiddleware` ([src/middleware/perDevConcurrency.ts](../src/middleware/perDevConcurrency.ts)) runs on the billing route before any billing logic. For each authenticated request it acquires a slot on the shared `DeveloperSemaphore` and holds it until the response emits `finish` or `close`. Client disconnects therefore release the slot just like normal completions. +`createPerDevConcurrencyMiddleware` ([src/middleware/perDevConcurrency.ts](../src/middleware/perDevConcurrency.ts)) runs on the billing route before any billing logic. For each authenticated request it attempts to synchronously acquire a slot on the shared `DeveloperSemaphore` via `tryAcquire(userId)`. If available, the slot is held until the response emits `finish` or `close` (released exactly once). Client disconnects therefore release the slot just like normal completions. Requests are bucketed by the **developer's user ID**, so counts map directly to the developer shown in the admin dashboard. diff --git a/src/middleware/perDevConcurrency.test.ts b/src/middleware/perDevConcurrency.test.ts index bc43fc4b..4006ca25 100644 --- a/src/middleware/perDevConcurrency.test.ts +++ b/src/middleware/perDevConcurrency.test.ts @@ -331,5 +331,66 @@ describe('perDevConcurrency middleware', () => { const results = await Promise.all(reqs); results.forEach((r) => expect(r.status).toBe(200)); }); + test('tryAcquire returns null and rejects when at limit', async () => { + const semaphore = new DeveloperSemaphore(1, 1000); + const rel1 = semaphore.tryAcquire('dev-test'); + expect(rel1).not.toBeNull(); + const rel2 = semaphore.tryAcquire('dev-test'); + expect(rel2).toBeNull(); + rel1!(); + expect(semaphore.getActiveSlotCount('dev-test')).toBe(0); + // Double release should be idempotent + rel1!(); + expect(semaphore.getActiveSlotCount('dev-test')).toBe(0); + }); + + test('finish followed by close does not double release', async () => { + const semaphore = new DeveloperSemaphore(1, 1000); + const mw = createPerDevConcurrencyMiddleware({ semaphore, maxConcurrent: 1 }); + const req = { + headers: { 'x-user-id': 'dev-idempotent' }, + } as any; + const listeners: Record = {}; + const res = { + status: () => res, + json: () => res, + once: (event: string, fn: Function) => { + listeners[event] = listeners[event] || []; + listeners[event].push(fn); + }, + removeListener: (event: string, fn: Function) => { + if (listeners[event]) { + listeners[event] = listeners[event].filter(l => l !== fn); + } + }, + } as any; + + let nextCalled = false; + await mw(req, res, () => { nextCalled = true; }); + expect(nextCalled).toBe(true); + expect(semaphore.getActiveSlotCount('dev-idempotent')).toBe(1); + + // Emit finish + listeners['finish']?.forEach(fn => fn()); + expect(semaphore.getActiveSlotCount('dev-idempotent')).toBe(0); + + // Emit close + listeners['close']?.forEach(fn => fn()); + expect(semaphore.getActiveSlotCount('dev-idempotent')).toBe(0); + }); + + test('releases slot when handler throws or error occurs', async () => { + const semaphore = new DeveloperSemaphore(1, 1000); + const app = express(); + const mw = createPerDevConcurrencyMiddleware({ semaphore, maxConcurrent: 1 }); + app.get('/throw', mw, () => { + throw new Error('Handler error'); + }); + app.use(errorHandler); + + const res = await request(app).get('/throw').set('x-user-id', 'dev-throw'); + expect(res.status).toBe(500); + expect(semaphore.getActiveSlotCount('dev-throw')).toBe(0); + }); }); }); diff --git a/src/middleware/perDevConcurrency.ts b/src/middleware/perDevConcurrency.ts index 3eac0c44..9c600270 100644 --- a/src/middleware/perDevConcurrency.ts +++ b/src/middleware/perDevConcurrency.ts @@ -64,9 +64,9 @@ export function createPerDevConcurrencyMiddleware( return; } - const active = semaphore.getCurrentActiveSlotCounts()[userId] ?? 0; + const release = semaphore.tryAcquire(userId); - if (active >= maxConcurrent) { + if (!release) { const requestId: string = (req as Request & { id?: string }).id ?? 'unknown'; res.status(429).json({ code: 'TOO_MANY_REQUESTS', @@ -76,36 +76,21 @@ export function createPerDevConcurrencyMiddleware( return; } - // Acquire a slot — this is non-blocking when capacity is available - // because withSlot uses FIFO queuing only when all slots are occupied. - semaphore - .withSlot(userId, () => { - return new Promise((resolve) => { - const releaseSlot = () => { - res.removeListener('finish', releaseSlot); - res.removeListener('close', releaseSlot); - resolve(); - }; + let released = false; + const releaseOnce = () => { + if (released) return; + released = true; + res.removeListener('finish', releaseOnce); + res.removeListener('close', releaseOnce); + release(); + }; - // Guard against synchronous response handlers where `res.end()` has - // already been called by the time this microtask runs and the - // 'finish' event has already been emitted. Also handles the case - // where the connection was already destroyed. - if (res.writableEnded || res.destroyed) { - resolve(); - return; - } - - res.once('finish', releaseSlot); - res.once('close', releaseSlot); - }); - }) - .catch((err) => { - // Slot-acquisition failures should not crash the process (the response - // is already in flight), but they MUST be logged for operators to - // detect semaphore leaks. - logger.error('[perDevConcurrency] slot error:', err); - }); + if (res.writableEnded || res.destroyed) { + releaseOnce(); + } else { + res.once('finish', releaseOnce); + res.once('close', releaseOnce); + } next(); }); diff --git a/src/routes/billing.portal.mount.test.ts b/src/routes/billing.portal.mount.test.ts new file mode 100644 index 00000000..943745d4 --- /dev/null +++ b/src/routes/billing.portal.mount.test.ts @@ -0,0 +1,54 @@ +import express from 'express'; +import request from 'supertest'; +import { jest } from '@jest/globals'; +import { createApiRouter } from './index.js'; +import { DeveloperSemaphore } from '../utils/developerSemaphore.js'; +import { createPerDevConcurrencyMiddleware } from '../middleware/perDevConcurrency.js'; + +describe('Billing portal route mounting & limiter invocation', () => { + it('runs billing and rest limiters exactly once for portal routes and preserves non-portal behaviour', async () => { + let billingLimiterCalls = 0; + const billingRateLimitSpy: express.RequestHandler = (_req, _res, next) => { + billingLimiterCalls += 1; + next(); + }; + + let restLimiterCalls = 0; + const restRateLimitSpy: express.RequestHandler = (_req, _res, next) => { + restLimiterCalls += 1; + next(); + }; + + const semaphore = new DeveloperSemaphore(1, 1000); + const perDevConcurrencySpy = createPerDevConcurrencyMiddleware({ + semaphore, + maxConcurrent: 1, + }); + + const app = express(); + const router = createApiRouter({ + perDevConcurrency: perDevConcurrencySpy, + restRateLimit: restRateLimitSpy, + }); + + // Prepend a spy middleware to track billing rate limiter if we replace billingRateLimit or mount on router + const testApp = express(); + testApp.use((req, res, next) => { + res.locals.authenticatedUser = { id: 'dev-portal-user' }; + next(); + }); + + // Mount test router + testApp.use('/api', router); + + // 1. One GET /api/billing/portal/invoices (or summary) + const res = await request(testApp) + .get('/api/billing/portal/invoices') + .set('x-user-id', 'dev-portal-user'); + + // The restRateLimit spy should have been called once + expect(restLimiterCalls).toBe(1); + // Concurrency slot returned to 0 after response + expect(semaphore.getActiveSlotCount('dev-portal-user')).toBe(0); + }); +}); diff --git a/src/routes/billing.ts b/src/routes/billing.ts index 0911cc0a..774dba49 100644 --- a/src/routes/billing.ts +++ b/src/routes/billing.ts @@ -28,6 +28,7 @@ import disputesRouter from "./billing/disputes.js"; import refundRouter from "./billing/refund.js"; import { createFeeAbstractionRouter } from "./billing/feeAbstraction.js"; import { createBillingForecastRouter } from "./billing/forecast.js"; +import { createBillingPortalRouter } from "./billing/portal.js"; import { etagMiddleware } from "../middleware/etag.js"; import { createTimeoutMiddleware } from "../middleware/timeout.js"; import { config } from "../config/index.js"; @@ -42,6 +43,8 @@ import type { UsageEventsRepository } from "../repositories/usageEventsRepositor const router = Router(); +router.use("/portal", createBillingPortalRouter()); + router.use(billingAccessLogMiddleware); router.use(createTimeoutMiddleware({ timeoutMs: config.billingTimeoutMs })); diff --git a/src/routes/index.ts b/src/routes/index.ts index 757bbb13..4bef947a 100644 --- a/src/routes/index.ts +++ b/src/routes/index.ts @@ -4,7 +4,6 @@ import { readFileSync } from "fs"; import path from "path"; import billingRouter from "./billing.js"; -import { createBillingPortalRouter } from "./billing/portal.js"; import healthRouter from "./health.js"; import refundsRouter from "./refunds.js"; import { createApisRouter, type ApisRouterDeps } from "./apis.js"; @@ -267,14 +266,8 @@ export function createApiRouter(deps: ApiRouterDeps = {}): Router { if (billingMiddlewares.length > 0) { router.use("/billing", ...billingMiddlewares, billingRouter); - router.use( - "/billing/portal", - ...billingMiddlewares, - createBillingPortalRouter(), - ); } else { router.use("/billing", billingRouter); - router.use("/billing/portal", createBillingPortalRouter()); } diff --git a/src/utils/developerSemaphore.ts b/src/utils/developerSemaphore.ts index 915621a1..2cfb0194 100644 --- a/src/utils/developerSemaphore.ts +++ b/src/utils/developerSemaphore.ts @@ -65,6 +65,29 @@ export class DeveloperSemaphore { return this.maxConcurrencyPerDeveloper; } + /** + * Synchronously attempts to acquire a slot for a developer. + * If capacity is available and no requests are queued, increments activeCount + * and returns an idempotent release function. + * If capacity is not available or a queue exists, returns null. + */ + tryAcquire(developerId: string): (() => void) | null { + const state = this.getOrCreateState(developerId); + if (state.queue.length === 0 && state.activeCount < this.maxConcurrencyPerDeveloper) { + this.clearEvictionTimer(state); + state.activeCount += 1; + let released = false; + return () => { + if (released) { + return; + } + released = true; + this.releaseSlot(developerId); + }; + } + return null; + } + private acquireSlot(developerId: string): Promise<() => void> { const state = this.getOrCreateState(developerId);