Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/per-dev-concurrency.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down
61 changes: 61 additions & 0 deletions src/middleware/perDevConcurrency.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<string, Function[]> = {};
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);
});
});
});
47 changes: 16 additions & 31 deletions src/middleware/perDevConcurrency.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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',
Expand All @@ -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<void>((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();
});
Expand Down
54 changes: 54 additions & 0 deletions src/routes/billing.portal.mount.test.ts
Original file line number Diff line number Diff line change
@@ -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);
});
});
3 changes: 3 additions & 0 deletions src/routes/billing.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -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 }));

Expand Down
7 changes: 0 additions & 7 deletions src/routes/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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());
}


Expand Down
23 changes: 23 additions & 0 deletions src/utils/developerSemaphore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand Down