From d0e3ae17f8ceeb288879e54d2478763bee6c8bd0 Mon Sep 17 00:00:00 2001 From: "@daniel-lxs" <57051444+daniel-lxs@users.noreply.github.com> Date: Sat, 5 Sep 2026 20:26:49 +0000 Subject: [PATCH 1/2] fix: retain PR feedback claims during preparation retries --- .../pr-review-notification-retry.db.test.ts | 117 ++++++++++++++++++ .../src/jobs/pr-review-notification.test.ts | 48 +++++++ .../bullmq/src/jobs/pr-review-notification.ts | 17 ++- 3 files changed, 180 insertions(+), 2 deletions(-) create mode 100644 apps/bullmq/src/jobs/pr-review-notification-retry.db.test.ts diff --git a/apps/bullmq/src/jobs/pr-review-notification-retry.db.test.ts b/apps/bullmq/src/jobs/pr-review-notification-retry.db.test.ts new file mode 100644 index 000000000..58959816e --- /dev/null +++ b/apps/bullmq/src/jobs/pr-review-notification-retry.db.test.ts @@ -0,0 +1,117 @@ +import { randomUUID } from 'node:crypto'; +import { Queue, QueueEvents, Worker } from 'bullmq'; +import Redis from 'ioredis'; + +const prepareDelivery = vi.hoisted(() => vi.fn()); +vi.mock('@roomote/sdk/server', async (importOriginal) => ({ + ...(await importOriginal()), + preparePrReviewNotificationDelivery: prepareDelivery, +})); + +import { + claimDueCanonicalPrReviewDeliveries, + db, + eq, + persistPrReviewEvent, + prReviewNotificationDeliveries, + runFactory, + taskFactory, + taskPullRequests, +} from '@roomote/db/server'; +import { Env } from '@roomote/env'; +import { + finalizePrReviewNotificationRequest, + prReviewNotificationRequestSchema, + type PrReviewNotificationRequest, +} from '@roomote/sdk/server'; +import { RunStatus } from '@roomote/types'; + +import { prReviewNotificationJob } from './pr-review-notification'; + +it.each([false, true])( + 'uses the real queue retry without a scheduler drain and respects invalidated claims (invalidated: %s)', + async (invalidateClaim) => { + const task = await taskFactory.create(); + await runFactory.create({ taskId: task.id, status: RunStatus.Completed }); + const repository = `owner/retry-${task.id}`; + const prUrl = `https://github.com/${repository}/pull/1`; + await db.insert(taskPullRequests).values({ + taskId: task.id, + sourceControlProvider: 'github', + repository, + prNumber: 1, + prUrl, + status: 'open', + }); + await persistPrReviewEvent({ + eventKey: `retry-${task.id}`, + sourceControlProvider: 'github', + repository, + prNumber: 1, + prUrl, + event: { kind: 'review_comment', authorLogin: 'reviewer' }, + batchKind: 'human', + batchId: null, + dueAt: new Date(0), + observedAt: new Date(), + }); + const [claim] = await claimDueCanonicalPrReviewDeliveries(new Date(), { + repository, + }); + expect(claim).toBeDefined(); + const request = prReviewNotificationRequestSchema.parse({ + ...claim, + deliveryState: claim!.state, + }); + const readDelivery = () => + db.query.prReviewNotificationDeliveries.findFirst({ + where: eq(prReviewNotificationDeliveries.id, request.deliveryId!), + }); + + prepareDelivery.mockReset(); + prepareDelivery + .mockRejectedValueOnce(new Error('simulated triage timeout')) + .mockResolvedValue({ post: false, reason: 'not_worth_notifying' }); + const connection = new Redis(Env.REDIS_URL, { maxRetriesPerRequest: null }); + const queueName = `pr-review-retry-test-${randomUUID()}`; + const queue = new Queue(queueName, { + connection, + defaultJobOptions: { + attempts: 3, + backoff: { type: 'exponential', delay: 2000 }, + }, + }); + const queueEvents = new QueueEvents(queueName, { connection }); + const worker = new Worker(queueName, prReviewNotificationJob, { + connection, + }); + try { + await queueEvents.waitUntilReady(); + const job = await queue.add('notify-pr-review-activity', request); + await vi.waitFor(async () => { + expect(await job.getState()).toBe('delayed'); + }); + expect(await readDelivery()).toMatchObject({ + status: 'claimed', + leaseToken: request.leaseToken, + }); + if (invalidateClaim) { + await finalizePrReviewNotificationRequest(request, 'suppressed'); + } + + await job.waitUntilFinished(queueEvents, 10_000); + expect(prepareDelivery).toHaveBeenCalledTimes(invalidateClaim ? 1 : 2); + expect(await readDelivery()).toMatchObject({ + status: 'suppressed', + leaseToken: null, + }); + } finally { + await worker.close(); + await queueEvents.close(); + await queue.obliterate({ force: true }); + await queue.close(); + await connection.quit(); + } + }, + 15_000, +); diff --git a/apps/bullmq/src/jobs/pr-review-notification.test.ts b/apps/bullmq/src/jobs/pr-review-notification.test.ts index 614ec7c25..c6bcdbefa 100644 --- a/apps/bullmq/src/jobs/pr-review-notification.test.ts +++ b/apps/bullmq/src/jobs/pr-review-notification.test.ts @@ -1869,6 +1869,54 @@ describe('prReviewNotificationJob', () => { expect(mockRequeuePending).not.toHaveBeenCalled(); }); + it.each([0, 1, 2])( + 'retains a canonical preparation claim only while queue retries remain (prior failures: %s)', + async (attemptsMade) => { + mockPrepareDelivery.mockRejectedValue(new Error('model unavailable')); + const job = { + ...makeJob({ + ownershipVersion: 'canonical', + deliveryId: '11111111-1111-4111-8111-111111111111', + deliveryState: 'claimed', + deliveryIds: ['11111111-1111-4111-8111-111111111111'], + leaseToken: '22222222-2222-4222-8222-222222222222', + events, + }), + attemptsMade, + opts: { attempts: 3 }, + }; + + await expect(prReviewNotificationJob(job as never)).rejects.toThrow( + 'model unavailable', + ); + expect(mockRequeuePending).toHaveBeenCalledTimes( + attemptsMade === 2 ? 1 : 0, + ); + expect(mockPostMessage).not.toHaveBeenCalled(); + }, + ); + + it('releases a canonical claim when its preparation transition fails despite remaining retries', async () => { + mockPrepareCanonical.mockRejectedValue(new Error('transition failed')); + const job = { + ...makeJob({ + ownershipVersion: 'canonical', + deliveryId: '11111111-1111-4111-8111-111111111111', + deliveryState: 'claimed', + deliveryIds: ['11111111-1111-4111-8111-111111111111'], + leaseToken: '22222222-2222-4222-8222-222222222222', + events, + }), + attemptsMade: 0, + opts: { attempts: 3 }, + }; + + await expect(prReviewNotificationJob(job as never)).rejects.toThrow( + 'transition failed', + ); + expect(mockRequeuePending).toHaveBeenCalledOnce(); + }); + it('uses the canonical delivery id as the sole interactive action owner', async () => { const deliveryId = '11111111-1111-4111-8111-111111111111'; mockPrepareDelivery.mockResolvedValue({ diff --git a/apps/bullmq/src/jobs/pr-review-notification.ts b/apps/bullmq/src/jobs/pr-review-notification.ts index 0df511cf7..78b28a80e 100644 --- a/apps/bullmq/src/jobs/pr-review-notification.ts +++ b/apps/bullmq/src/jobs/pr-review-notification.ts @@ -528,6 +528,7 @@ export const prReviewNotificationJob = async ( const deliveryStartedAt = Date.now(); const telemetry = createPrReviewNotificationTelemetry(events.length); + let preparationCompleted = false; try { const delivery = await preparePrReviewNotificationDelivery({ @@ -536,6 +537,7 @@ export const prReviewNotificationJob = async ( events, telemetry, }); + preparationCompleted = true; logPrReviewNotificationTriage({ data, @@ -1171,11 +1173,22 @@ ${delivery.text}`; telemetry, }); - // Put the drained events back so a retried job can deliver them. + // BullMQ retries carry the same token. Releasing it here makes that retry + // look superseded and forces preparation to wait for the scheduled drain. + if ( + !preparationCompleted && + data.ownershipVersion === 'canonical' && + data.deliveryState === 'claimed' && + job.attemptsMade + 1 < Math.max(job.opts.attempts ?? 1, 1) + ) { + throw error; + } + + // Exhausted retries and failures after preparation need a fresh durable claim. try { await requeuePendingPrReviewActivity({ target, events }); } catch { - // Best effort; the events are lost if Redis is unavailable too. + // Best effort; lease expiry remains the recovery backstop. } throw error; From 410f5c998747bb91ecc875880bfb978e4beee3ac Mon Sep 17 00:00:00 2001 From: "@daniel-lxs" <57051444+daniel-lxs@users.noreply.github.com> Date: Sat, 5 Sep 2026 20:38:24 +0000 Subject: [PATCH 2/2] fix: pass CI Redis URL through Turbo --- turbo.json | 1 + 1 file changed, 1 insertion(+) diff --git a/turbo.json b/turbo.json index 779569710..4caf6e5cd 100644 --- a/turbo.json +++ b/turbo.json @@ -13,6 +13,7 @@ "globalPassThroughEnv": [ "APP_ENV", "DATABASE_URL", + "REDIS_URL", "MISE_DATA_DIR", "MISE_CACHE_DIR", "SKIP_ENV_VALIDATION",