fix(queue): run GitHub reaction lookups in parallel so the poll fits its timeout

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 } });
+        });
+    });
 });

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,19 +86,19 @@ export async function handleGithubReactionPoll(
     let checked = 0;
     let updated = 0;
 
-    for (const message of pendingMessages) {
+    async function checkMessage(message: (typeof pendingMessages)[number]): Promise<void> {
         checked += 1;
 
         if (
             !message.externalCommentId ||
             !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<void> {
+        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(