diff --git a/apps/server/node/http.ts b/apps/server/node/http.ts index d986d224..0ccd6ae6 100644 --- a/apps/server/node/http.ts +++ b/apps/server/node/http.ts @@ -72,6 +72,7 @@ export function createServerApp(service: ServerService, options: ServerAppOption } const app = new Hono() const callbackWindows = new Map() + let callbackCleanupAt = 0 const logger = (options.logger ?? silentLogger).child({ component: 'http' }) const operator = options.operator const resolveActor = operator == null ? options.resolveControlActor : (request: Request) => operator.actor(request) @@ -101,7 +102,16 @@ export function createServerApp(service: ServerService, options: ServerAppOption }) app.route('/auth', createOperatorApp(operator, options.operatorLoginAttemptsPerMinute)) - const admitCallback = (key: string): number | undefined => callbackRetryAfter(callbackWindows, key, callbackRequestsPerMinute, Date.now()) + const admitCallback = (key: string): number | undefined => { + const now = Date.now() + if (now >= callbackCleanupAt) { + for (const [storedKey, window] of callbackWindows) { + if (window.resetAt <= now) callbackWindows.delete(storedKey) + } + callbackCleanupAt = now + 60_000 + } + return callbackRetryAfter(callbackWindows, key, callbackRequestsPerMinute, now) + } app.all('/v1/integrations', (context) => integration(service, context.req.raw, logger, context.get('requestId'), admitCallback)) app.all('/v1/integrations/*', (context) => integration(service, context.req.raw, logger, context.get('requestId'), admitCallback)) app.all('/v1/webhooks', (context) => webhook(service, context.req.raw, logger, context.get('requestId'), admitCallback)) diff --git a/apps/server/test/webhook.test.ts b/apps/server/test/webhook.test.ts index 18cb74a4..00358539 100644 --- a/apps/server/test/webhook.test.ts +++ b/apps/server/test/webhook.test.ts @@ -4,7 +4,7 @@ import { webhookEndpointId } from '@oomol-lab/open-flow/webhook-trigger' import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import path from 'node:path' -import { afterEach, describe, expect, it } from 'vitest' +import { afterEach, describe, expect, it, vi } from 'vitest' import { createServerApp } from '../node/http.ts' import { ServerService } from '../node/service.ts' import { storeRevision } from './runFixture.ts' @@ -168,6 +168,45 @@ describe('Server Webhook Trigger admission', () => { expect(Number(limited.headers.get('retry-after'))).toBeGreaterThan(0) }) + it('reclaims expired callback windows without resetting active limits', async () => { + const service = await openService(await databaseFile()) + services.push(service) + const target = await publishedWebhook(service) + const app = createServerApp(service, { callbackRequestsPerMinute: 1 }) + const url = `http://server.local/v1/webhooks/${target.endpointId}` + const clock = vi.spyOn(Date, 'now').mockReturnValue(0) + const writes = vi.spyOn(Map.prototype, 'set') + try { + expect((await app.request(url)).status).toBe(405) + const index = writes.mock.calls.findIndex(([key]) => key == `webhook:${target.endpointId}`) + const windows = writes.mock.contexts[index] as Map | undefined + if (windows == null) throw new Error('Callback window was not recorded.') + expect(windows.size).toBe(1) + + clock.mockReturnValue(30_000) + const waitUrl = 'http://server.local/v1/wait-actions/unknown/approve' + expect((await app.request(waitUrl)).status).toBe(404) + expect(windows.size).toBe(2) + + clock.mockReturnValue(60_000) + const limited = await app.request(waitUrl) + expect(limited.status).toBe(429) + expect(limited.headers.get('retry-after')).toBe('30') + expect(windows.has(`webhook:${target.endpointId}`)).toBe(false) + expect(windows.size).toBe(1) + + expect((await app.request(url)).status).toBe(405) + expect((await app.request(url)).status).toBe(429) + clock.mockReturnValue(120_000) + expect((await app.request(url)).status).toBe(405) + expect(windows.size).toBe(1) + expect(windows.has('wait-action')).toBe(false) + } finally { + writes.mockRestore() + clock.mockRestore() + } + }) + it('recovers a queued occurrence into the same Run after reopening SQLite', async () => { const file = await databaseFile() let service = await openService(file)