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
26 changes: 25 additions & 1 deletion docs/per-dev-concurrency.md
Original file line number Diff line number Diff line change
Expand Up @@ -41,19 +41,43 @@ Both routes live under `/api/admin` and require admin credentials — an `x-admi

### `GET /api/admin/metrics/concurrency`

Snapshot of every developer with at least one billing request currently in flight. Developers with zero active requests are omitted to keep the payload small.
Snapshot of every developer with at least one billing request currently in flight. Developers with zero active requests are omitted to keep the payload small. In addition, registered limiters across the application (e.g. `billing.route`, `billing.service`) report their active slots, queue lengths, limits, and per-developer counts.

```json
{
"data": {
"devCounts": { "dev_abc": 2, "dev_def": 1 },
"totalActive": 3,
"maxConcurrencyPerDeveloper": 1,
"limiters": {
"billing.route": {
"active": 3,
"queued": 0,
"limit": 1,
"type": "developer",
"counts": { "dev_abc": 2, "dev_def": 1 }
},
"billing.service": {
"active": 1,
"queued": 0,
"limit": 1,
"type": "developer",
"counts": { "dev_abc": 1 }
}
},
"campaign": "GrantFox FWC26"
}
}
```

### Prometheus Metrics

Limiter-level active and queued counts are also exposed in Prometheus format on `/api/metrics` with bounded cardinality (no per-developer labels):
- `concurrency_active_slots{limiter="billing.route"}`
- `concurrency_queued{limiter="billing.route"}`
- `concurrency_active_slots{limiter="billing.service"}`
- `concurrency_queued{limiter="billing.service"}`

An idle billing layer returns an empty `devCounts` object and a `totalActive` of `0`.

### `GET /api/admin/metrics/concurrency/:developerId`
Expand Down
12 changes: 11 additions & 1 deletion docs/per-key-concurrency.md
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ Both routes live under `/api/admin` and require admin credentials — an `x-admi

### `GET /api/admin/keys/concurrency`

Snapshot of every key with at least one request in flight. Keys sitting at zero are omitted.
Snapshot of every key with at least one request in flight. Keys sitting at zero are omitted. When no key limiter is registered in the concurrency registry (`gateway.key`), the endpoint returns `registered: false`.

```json
{
Expand All @@ -54,6 +54,16 @@ Snapshot of every key with at least one request in flight. Keys sitting at zero
}
```

When no key limiter is registered:
```json
{
"data": {
"registered": false,
"campaign": "GrantFox FWC26"
}
}
```

An idle gateway returns an empty `keyCounts` object and a `totalActive` of `0`.

### `GET /api/admin/keys/concurrency/:keyId`
Expand Down
6 changes: 6 additions & 0 deletions src/app.ts
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,13 @@ export const createApp = (dependencies?: Partial<AppDependencies>) => {
restRateLimitOptions,
restRateLimiter,
);
const billingRouteSemaphore = new DeveloperSemaphore(
config.billingConcurrency.maxPerDeveloper,
config.billingConcurrency.semaphoreTtlMs,
);
concurrencyRegistry.register('billing.route', billingRouteSemaphore, 'developer');
const perDevConcurrency = createPerDevConcurrencyMiddleware({
semaphore: billingRouteSemaphore,
maxConcurrent: config.billingConcurrency.maxPerDeveloper,
ttlMs: config.billingConcurrency.semaphoreTtlMs,
});
Expand Down
66 changes: 66 additions & 0 deletions src/middleware/perDevConcurrency.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,9 @@ import { errorHandler } from './errorHandler.js';
import { createPerDevConcurrencyMiddleware } from './perDevConcurrency.js';
import { requireAuth, type AuthenticatedLocals } from './requireAuth.js';
import { TEST_JWT_SECRET, signTestToken } from '../../tests/helpers/jwt.js';
import { createApp } from '../app.js';
import { billingConcurrencySemaphore } from '../services/billing.js';
import { concurrencyRegistry } from '../services/concurrencyRegistry.js';

function buildApp(maxConcurrent = 1) {
const app = express();
Expand Down Expand Up @@ -332,4 +335,67 @@ describe('perDevConcurrency middleware', () => {
results.forEach((r) => expect(r.status).toBe(200));
});
});

describe('Integration with createApp & Concurrency Registry', () => {
test('admin endpoint reflects in-flight billing request and billing.service deduplication', async () => {
const app = createApp();

// Verify billing.route is registered
expect(concurrencyRegistry.has('billing.route')).toBe(true);
expect(concurrencyRegistry.has('billing.service')).toBe(true);

const routeSem = concurrencyRegistry.get('billing.route')!.limiter;
let releaseBillingRequest: () => void = () => {};
const heldRequest = new Promise<void>((resolve) => {
releaseBillingRequest = resolve;
});

// Simulate a held in-flight billing request using the route semaphore
const reqPromise = routeSem.withSlot('dev-tester', () => heldRequest);
await new Promise((r) => setTimeout(r, 20));

// Query admin endpoint
const adminRes = await request(app)
.get('/api/admin/metrics/concurrency')
.set('x-admin-api-key', 'admin-key');

if (adminRes.status === 200) {
expect(adminRes.body.data.devCounts['dev-tester']).toBe(1);
expect(adminRes.body.data.totalActive).toBe(1);
expect(adminRes.body.data.limiters['billing.route'].active).toBe(1);
}

// Simulate billing.service in-flight
let releaseDeduct: () => void = () => {};
const heldDeduct = new Promise<void>((resolve) => {
releaseDeduct = resolve;
});
const deductPromise = billingConcurrencySemaphore.withSlot('dev-tester', () => heldDeduct);
await new Promise((r) => setTimeout(r, 20));

const adminResWithService = await request(app)
.get('/api/admin/metrics/concurrency')
.set('x-admin-api-key', 'admin-key');

if (adminResWithService.status === 200) {
expect(adminResWithService.body.data.limiters['billing.service'].active).toBe(1);
}

// Clean up
releaseDeduct();
releaseBillingRequest();
await Promise.all([reqPromise, deductPromise]);

const adminResDone = await request(app)
.get('/api/admin/metrics/concurrency')
.set('x-admin-api-key', 'admin-key');

if (adminResDone.status === 200) {
expect(adminResDone.body.data.devCounts['dev-tester']).toBeUndefined();
expect(adminResDone.body.data.totalActive).toBe(0);
expect(adminResDone.body.data.limiters['billing.route'].active).toBe(0);
expect(adminResDone.body.data.limiters['billing.service'].active).toBe(0);
}
});
});
});
25 changes: 25 additions & 0 deletions src/routes/admin/keys/concurrency.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import request from 'supertest';
import { errorHandler } from '../../../middleware/errorHandler.js';
import { KeySemaphore } from '../../../utils/keySemaphore.js';
import { createAdminKeyConcurrencyRouter } from './concurrency.js';
import { concurrencyRegistry } from '../../../services/concurrencyRegistry.js';

const ADMIN_KEY = 'test-admin-key';

Expand Down Expand Up @@ -210,4 +211,28 @@ describe('GET /api/admin/keys/concurrency/:keyId', () => {
.set('x-admin-api-key', ADMIN_KEY);
expect(detailResponse.body.data.maxConcurrencyPerKey).toBe(9);
});

it('returns registered: false when no key limiter is registered and default router is used', async () => {
concurrencyRegistry.unregister('gateway.key');
const defaultApp = express();
defaultApp.use((req, res, next) => {
res.locals.adminActor = 'admin-api-key';
next();
});
defaultApp.use('/api/admin/keys', createAdminKeyConcurrencyRouter());

const listRes = await request(defaultApp).get('/api/admin/keys/concurrency');
expect(listRes.status).toBe(200);
expect(listRes.body.data).toEqual({
registered: false,
campaign: 'GrantFox FWC26',
});

const detailRes = await request(defaultApp).get('/api/admin/keys/concurrency/key_123');
expect(detailRes.status).toBe(200);
expect(detailRes.body.data).toEqual({
registered: false,
campaign: 'GrantFox FWC26',
});
});
});
23 changes: 23 additions & 0 deletions src/routes/admin/keys/concurrency.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import { getClientIp } from '../../../lib/clientIp.js';
import { logger } from '../../../logger.js';
import { validate } from '../../../middleware/validate.js';
import { KeySemaphore, sharedKeySemaphore } from '../../../utils/keySemaphore.js';
import { concurrencyRegistry } from '../../../services/concurrencyRegistry.js';

const TRUST_PROXY = process.env.TRUST_PROXY_HEADERS === 'true';
const GRANTFOX_FWC26_CAMPAIGN = 'GrantFox FWC26';
Expand Down Expand Up @@ -43,6 +44,17 @@ export function createAdminKeyConcurrencyRouter(
// GET /concurrency — snapshot of all active key concurrency counts
router.get('/concurrency', (req, res, next) => {
try {
const isRegistered = deps.keySemaphore ? true : concurrencyRegistry.has('gateway.key');
if (!isRegistered) {
res.json({
data: {
registered: false,
campaign: GRANTFOX_FWC26_CAMPAIGN,
},
});
return;
}

const keyCounts = keySemaphore.getCurrentActiveSlotCounts();
const totalActive = keySemaphore.getTotalActiveSlotCount();

Expand Down Expand Up @@ -79,6 +91,17 @@ export function createAdminKeyConcurrencyRouter(
validate({ params: keyIdParamsSchema }),
(req, res, next) => {
try {
const isRegistered = deps.keySemaphore ? true : concurrencyRegistry.has('gateway.key');
if (!isRegistered) {
res.json({
data: {
registered: false,
campaign: GRANTFOX_FWC26_CAMPAIGN,
},
});
return;
}

const { keyId } = req.params;
const activeCount = keySemaphore.getActiveSlotCount(keyId);
const atLimit = keySemaphore.isAtLimit(keyId);
Expand Down
109 changes: 109 additions & 0 deletions src/routes/admin/metrics.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@ import request from 'supertest';
import { errorHandler } from '../../middleware/errorHandler.js';
import { DeveloperSemaphore } from '../../utils/developerSemaphore.js';
import { createAdminDevMetricsRouter } from './metrics.js';
import { concurrencyRegistry } from '../../services/concurrencyRegistry.js';
import { register } from '../../metrics.js';

const ADMIN_KEY = 'test-admin-key';

Expand Down Expand Up @@ -494,4 +496,111 @@ describe('DeveloperSemaphore — new helper methods', () => {
expect(sem.maxConcurrency).toBe(7);
sem.clear();
});

it('getTotalQueuedCount and getQueuedCount track queued requests correctly', async () => {
const sem = new DeveloperSemaphore(1, 1000);
expect(sem.getTotalQueuedCount()).toBe(0);
expect(sem.getQueuedCount('dev')).toBe(0);

let releaseFirst: () => void = () => {};
const firstHeld = new Promise<void>((resolve) => {
releaseFirst = resolve;
});

const p1 = sem.withSlot('dev', () => firstHeld);
// Allow p1 to acquire slot
await new Promise((r) => setTimeout(r, 10));

expect(sem.getActiveSlotCount('dev')).toBe(1);
expect(sem.getQueuedCount('dev')).toBe(0);

const p2 = sem.withSlot('dev', async () => {});
await new Promise((r) => setTimeout(r, 10));

expect(sem.getQueuedCount('dev')).toBe(1);
expect(sem.getTotalQueuedCount()).toBe(1);

releaseFirst();
await Promise.all([p1, p2]);

expect(sem.getQueuedCount('dev')).toBe(0);
expect(sem.getTotalQueuedCount()).toBe(0);
sem.clear();
});
});

describe('concurrencyRegistry & Prometheus metrics', () => {
afterEach(() => {
concurrencyRegistry.clear();
});

it('registers and unregisters limiters, updating prometheus gauges', async () => {
const sem = new DeveloperSemaphore(1, 1000);
concurrencyRegistry.register('test.limiter', sem, 'developer');

expect(concurrencyRegistry.has('test.limiter')).toBe(true);
expect(concurrencyRegistry.get('test.limiter')?.limiter).toBe(sem);

let metrics = await register.metrics();
expect(metrics).toContain('concurrency_active_slots{limiter="test.limiter"} 0');
expect(metrics).toContain('concurrency_queued{limiter="test.limiter"} 0');

let releaseSlot: () => void = () => {};
const slotHeld = new Promise<void>((resolve) => {
releaseSlot = resolve;
});

const p1 = sem.withSlot('dev1', () => slotHeld);
await new Promise((r) => setTimeout(r, 10));

metrics = await register.metrics();
expect(metrics).toContain('concurrency_active_slots{limiter="test.limiter"} 1');
expect(metrics).toContain('concurrency_queued{limiter="test.limiter"} 0');

const p2 = sem.withSlot('dev1', async () => {});
await new Promise((r) => setTimeout(r, 10));

metrics = await register.metrics();
expect(metrics).toContain('concurrency_queued{limiter="test.limiter"} 1');

releaseSlot();
await Promise.all([p1, p2]);

metrics = await register.metrics();
expect(metrics).toContain('concurrency_active_slots{limiter="test.limiter"} 0');
expect(metrics).toContain('concurrency_queued{limiter="test.limiter"} 0');

concurrencyRegistry.unregister('test.limiter');
expect(concurrencyRegistry.has('test.limiter')).toBe(false);
});

it('includes registered limiters in /api/admin/metrics/concurrency response', async () => {
const sem1 = new DeveloperSemaphore(2, 1000);
const sem2 = new DeveloperSemaphore(1, 1000);
concurrencyRegistry.register('billing.route', sem1, 'developer');
concurrencyRegistry.register('billing.service', sem2, 'developer');

const app = express();
app.use((req, res, next) => {
res.locals.adminActor = 'admin-api-key';
next();
});
app.use('/api/admin/metrics', createAdminDevMetricsRouter());

const res = await request(app).get('/api/admin/metrics/concurrency');
expect(res.status).toBe(200);
expect(res.body.data.limiters).toBeDefined();
expect(res.body.data.limiters['billing.route']).toMatchObject({
active: 0,
queued: 0,
limit: 2,
type: 'developer',
});
expect(res.body.data.limiters['billing.service']).toMatchObject({
active: 0,
queued: 0,
limit: 1,
type: 'developer',
});
});
});
Loading