diff --git a/packages/outpost/queue/src/handlers/__tests__/github-reaction-poll.test.ts b/packages/outpost/queue/src/handlers/__tests__/github-reaction-poll.test.ts index be33c11..3b8caf4 100644 --- a/packages/outpost/queue/src/handlers/__tests__/github-reaction-poll.test.ts +++ b/packages/outpost/queue/src/handlers/__tests__/github-reaction-poll.test.ts @@ -24,7 +24,10 @@ vi.mock('../../create-job.js', () => ({ createJob: (...args: unknown[]) => mockCreateJob(...args), })); -import { handleGithubReactionPoll } from '../github-reaction-poll.js'; +import { + handleGithubReactionPoll, + GITHUB_REACTION_POLL_CONCURRENCY, +} from '../github-reaction-poll.js'; import { JobType } from '../../types.js'; const context = { reportProgress: vi.fn().mockResolvedValue(undefined), jobId: 'job-1' }; @@ -214,4 +217,77 @@ describe('handleGithubReactionPoll', () => { }, }); }); + describe('GitHub lookup concurrency', () => { + // Production had 256 pending messages. Read one at a time at ~240ms each, + // the poll ran past its 30s job timeout on every attempt and dead-lettered. + function pendingMessages(count: number) { + return Array.from({ length: count }, (_, i) => ({ + id: `msg-${i}`, + ticketId: `ticket-${i}`, + externalCommentId: String(1000 + i), + ticket: { sourceId: 'owner/repo#42', user: { externalId: 'reporter-login' } }, + })); + } + + function trackInFlight() { + const state = { inFlight: 0, peak: 0 }; + mockListCommentReactions.mockImplementation(async () => { + state.inFlight += 1; + state.peak = Math.max(state.peak, state.inFlight); + await new Promise((resolve) => setTimeout(resolve, 5)); + state.inFlight -= 1; + return []; + }); + return state; + } + + it('reads reactions for several messages in parallel instead of one at a time', async () => { + mockMessageFindMany.mockResolvedValue(pendingMessages(40)); + const state = trackInFlight(); + + const result = await handleGithubReactionPoll({}, context); + + expect(state.peak).toBeGreaterThan(1); + expect(mockListCommentReactions).toHaveBeenCalledTimes(40); + expect(result).toEqual({ success: true, data: { checked: 40, updated: 0 } }); + }); + + it('caps the number of GitHub reads in flight', async () => { + mockMessageFindMany.mockResolvedValue(pendingMessages(40)); + const state = trackInFlight(); + + await handleGithubReactionPoll({}, context); + + expect(state.peak).toBe(GITHUB_REACTION_POLL_CONCURRENCY); + }); + + it('still applies every reporter reaction when reads run in parallel', async () => { + mockMessageFindMany.mockResolvedValue(pendingMessages(20)); + mockListCommentReactions.mockImplementation(async (_c, _o, _r, commentId: number) => { + await new Promise((resolve) => setTimeout(resolve, 1)); + if (commentId === 1003) return [{ content: '+1', login: 'reporter-login' }]; + if (commentId === 1017) return [{ content: '-1', login: 'reporter-login' }]; + if (commentId === 1010) throw new Error('Not Found'); + return []; + }); + + const result = await handleGithubReactionPoll({}, context); + + expect(mockMessageUpdate).toHaveBeenCalledTimes(2); + expect(mockMessageUpdate).toHaveBeenCalledWith({ + where: { id: 'msg-3' }, + data: { feedback: 'POSITIVE' }, + }); + expect(mockMessageUpdate).toHaveBeenCalledWith({ + where: { id: 'msg-17' }, + data: { feedback: 'NEGATIVE' }, + }); + expect(mockCreateJob).toHaveBeenCalledTimes(1); + expect(mockCreateJob).toHaveBeenCalledWith(JobType.ESCALATION, { + ticketId: 'ticket-17', + reason: expect.stringContaining('GitHub reaction'), + }); + expect(result).toEqual({ success: true, data: { checked: 20, updated: 2 } }); + }); + }); }); diff --git a/packages/outpost/queue/src/handlers/github-reaction-poll.ts b/packages/outpost/queue/src/handlers/github-reaction-poll.ts index 376b799..3bd8351 100644 --- a/packages/outpost/queue/src/handlers/github-reaction-poll.ts +++ b/packages/outpost/queue/src/handlers/github-reaction-poll.ts @@ -8,6 +8,12 @@ * ticket's original reporter reacted +1 or -1 on that specific comment. * A -1 also enqueues an ESCALATION job, matching Discord's * "Need more help" button behavior. + * + * Lookups run GITHUB_REACTION_POLL_CONCURRENCY at a time. One at a time, the + * poll needs one GitHub round trip per pending message, and that list only + * grows: comments deleted on GitHub stay pending forever. At 256 messages + * the serial loop overran the 30s job timeout on every attempt, and the job + * was dead-lettered after each scheduled run. */ import { prisma } from '@copilotkit/outpost/db'; @@ -16,6 +22,13 @@ import { createJob } from '../create-job.js'; import { JobType } from '../types.js'; import type { GithubReactionPollPayload, JobResult, JobHandlerContext } from '../types.js'; +/** + * Number of reaction lookups the poll keeps in flight. Well under GitHub's + * secondary rate limit of 100 concurrent requests, and enough to check + * about 1,000 messages inside the 30s job timeout. + */ +export const GITHUB_REACTION_POLL_CONCURRENCY = 8; + /** * Parse "owner/repo#number" into its components. */ @@ -73,7 +86,7 @@ export async function handleGithubReactionPoll( let checked = 0; let updated = 0; - for (const message of pendingMessages) { + async function checkMessage(message: (typeof pendingMessages)[number]): Promise { checked += 1; if ( @@ -81,11 +94,11 @@ export async function handleGithubReactionPoll( !message.ticket.sourceId || !message.ticket.user?.externalId ) { - continue; + return; } const parsed = parseSourceId(message.ticket.sourceId); - if (!parsed) continue; + if (!parsed) return; const reporterLogin = message.ticket.user.externalId; const commentId = Number(message.externalCommentId); @@ -98,11 +111,11 @@ export async function handleGithubReactionPoll( `[GithubReactionPoll] Failed to list reactions for message ${message.id}:`, error instanceof Error ? error.message : String(error), ); - continue; + return; } const reporterReaction = reactions.find((r) => r.login === reporterLogin); - if (!reporterReaction) continue; + if (!reporterReaction) return; if (reporterReaction.content === '+1') { await prisma.message.update({ @@ -123,6 +136,22 @@ export async function handleGithubReactionPoll( } } + let next = 0; + async function drain(): Promise { + while (next < pendingMessages.length) { + const message = pendingMessages[next]; + next += 1; + await checkMessage(message); + } + } + + await Promise.all( + Array.from( + { length: Math.min(GITHUB_REACTION_POLL_CONCURRENCY, pendingMessages.length) }, + drain, + ), + ); + await context.reportProgress(100); console.log(