fix(server): reclaim expired callback rate-limit windows

apps/server/node/http.ts

@@ -72,6 +72,7 @@ export function createServerApp(service: ServerService, options: ServerAppOption
   }
   const app = new Hono<AppEnv>()
   const callbackWindows = new Map<string, CallbackWindow>()
+  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))

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<string, unknown> | 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)