* feat(webhooks): dispatch on emit instead of waiting for the next cron tick The webhook dispatcher ran only on a per-minute cron, so the floor on delivery latency was up to 60 seconds plus the request. An external consumer that wanted to react as a transaction landed had only one alternative: polling /api/events, which the 100 rpm per-key limit makes expensive and which still cannot beat the tick interval. Schedules one dispatch cycle as soon as deliveries are enqueued. The cron is unchanged and remains the retry and sweep path; this only moves the first attempt forward. Wired into the event-bus fanout plus the two routes that enqueue a delivery directly: the :test verb, whose entire purpose is telling someone whether their receiver works, and the manual delivery retry. Three properties are load-bearing and covered by tests. The kick is never awaited, because eventBus.emit is awaited at ~99 call sites including journal_entry.committed and each delivery can burn a 10 s receiver timeout. It coalesces per function instance, so a bulk booking that emits once per row does not schedule one claim round trip per row. It claims 5 rows rather than the cron's 50, because it runs on the tail of a user-facing request. Double delivery is not a risk: claim_due_webhook_deliveries already claims FOR UPDATE SKIP LOCKED and flips rows to in_flight in the same statement, so a kick racing the cron sees disjoint rows. Does not close #1201, which asks for a realtime stream for API consumers. This is the cheap half. Refs #1201 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * docs(webhooks): stop claiming the kick makes double delivery impossible Adversarial review of the previous commit caught an overstatement in its own comments. SKIP LOCKED keeps a kick and the cron from claiming the same row at the same moment, but claim_due_webhook_deliveries autocommits before any POST is issued, so from then on ownership is only status='in_flight' and a later cycle's recoverStuckInFlight sweep can re-arm a row still queued behind an earlier cycle's serial loop. Delivery is at-least-once, which is what the public docs already tell receivers ("the same delivery id may arrive more than once ... idempotency is on you"). The comments contradicted that. No behaviour change. The kick does not create this window: the cron claims 50 rows serially against the same 20 s stuck threshold, which is wider than what a batch of 5 can open. Refs #1201 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
111 lines
3.7 KiB
TypeScript
111 lines
3.7 KiB
TypeScript
import { describe, it, expect, vi, beforeEach } from 'vitest'
|
|
|
|
vi.mock('@/lib/auth/api-keys', () => ({
|
|
createServiceClientNoCookies: vi.fn(() => ({ __client: true })),
|
|
}))
|
|
|
|
// `after` is a no-op outside a request scope in these tests; the module falls
|
|
// back to a floating promise, which is the path a plain node server takes too.
|
|
vi.mock('next/server', () => ({
|
|
after: () => {
|
|
throw new Error('after() called outside a request scope')
|
|
},
|
|
}))
|
|
|
|
import { kickWebhookDispatch, resetKickStateForTests, KICK_BATCH_SIZE } from '../dispatch-kick'
|
|
|
|
/** Resolves after the microtask queue drains, so floating work has run. */
|
|
const flush = () => new Promise((resolve) => setTimeout(resolve, 0))
|
|
|
|
beforeEach(() => {
|
|
vi.clearAllMocks()
|
|
resetKickStateForTests()
|
|
})
|
|
|
|
describe('kickWebhookDispatch', () => {
|
|
it('runs one dispatch cycle with the small kick batch size', async () => {
|
|
const dispatch = vi.fn().mockResolvedValue({ picked: 1, delivered: 1, failed: 0, dead: 0 })
|
|
|
|
kickWebhookDispatch(dispatch)
|
|
await flush()
|
|
|
|
expect(dispatch).toHaveBeenCalledTimes(1)
|
|
expect(dispatch.mock.calls[0][0]).toMatchObject({ batchSize: KICK_BATCH_SIZE })
|
|
})
|
|
|
|
it('claims far fewer rows than the cron, so a backlog cannot own a request tail', () => {
|
|
// Each delivery can burn the full 10 s receiver timeout; the cron's 50 is
|
|
// fine on a dedicated invocation, not on the tail of a user request.
|
|
expect(KICK_BATCH_SIZE).toBeLessThan(50)
|
|
})
|
|
|
|
it('returns synchronously: the emitter never waits on receiver HTTP', async () => {
|
|
let settled = false
|
|
const dispatch = vi.fn(
|
|
() =>
|
|
new Promise((resolve) => {
|
|
setTimeout(() => {
|
|
settled = true
|
|
resolve({ picked: 0, delivered: 0, failed: 0, dead: 0 })
|
|
}, 20)
|
|
}),
|
|
)
|
|
|
|
const returned = kickWebhookDispatch(dispatch as never)
|
|
|
|
// eventBus.emit awaits its subscribers, so a kick that blocked here would
|
|
// put a stranger's slow endpoint on the critical path of committing a
|
|
// verifikat.
|
|
expect(returned).toBeUndefined()
|
|
expect(settled).toBe(false)
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, 40))
|
|
expect(settled).toBe(true)
|
|
})
|
|
|
|
it('coalesces a burst of kicks into a single cycle', async () => {
|
|
// A bulk booking emits once per row. Without coalescing, 100 rows would
|
|
// mean 100 claim round trips against the same handful of due deliveries.
|
|
const dispatch = vi.fn().mockResolvedValue({ picked: 0, delivered: 0, failed: 0, dead: 0 })
|
|
|
|
for (let i = 0; i < 100; i++) kickWebhookDispatch(dispatch)
|
|
await flush()
|
|
|
|
expect(dispatch).toHaveBeenCalledTimes(1)
|
|
})
|
|
|
|
it('accepts a new kick once the previous cycle has started', async () => {
|
|
const dispatch = vi.fn().mockResolvedValue({ picked: 0, delivered: 0, failed: 0, dead: 0 })
|
|
|
|
kickWebhookDispatch(dispatch)
|
|
await flush()
|
|
kickWebhookDispatch(dispatch)
|
|
await flush()
|
|
|
|
expect(dispatch).toHaveBeenCalledTimes(2)
|
|
})
|
|
|
|
it('swallows a dispatch failure: a failed kick is latency, not a lost delivery', async () => {
|
|
const dispatch = vi.fn().mockRejectedValue(new Error('receiver unreachable'))
|
|
|
|
expect(() => kickWebhookDispatch(dispatch)).not.toThrow()
|
|
await flush()
|
|
|
|
expect(dispatch).toHaveBeenCalledTimes(1)
|
|
})
|
|
|
|
it('does not wedge the coalescing flag when a cycle fails', async () => {
|
|
// The cron still covers every row, but a stuck flag would silently
|
|
// disable the fast path for the life of the function instance.
|
|
const failing = vi.fn().mockRejectedValue(new Error('boom'))
|
|
kickWebhookDispatch(failing)
|
|
await flush()
|
|
|
|
const ok = vi.fn().mockResolvedValue({ picked: 0, delivered: 0, failed: 0, dead: 0 })
|
|
kickWebhookDispatch(ok)
|
|
await flush()
|
|
|
|
expect(ok).toHaveBeenCalledTimes(1)
|
|
})
|
|
})
|