diff --git a/docs/per-dev-concurrency.md b/docs/per-dev-concurrency.md index 4087f247..d6597a81 100644 --- a/docs/per-dev-concurrency.md +++ b/docs/per-dev-concurrency.md @@ -41,7 +41,7 @@ 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 { @@ -49,11 +49,35 @@ Snapshot of every developer with at least one billing request currently in fligh "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` diff --git a/docs/per-key-concurrency.md b/docs/per-key-concurrency.md index bbed8bbd..599bef0e 100644 --- a/docs/per-key-concurrency.md +++ b/docs/per-key-concurrency.md @@ -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 { @@ -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` diff --git a/src/app.ts b/src/app.ts index 224348fa..6637fe57 100644 --- a/src/app.ts +++ b/src/app.ts @@ -91,7 +91,13 @@ export const createApp = (dependencies?: Partial) => { 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, }); diff --git a/src/middleware/perDevConcurrency.test.ts b/src/middleware/perDevConcurrency.test.ts index bc43fc4b..d2cc017f 100644 --- a/src/middleware/perDevConcurrency.test.ts +++ b/src/middleware/perDevConcurrency.test.ts @@ -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(); @@ -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((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((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); + } + }); + }); }); diff --git a/src/routes/admin/keys/concurrency.test.ts b/src/routes/admin/keys/concurrency.test.ts index f8862c41..cf086b78 100644 --- a/src/routes/admin/keys/concurrency.test.ts +++ b/src/routes/admin/keys/concurrency.test.ts @@ -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'; @@ -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', + }); + }); }); diff --git a/src/routes/admin/keys/concurrency.ts b/src/routes/admin/keys/concurrency.ts index 33633dd4..ff935fc4 100644 --- a/src/routes/admin/keys/concurrency.ts +++ b/src/routes/admin/keys/concurrency.ts @@ -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'; @@ -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(); @@ -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); diff --git a/src/routes/admin/metrics.test.ts b/src/routes/admin/metrics.test.ts index d4bb367b..35ad5257 100644 --- a/src/routes/admin/metrics.test.ts +++ b/src/routes/admin/metrics.test.ts @@ -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'; @@ -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((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((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', + }); + }); }); diff --git a/src/routes/admin/metrics.ts b/src/routes/admin/metrics.ts index 061547e3..29259313 100644 --- a/src/routes/admin/metrics.ts +++ b/src/routes/admin/metrics.ts @@ -7,6 +7,7 @@ import { logger } from '../../logger.js'; import { validate } from '../../middleware/validate.js'; import { DeveloperSemaphore, sharedDeveloperSemaphore } from '../../utils/developerSemaphore.js'; import { computeConcurrencyStats } from '../../services/concurrency.js'; +import { concurrencyRegistry } from '../../services/concurrencyRegistry.js'; const TRUST_PROXY = process.env.TRUST_PROXY_HEADERS === 'true'; const GRANTFOX_FWC26_CAMPAIGN = 'GrantFox FWC26'; @@ -47,7 +48,16 @@ export function createAdminDevMetricsRouter( deps: Partial = {}, ): Router { const router = Router(); - const developerSemaphore = deps.developerSemaphore ?? sharedDeveloperSemaphore; + const getBillingRouteSemaphore = (): DeveloperSemaphore => { + if (deps.developerSemaphore) { + return deps.developerSemaphore; + } + const registered = concurrencyRegistry.get('billing.route'); + if (registered && registered.limiter instanceof DeveloperSemaphore) { + return registered.limiter; + } + return sharedDeveloperSemaphore; + }; /** * GET / @@ -74,6 +84,7 @@ export function createAdminDevMetricsRouter( */ router.get('/', (req, res, next) => { try { + const developerSemaphore = getBillingRouteSemaphore(); const stats = computeConcurrencyStats(developerSemaphore); logger.audit('READ_DEV_CONCURRENCY_STATS', res.locals.adminActor, { @@ -117,10 +128,40 @@ export function createAdminDevMetricsRouter( */ router.get('/concurrency', (req, res, next) => { try { + const developerSemaphore = getBillingRouteSemaphore(); const devCounts = developerSemaphore.getCurrentActiveSlotCounts(); const totalActive = developerSemaphore.getTotalActiveSlotCount(); const maxConcurrencyPerDeveloper = developerSemaphore.maxConcurrency; + const registeredLimiters = concurrencyRegistry.getAll(); + const limiters: Record; + }> = {}; + + for (const item of registeredLimiters) { + limiters[item.name] = { + active: item.limiter.getTotalActiveSlotCount(), + queued: item.limiter.getTotalQueuedCount(), + limit: item.limiter.maxConcurrency, + type: item.type, + counts: item.limiter.getCurrentActiveSlotCounts(), + }; + } + + if (!limiters['billing.route'] && developerSemaphore) { + limiters['billing.route'] = { + active: developerSemaphore.getTotalActiveSlotCount(), + queued: developerSemaphore.getTotalQueuedCount(), + limit: developerSemaphore.maxConcurrency, + type: 'developer', + counts: developerSemaphore.getCurrentActiveSlotCounts(), + }; + } + logger.audit('READ_DEV_CONCURRENCY', res.locals.adminActor, { campaign: GRANTFOX_FWC26_CAMPAIGN, developerCount: Object.keys(devCounts).length, @@ -135,6 +176,7 @@ export function createAdminDevMetricsRouter( devCounts, totalActive, maxConcurrencyPerDeveloper, + limiters, campaign: GRANTFOX_FWC26_CAMPAIGN, }, }); @@ -177,6 +219,7 @@ export function createAdminDevMetricsRouter( validate({ params: developerIdParamsSchema }), (req, res, next) => { try { + const developerSemaphore = getBillingRouteSemaphore(); const { developerId } = req.params; const activeCount = developerSemaphore.getActiveSlotCount(developerId); const atLimit = developerSemaphore.isAtLimit(developerId); diff --git a/src/services/billing.ts b/src/services/billing.ts index daf57546..b5d456c6 100644 --- a/src/services/billing.ts +++ b/src/services/billing.ts @@ -34,6 +34,7 @@ import type { Pool, PoolClient } from "pg"; import type { SimulationDetails } from "../lib/simulationDiagnostics.js"; import { computeJitteredDelay, type RandomSource } from "../lib/retry.js"; import { DeveloperSemaphore } from "../utils/developerSemaphore.js"; +import { concurrencyRegistry } from "./concurrencyRegistry.js"; import { logger } from "../logger.js"; import { parseUsdcToContractUnits, @@ -56,6 +57,7 @@ const DEFAULT_RETRY_DELAYS_MS = [150, 500, 1_000]; * leaving distinct users fully concurrent. */ export const billingConcurrencySemaphore = new DeveloperSemaphore(1); +concurrencyRegistry.register('billing.service', billingConcurrencySemaphore, 'developer'); export interface BillingDeductRequest { requestId: string; diff --git a/src/services/concurrencyRegistry.ts b/src/services/concurrencyRegistry.ts new file mode 100644 index 00000000..5289d689 --- /dev/null +++ b/src/services/concurrencyRegistry.ts @@ -0,0 +1,94 @@ +import { DeveloperSemaphore } from '../utils/developerSemaphore.js'; +import { KeySemaphore } from '../utils/keySemaphore.js'; +import client from 'prom-client'; +import { register } from '../metrics.js'; + +export type LimiterInstance = DeveloperSemaphore | KeySemaphore; + +export interface LimiterInfo { + name: string; + limiter: LimiterInstance; + type: 'developer' | 'key'; +} + +const concurrencyActiveSlots = new client.Gauge({ + name: 'concurrency_active_slots', + help: 'Total active concurrency slots held per limiter', + labelNames: ['limiter'], +}); + +const concurrencyQueued = new client.Gauge({ + name: 'concurrency_queued', + help: 'Total waiting/queued concurrency requests per limiter', + labelNames: ['limiter'], +}); + +register.registerMetric(concurrencyActiveSlots); +register.registerMetric(concurrencyQueued); + +export { concurrencyActiveSlots, concurrencyQueued }; + +export class ConcurrencyRegistry { + private readonly limiters = new Map(); + + register(name: string, limiter: LimiterInstance, type: 'developer' | 'key' = 'developer'): void { + limiter.setOnStateChange(() => { + this.updateMetrics(name); + }); + this.limiters.set(name, { name, limiter, type }); + this.updateMetrics(name); + } + + unregister(name: string): void { + const info = this.limiters.get(name); + if (info) { + info.limiter.setOnStateChange(undefined); + this.limiters.delete(name); + concurrencyActiveSlots.remove({ limiter: name }); + concurrencyQueued.remove({ limiter: name }); + } + } + + get(name: string): LimiterInfo | undefined { + return this.limiters.get(name); + } + + has(name: string): boolean { + return this.limiters.has(name); + } + + getAll(): LimiterInfo[] { + return Array.from(this.limiters.values()); + } + + clear(): void { + for (const [name, info] of this.limiters.entries()) { + info.limiter.setOnStateChange(undefined); + concurrencyActiveSlots.remove({ limiter: name }); + concurrencyQueued.remove({ limiter: name }); + } + this.limiters.clear(); + } + + updateMetrics(name?: string): void { + if (name) { + const info = this.limiters.get(name); + if (info) { + const active = info.limiter.getTotalActiveSlotCount(); + const queued = info.limiter.getTotalQueuedCount(); + concurrencyActiveSlots.set({ limiter: name }, active); + concurrencyQueued.set({ limiter: name }, queued); + } + return; + } + + for (const [limiterName, info] of this.limiters.entries()) { + const active = info.limiter.getTotalActiveSlotCount(); + const queued = info.limiter.getTotalQueuedCount(); + concurrencyActiveSlots.set({ limiter: limiterName }, active); + concurrencyQueued.set({ limiter: limiterName }, queued); + } + } +} + +export const concurrencyRegistry = new ConcurrencyRegistry(); diff --git a/src/utils/developerSemaphore.ts b/src/utils/developerSemaphore.ts index 915621a1..992764d2 100644 --- a/src/utils/developerSemaphore.ts +++ b/src/utils/developerSemaphore.ts @@ -17,12 +17,25 @@ interface DeveloperState { export class DeveloperSemaphore { private readonly developers = new Map(); + private onStateChange?: () => void; constructor( private readonly maxConcurrencyPerDeveloper = 1, private readonly ttlMs = 300_000, ) {} + setOnStateChange(listener?: () => void): void { + this.onStateChange = listener; + } + + private notifyStateChange(): void { + try { + this.onStateChange?.(); + } catch { + // Ignore listener error + } + } + async withSlot(developerId: string, fn: () => Promise): Promise { const release = await this.acquireSlot(developerId); try { @@ -50,6 +63,20 @@ export class DeveloperSemaphore { return total; } + /** Returns the total number of queued/waiting requests across all developers. */ + getTotalQueuedCount(): number { + let total = 0; + for (const state of this.developers.values()) { + total += state.queue.length; + } + return total; + } + + /** Returns the queue length for a specific developer. */ + getQueuedCount(developerId: string): number { + return this.developers.get(developerId)?.queue.length ?? 0; + } + /** Returns the active slot count for a specific developer, or 0 if not tracked. */ getActiveSlotCount(developerId: string): number { return this.developers.get(developerId)?.activeCount ?? 0; @@ -65,6 +92,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); @@ -73,14 +123,17 @@ export class DeveloperSemaphore { if (state.queue.length === 0 && state.activeCount < this.maxConcurrencyPerDeveloper) { this.clearEvictionTimer(state); state.activeCount += 1; + this.notifyStateChange(); return Promise.resolve(() => this.releaseSlot(developerId)); } return new Promise<() => void>((resolve) => { state.queue.push((release) => { this.clearEvictionTimer(state); + this.notifyStateChange(); resolve(release); }); + this.notifyStateChange(); }); } @@ -93,10 +146,12 @@ export class DeveloperSemaphore { if (state.queue.length > 0) { const next = state.queue.shift()!; next(() => this.releaseSlot(developerId)); + this.notifyStateChange(); return; } state.activeCount -= 1; + this.notifyStateChange(); if (state.activeCount === 0) { this.scheduleEviction(developerId, state); diff --git a/src/utils/keySemaphore.ts b/src/utils/keySemaphore.ts index 4cc4d6ab..a641f501 100644 --- a/src/utils/keySemaphore.ts +++ b/src/utils/keySemaphore.ts @@ -21,12 +21,25 @@ interface KeyState { export class KeySemaphore { private readonly keys = new Map(); + private onStateChange?: () => void; constructor( private readonly maxConcurrencyPerKey = 1, private readonly ttlMs = 300_000, ) {} + setOnStateChange(listener?: () => void): void { + this.onStateChange = listener; + } + + private notifyStateChange(): void { + try { + this.onStateChange?.(); + } catch { + // Ignore listener error + } + } + /** * Execute `fn` while holding a concurrency slot for the given API key. * The slot is automatically released when the returned promise settles. @@ -63,6 +76,20 @@ export class KeySemaphore { return total; } + /** Returns the total number of queued/waiting requests across all keys. */ + getTotalQueuedCount(): number { + let total = 0; + for (const state of this.keys.values()) { + total += state.queue.length; + } + return total; + } + + /** Returns the queue length for a specific key. */ + getQueuedCount(keyId: string): number { + return this.keys.get(keyId)?.queue.length ?? 0; + } + /** Returns the active slot count for a specific key, or 0 if not tracked. */ getActiveSlotCount(keyId: string): number { return this.keys.get(keyId)?.activeCount ?? 0; @@ -87,14 +114,17 @@ export class KeySemaphore { if (state.queue.length === 0 && state.activeCount < this.maxConcurrencyPerKey) { this.clearEvictionTimer(state); state.activeCount += 1; + this.notifyStateChange(); return Promise.resolve(() => this.releaseSlot(keyId)); } return new Promise<() => void>((resolve) => { state.queue.push((release) => { this.clearEvictionTimer(state); + this.notifyStateChange(); resolve(release); }); + this.notifyStateChange(); }); } @@ -107,10 +137,12 @@ export class KeySemaphore { if (state.queue.length > 0) { const next = state.queue.shift()!; next(() => this.releaseSlot(keyId)); + this.notifyStateChange(); return; } state.activeCount -= 1; + this.notifyStateChange(); if (state.activeCount === 0) { this.scheduleEviction(keyId, state);