From 77e2be5750bacac2f23615ee7955129c5bc0cdc0 Mon Sep 17 00:00:00 2001 From: "@daniel-lxs" <57051444+daniel-lxs@users.noreply.github.com> Date: Thu, 10 Sep 2026 06:03:34 +0000 Subject: [PATCH] fix: finalize unclassified webhook redeliveries --- .../github/__tests__/recordWebhook.test.ts | 148 ++++++++++++++- apps/api/src/handlers/github/recordWebhook.ts | 41 +++- .../recordWebhook.insertFailure.test.ts | 64 +++++++ .../linear/__tests__/recordWebhook.test.ts | 177 ++++++++++++++++++ apps/api/src/handlers/linear/recordWebhook.ts | 41 +++- 5 files changed, 468 insertions(+), 3 deletions(-) create mode 100644 apps/api/src/handlers/linear/__tests__/recordWebhook.insertFailure.test.ts create mode 100644 apps/api/src/handlers/linear/__tests__/recordWebhook.test.ts diff --git a/apps/api/src/handlers/github/__tests__/recordWebhook.test.ts b/apps/api/src/handlers/github/__tests__/recordWebhook.test.ts index d74137e6c2..6b38eee55a 100644 --- a/apps/api/src/handlers/github/__tests__/recordWebhook.test.ts +++ b/apps/api/src/handlers/github/__tests__/recordWebhook.test.ts @@ -1,11 +1,12 @@ // pnpm --filter @roomote/api test github/__tests__/recordWebhook.test.ts -import { db, webhooks, eq, inArray } from '@roomote/db/server'; +import { db, webhooks, eq, inArray, sql } from '@roomote/db/server'; import { recordWebhook } from '../recordWebhook'; describe('recordWebhook', () => { const testDeliveryIds: string[] = []; + let terminalFailureTriggerInstalled = false; const deleteTestData = async () => { if (testDeliveryIds.length > 0) { @@ -21,6 +22,16 @@ describe('recordWebhook', () => { }); afterEach(async () => { + if (terminalFailureTriggerInstalled) { + await db.execute( + sql`DROP TRIGGER test_fail_github_webhook_terminal_update ON webhooks`, + ); + await db.execute( + sql`DROP FUNCTION test_fail_github_webhook_terminal_update()`, + ); + terminalFailureTriggerInstalled = false; + } + vi.restoreAllMocks(); await deleteTestData(); }); @@ -258,6 +269,141 @@ describe('recordWebhook', () => { expect(records[0]!.payload).toEqual({ test: 'first' }); }); + it.each([ + { + outcome: 'successful GitHub', + provider: 'github' as const, + response: { status: 'ok' as const }, + }, + { + outcome: 'failed GitLab', + provider: 'gitlab' as const, + response: { status: 'error' as const, message: 'handler failed' }, + }, + ])( + 'finalizes an unclassified placeholder after a $outcome handler without replay', + async ({ provider, response }) => { + const deliveryId = `test-delivery-${Date.now()}-${response.status}-unclassified-placeholder`; + testDeliveryIds.push(deliveryId); + const handler = vi.fn(async () => response); + const consoleErrorSpy = vi + .spyOn(console, 'error') + .mockImplementation(() => {}); + vi.spyOn(console, 'warn').mockImplementation(() => {}); + await db.execute( + sql.raw(` + CREATE FUNCTION test_fail_github_webhook_terminal_update() + RETURNS trigger AS $$ + BEGIN + IF NEW.delivery_id = '${deliveryId}' + AND (NEW.succeeded_at IS NOT NULL OR NEW.failed_at IS NOT NULL) + THEN + RAISE EXCEPTION 'terminal update failed'; + END IF; + RETURN NEW; + END; + $$ LANGUAGE plpgsql + `), + ); + await db.execute(sql` + CREATE TRIGGER test_fail_github_webhook_terminal_update + BEFORE UPDATE ON webhooks + FOR EACH ROW EXECUTE FUNCTION test_fail_github_webhook_terminal_update() + `); + terminalFailureTriggerInstalled = true; + + await recordWebhook( + deliveryId, + 'pull_request.opened', + { test: 'first' }, + handler, + { provider }, + ); + + await db.execute( + sql`DROP TRIGGER test_fail_github_webhook_terminal_update ON webhooks`, + ); + await db.execute( + sql`DROP FUNCTION test_fail_github_webhook_terminal_update()`, + ); + terminalFailureTriggerInstalled = false; + + await recordWebhook( + deliveryId, + 'pull_request.opened', + { test: 'redelivery' }, + handler, + { provider }, + ); + + const [webhook] = await db + .select() + .from(webhooks) + .where(eq(webhooks.deliveryId, deliveryId)); + + expect(handler).toHaveBeenCalledTimes(1); + expect(webhook!.succeededAt).toBeNull(); + expect(webhook!.failedAt).not.toBeNull(); + expect(webhook!.error).toContain('outcome is unknown'); + expect(consoleErrorSpy).toHaveBeenCalledWith( + `[recordWebhook] Failed to update webhook ${deliveryId} for event pull_request.opened:`, + expect.any(String), + ); + }, + ); + + it('lets the original result replace concurrent redelivery recovery without replay', async () => { + const deliveryId = `test-delivery-${Date.now()}-concurrent`; + testDeliveryIds.push(deliveryId); + vi.spyOn(console, 'warn').mockImplementation(() => {}); + let releaseHandler!: () => void; + let handlerStarted!: () => void; + const started = new Promise((resolve) => { + handlerStarted = resolve; + }); + const release = new Promise((resolve) => { + releaseHandler = resolve; + }); + const handler = vi.fn(async () => { + handlerStarted(); + await release; + return { status: 'ok' as const }; + }); + + const firstDelivery = recordWebhook( + deliveryId, + 'pull_request.opened', + { test: 'first' }, + handler, + ); + await started; + await recordWebhook( + deliveryId, + 'pull_request.opened', + { test: 'concurrent' }, + handler, + ); + + const [inProgress] = await db + .select() + .from(webhooks) + .where(eq(webhooks.deliveryId, deliveryId)); + expect(inProgress!.succeededAt).toBeNull(); + expect(inProgress!.failedAt).not.toBeNull(); + expect(inProgress!.error).toContain('outcome is unknown'); + + releaseHandler(); + await firstDelivery; + + const [completed] = await db + .select() + .from(webhooks) + .where(eq(webhooks.deliveryId, deliveryId)); + expect(handler).toHaveBeenCalledTimes(1); + expect(completed!.succeededAt).not.toBeNull(); + expect(completed!.failedAt).toBeNull(); + }); + it('should record different event types correctly', async () => { const events = [ 'pull_request.opened', diff --git a/apps/api/src/handlers/github/recordWebhook.ts b/apps/api/src/handlers/github/recordWebhook.ts index 323018fb10..561ce472c0 100644 --- a/apps/api/src/handlers/github/recordWebhook.ts +++ b/apps/api/src/handlers/github/recordWebhook.ts @@ -1,9 +1,18 @@ -import { db, webhooks as webhooksTable, eq } from '@roomote/db/server'; +import { + and, + db, + webhooks as webhooksTable, + eq, + isNull, +} from '@roomote/db/server'; import type { SourceControlProvider } from '@roomote/types'; import type { WebhookResponse } from '../../types'; import { redactWebhookPayload } from '../webhook-payload-redaction'; +const UNKNOWN_HANDLER_OUTCOME_ERROR = + 'Webhook handler outcome is unknown because a redelivery found its durable claim nonterminal; the handler was not replayed'; + /** * Records a webhook after executing the handler, setting status based on the response. * Uses INSERT ... ON CONFLICT DO NOTHING to atomically claim the deliveryId before @@ -50,6 +59,36 @@ export async function recordWebhook( // Skip only if there was a conflict (not if there was a DB error) if (!hadInsertError && insertedRecord === undefined) { + // A nonterminal duplicate can be in progress or missing its final audit update. + // Never replay it; the original handler can still overwrite this unknown result. + try { + const [recovered] = await db + .update(webhooksTable) + .set({ + failedAt: new Date(), + error: UNKNOWN_HANDLER_OUTCOME_ERROR, + }) + .where( + and( + eq(webhooksTable.provider, provider), + eq(webhooksTable.deliveryId, deliveryId), + isNull(webhooksTable.succeededAt), + isNull(webhooksTable.failedAt), + ), + ) + .returning({ id: webhooksTable.id }); + + if (recovered) { + console.warn( + `[recordWebhook] Finalized unclassified ${provider} webhook ${deliveryId} without replaying its handler`, + ); + } + } catch (recoveryError) { + console.error( + `[recordWebhook] Failed to finalize unclassified ${provider} webhook ${deliveryId}:`, + recoveryError instanceof Error ? recoveryError.message : recoveryError, + ); + } return; } diff --git a/apps/api/src/handlers/linear/__tests__/recordWebhook.insertFailure.test.ts b/apps/api/src/handlers/linear/__tests__/recordWebhook.insertFailure.test.ts new file mode 100644 index 0000000000..42632156df --- /dev/null +++ b/apps/api/src/handlers/linear/__tests__/recordWebhook.insertFailure.test.ts @@ -0,0 +1,64 @@ +const dbMocks = vi.hoisted(() => ({ + insert: vi.fn(), + update: vi.fn(), + eq: vi.fn(), + webhooks: { + id: 'id', + }, +})); + +vi.mock('@roomote/db/server', () => ({ + and: vi.fn(), + db: { + insert: dbMocks.insert, + update: dbMocks.update, + }, + eq: dbMocks.eq, + isNull: vi.fn(), + webhooks: dbMocks.webhooks, +})); + +import { recordLinearWebhook } from '../recordWebhook'; + +describe('recordLinearWebhook insert failure fallback', () => { + beforeEach(() => { + vi.clearAllMocks(); + dbMocks.insert.mockReturnValue({ + values: () => ({ + onConflictDoNothing: () => ({ + returning: async () => { + throw new Error('insert failed'); + }, + }), + }), + }); + }); + + it('continues handler execution when the placeholder insert throws', async () => { + const handler = vi.fn(async () => ({ status: 'ok' as const })); + vi.spyOn(console, 'log').mockImplementation(() => {}); + const consoleErrorSpy = vi + .spyOn(console, 'error') + .mockImplementation(() => {}); + const consoleWarnSpy = vi + .spyOn(console, 'warn') + .mockImplementation(() => {}); + + await recordLinearWebhook( + 'linear-insert-failure', + 'Issue', + { test: 'payload' }, + handler, + ); + + expect(handler).toHaveBeenCalledTimes(1); + expect(dbMocks.update).not.toHaveBeenCalled(); + expect(consoleErrorSpy).toHaveBeenCalledWith( + '[recordLinearWebhook] Failed to insert placeholder for webhook linear-insert-failure - proceeding with handler anyway:', + 'insert failed', + ); + expect(consoleWarnSpy).toHaveBeenCalledWith( + '[recordLinearWebhook] Handler executed for webhook linear-insert-failure but no database record exists (insert failed)', + ); + }); +}); diff --git a/apps/api/src/handlers/linear/__tests__/recordWebhook.test.ts b/apps/api/src/handlers/linear/__tests__/recordWebhook.test.ts new file mode 100644 index 0000000000..c97e265485 --- /dev/null +++ b/apps/api/src/handlers/linear/__tests__/recordWebhook.test.ts @@ -0,0 +1,177 @@ +import { db, webhooks, eq, inArray, sql } from '@roomote/db/server'; + +import { recordLinearWebhook } from '../recordWebhook'; + +describe('recordLinearWebhook', () => { + const testDeliveryIds: string[] = []; + let terminalFailureTriggerInstalled = false; + + const deleteTestData = async () => { + if (testDeliveryIds.length > 0) { + await db + .delete(webhooks) + .where(inArray(webhooks.deliveryId, testDeliveryIds)); + } + }; + + beforeEach(async () => { + await deleteTestData(); + testDeliveryIds.length = 0; + }); + + afterEach(async () => { + if (terminalFailureTriggerInstalled) { + await db.execute( + sql`DROP TRIGGER test_fail_linear_webhook_terminal_update ON webhooks`, + ); + await db.execute( + sql`DROP FUNCTION test_fail_linear_webhook_terminal_update()`, + ); + terminalFailureTriggerInstalled = false; + } + vi.restoreAllMocks(); + await deleteTestData(); + }); + + it.each([ + { outcome: 'successful', response: { status: 'ok' as const } }, + { + outcome: 'failed', + response: { status: 'error' as const, message: 'handler failed' }, + }, + ])( + 'finalizes an unclassified placeholder after a $outcome handler without replay', + async ({ response }) => { + const webhookId = `linear-test-${Date.now()}-${response.status}-unclassified-placeholder`; + testDeliveryIds.push(webhookId); + const handler = vi.fn(async () => response); + vi.spyOn(console, 'log').mockImplementation(() => {}); + const consoleErrorSpy = vi + .spyOn(console, 'error') + .mockImplementation(() => {}); + vi.spyOn(console, 'warn').mockImplementation(() => {}); + await db.execute( + sql.raw(` + CREATE FUNCTION test_fail_linear_webhook_terminal_update() + RETURNS trigger AS $$ + BEGIN + IF NEW.delivery_id = '${webhookId}' + AND (NEW.succeeded_at IS NOT NULL OR NEW.failed_at IS NOT NULL) + THEN + RAISE EXCEPTION 'terminal update failed'; + END IF; + RETURN NEW; + END; + $$ LANGUAGE plpgsql + `), + ); + await db.execute(sql` + CREATE TRIGGER test_fail_linear_webhook_terminal_update + BEFORE UPDATE ON webhooks + FOR EACH ROW EXECUTE FUNCTION test_fail_linear_webhook_terminal_update() + `); + terminalFailureTriggerInstalled = true; + + await recordLinearWebhook(webhookId, 'Issue', { test: 'first' }, handler); + + await db.execute( + sql`DROP TRIGGER test_fail_linear_webhook_terminal_update ON webhooks`, + ); + await db.execute( + sql`DROP FUNCTION test_fail_linear_webhook_terminal_update()`, + ); + terminalFailureTriggerInstalled = false; + + await recordLinearWebhook( + webhookId, + 'Issue', + { test: 'redelivery' }, + handler, + ); + + const [webhook] = await db + .select() + .from(webhooks) + .where(eq(webhooks.deliveryId, webhookId)); + + expect(handler).toHaveBeenCalledTimes(1); + expect(webhook!.succeededAt).toBeNull(); + expect(webhook!.failedAt).not.toBeNull(); + expect(webhook!.error).toContain('outcome is unknown'); + expect(consoleErrorSpy).toHaveBeenCalledWith( + `[recordLinearWebhook] Failed to update webhook ${webhookId} for event Issue:`, + expect.any(String), + ); + }, + ); + + it('suppresses normal duplicate deliveries', async () => { + const webhookId = `linear-test-${Date.now()}-duplicate`; + testDeliveryIds.push(webhookId); + vi.spyOn(console, 'log').mockImplementation(() => {}); + const handler = vi.fn(async () => ({ status: 'ok' as const })); + + await recordLinearWebhook(webhookId, 'Issue', { test: 'first' }, handler); + await recordLinearWebhook( + webhookId, + 'Issue', + { test: 'duplicate' }, + handler, + ); + + expect(handler).toHaveBeenCalledTimes(1); + }); + + it('lets the original result replace concurrent redelivery recovery without replay', async () => { + const webhookId = `linear-test-${Date.now()}-concurrent`; + testDeliveryIds.push(webhookId); + vi.spyOn(console, 'log').mockImplementation(() => {}); + vi.spyOn(console, 'warn').mockImplementation(() => {}); + let releaseHandler!: () => void; + let handlerStarted!: () => void; + const started = new Promise((resolve) => { + handlerStarted = resolve; + }); + const release = new Promise((resolve) => { + releaseHandler = resolve; + }); + const handler = vi.fn(async () => { + handlerStarted(); + await release; + return { status: 'ok' as const }; + }); + + const firstDelivery = recordLinearWebhook( + webhookId, + 'Issue', + { test: 'first' }, + handler, + ); + await started; + await recordLinearWebhook( + webhookId, + 'Issue', + { test: 'concurrent' }, + handler, + ); + + const [inProgress] = await db + .select() + .from(webhooks) + .where(eq(webhooks.deliveryId, webhookId)); + expect(inProgress!.succeededAt).toBeNull(); + expect(inProgress!.failedAt).not.toBeNull(); + expect(inProgress!.error).toContain('outcome is unknown'); + + releaseHandler(); + await firstDelivery; + + const [completed] = await db + .select() + .from(webhooks) + .where(eq(webhooks.deliveryId, webhookId)); + expect(handler).toHaveBeenCalledTimes(1); + expect(completed!.succeededAt).not.toBeNull(); + expect(completed!.failedAt).toBeNull(); + }); +}); diff --git a/apps/api/src/handlers/linear/recordWebhook.ts b/apps/api/src/handlers/linear/recordWebhook.ts index eb15470a48..60fe88462f 100644 --- a/apps/api/src/handlers/linear/recordWebhook.ts +++ b/apps/api/src/handlers/linear/recordWebhook.ts @@ -1,8 +1,17 @@ -import { db, webhooks as webhooksTable, eq } from '@roomote/db/server'; +import { + and, + db, + webhooks as webhooksTable, + eq, + isNull, +} from '@roomote/db/server'; import type { WebhookResponse } from '../../types'; import { redactWebhookPayload } from '../webhook-payload-redaction'; +const UNKNOWN_HANDLER_OUTCOME_ERROR = + 'Webhook handler outcome is unknown because a redelivery found its durable claim nonterminal; the handler was not replayed'; + /** * Escape newline and carriage return characters to prevent log injection attacks. * This sanitizes untrusted input before including it in log messages. @@ -73,6 +82,36 @@ export async function recordLinearWebhook( // Skip only if there was a conflict (not if there was a DB error) if (!hadInsertError && insertedRecord === undefined) { + // A nonterminal duplicate can be in progress or missing its final audit update. + // Never replay it; the original handler can still overwrite this unknown result. + try { + const [recovered] = await db + .update(webhooksTable) + .set({ + failedAt: new Date(), + error: UNKNOWN_HANDLER_OUTCOME_ERROR, + }) + .where( + and( + eq(webhooksTable.provider, 'linear'), + eq(webhooksTable.deliveryId, webhookId), + isNull(webhooksTable.succeededAt), + isNull(webhooksTable.failedAt), + ), + ) + .returning({ id: webhooksTable.id }); + + if (recovered) { + console.warn( + `[recordLinearWebhook] Finalized unclassified webhook ${escapeForLog(webhookId)} without replaying its handler`, + ); + } + } catch (recoveryError) { + console.error( + `[recordLinearWebhook] Failed to finalize unclassified webhook ${escapeForLog(webhookId)}:`, + recoveryError instanceof Error ? recoveryError.message : recoveryError, + ); + } return; }