From 3cbab5bdc160424992feea788b60051f250002ae Mon Sep 17 00:00:00 2001 From: Daniel Riccio <57051444+daniel-lxs@users.noreply.github.com> Date: Thu, 23 Jul 2026 20:43:36 +0000 Subject: [PATCH] fix: prevent stale PR review actions --- .../src/server/__tests__/enqueue-task.test.ts | 12 ++- .../cloud-agents/src/server/task-run-queue.ts | 6 +- .../__tests__/pr-review-action.test.ts | 52 +++++++++++++ .../server/lib/task-runs/pr-review-action.ts | 77 +++++++++++-------- 4 files changed, 110 insertions(+), 37 deletions(-) create mode 100644 packages/sdk/src/server/lib/task-runs/__tests__/pr-review-action.test.ts diff --git a/packages/cloud-agents/src/server/__tests__/enqueue-task.test.ts b/packages/cloud-agents/src/server/__tests__/enqueue-task.test.ts index e74fb187..ccaee1de 100644 --- a/packages/cloud-agents/src/server/__tests__/enqueue-task.test.ts +++ b/packages/cloud-agents/src/server/__tests__/enqueue-task.test.ts @@ -841,6 +841,7 @@ describe('enqueueTask snapshot resume', () => { repo: 'acme/widgets', description: 'Do the thing', sourceControlProvider: 'ado', + sourceControlHost: 'dev.azure.com', }, }), initiator: { kind: 'user', userId }, @@ -865,9 +866,16 @@ describe('enqueueTask snapshot resume', () => { ); expect( - (resumeRun.payload as { sourceControlProvider?: string }) - .sourceControlProvider, + ( + resumeRun.payload as { + sourceControlProvider?: string; + sourceControlHost?: string; + } + ).sourceControlProvider, ).toBe('gitea'); + expect( + (resumeRun.payload as { sourceControlHost?: string }).sourceControlHost, + ).toBeUndefined(); }); it('rejects a resume without a source run id', async () => { diff --git a/packages/cloud-agents/src/server/task-run-queue.ts b/packages/cloud-agents/src/server/task-run-queue.ts index 52613f4f..cb750a02 100644 --- a/packages/cloud-agents/src/server/task-run-queue.ts +++ b/packages/cloud-agents/src/server/task-run-queue.ts @@ -1917,7 +1917,9 @@ function inheritSnapshotResumeSourceControlStamps( sourceControlHost?: unknown; }; - if (payload.sourceControlProvider === undefined) { + const inheritsProvider = payload.sourceControlProvider === undefined; + + if (inheritsProvider) { const provider = sourceControlProviderSchema.safeParse( source.sourceControlProvider, ); @@ -1927,7 +1929,7 @@ function inheritSnapshotResumeSourceControlStamps( } } - if (payload.sourceControlHost === undefined) { + if (inheritsProvider && payload.sourceControlHost === undefined) { const host = typeof source.sourceControlHost === 'string' ? source.sourceControlHost.trim() diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-action.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-action.test.ts new file mode 100644 index 00000000..db0e813f --- /dev/null +++ b/packages/sdk/src/server/lib/task-runs/__tests__/pr-review-action.test.ts @@ -0,0 +1,52 @@ +const mockEval = vi.fn(); + +vi.mock('@roomote/redis', () => ({ + getRedis: () => ({ eval: mockEval }), +})); + +import { + attachPendingPrReviewActionMessage, + claimPendingPrReviewActionsForThread, +} from '../pr-review-action'; + +describe('PR review action state', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('attaches notification ids with an atomic compare-and-update script', async () => { + mockEval.mockResolvedValue(1); + + await attachPendingPrReviewActionMessage('nonce-1', 'message-1'); + + expect(mockEval).toHaveBeenCalledWith( + expect.stringContaining("redis.call('get', KEYS[1])"), + 1, + 'pr-review-action:nonce-1', + 'message-1', + ); + expect(mockEval.mock.calls[0]?.[0]).toContain("'KEEPTTL'"); + }); + + it('claims every indexed offer through one atomic script', async () => { + mockEval.mockResolvedValue([ + JSON.stringify({ nonce: 'nonce-1', messageId: 'message-1' }), + ]); + + await expect( + claimPendingPrReviewActionsForThread({ + provider: 'discord', + channelId: 'channel-1', + threadId: 'thread-1', + }), + ).resolves.toEqual([{ nonce: 'nonce-1', messageId: 'message-1' }]); + + expect(mockEval).toHaveBeenCalledWith( + expect.stringContaining("redis.call('smembers', KEYS[1])"), + 1, + 'pr-review-action:thread:discord:channel-1:thread-1', + 'pr-review-action:', + ); + expect(mockEval.mock.calls[0]?.[0]).toContain("redis.call('del', KEYS[1])"); + }); +}); diff --git a/packages/sdk/src/server/lib/task-runs/pr-review-action.ts b/packages/sdk/src/server/lib/task-runs/pr-review-action.ts index 4c333774..e0304be6 100644 --- a/packages/sdk/src/server/lib/task-runs/pr-review-action.ts +++ b/packages/sdk/src/server/lib/task-runs/pr-review-action.ts @@ -51,6 +51,35 @@ redis.call('del', KEYS[1]) return val `; +// Attaching a message must not revive an offer that a typed reply or button +// click claimed after the notification was posted. +const ATTACH_PR_REVIEW_ACTION_MESSAGE_LUA = ` +local val = redis.call('get', KEYS[1]) +if not val then return 0 end +local pending = cjson.decode(val) +pending.messageId = ARGV[1] +redis.call('set', KEYS[1], cjson.encode(pending), 'KEEPTTL') +return 1 +`; + +// Read, clear, and claim the complete conversation index in one operation so +// offers added concurrently remain indexed for a later typed reply. +const CLAIM_PR_REVIEW_ACTIONS_FOR_THREAD_LUA = ` +local nonces = redis.call('smembers', KEYS[1]) +if #nonces == 0 then return {} end +redis.call('del', KEYS[1]) +local claimed = {} +for _, nonce in ipairs(nonces) do + local actionKey = ARGV[1] .. nonce + local val = redis.call('get', actionKey) + if val then + redis.call('del', actionKey) + table.insert(claimed, val) + end +end +return claimed +`; + function getPrReviewActionKey(nonce: string): string { return `${PR_REVIEW_ACTION_PREFIX}${nonce}`; } @@ -94,25 +123,14 @@ export async function attachPendingPrReviewActionMessage( messageId: string, ): Promise { const redis = getRedis(); - const key = getPrReviewActionKey(nonce); - const raw = await redis.get(key); - - if (typeof raw !== 'string') { - return; - } - - try { - const pending = JSON.parse(raw) as PendingPrReviewAction; - - await redis.set( - key, - JSON.stringify({ ...pending, messageId }), - 'EX', - PR_REVIEW_ACTION_TTL_SECONDS, - ); - } catch { - // Malformed record; leave it to expire. - } + await redis + .eval( + ATTACH_PR_REVIEW_ACTION_MESSAGE_LUA, + 1, + getPrReviewActionKey(nonce), + messageId, + ) + .catch(() => undefined); } export async function claimPendingPrReviewAction( @@ -155,23 +173,16 @@ export async function claimPendingPrReviewActionsForThread(input: { }): Promise { const redis = getRedis(); const threadKey = getPrReviewActionThreadKey(input); - const nonces = await redis.smembers(threadKey); - - if (nonces.length === 0) { - return []; - } - - await redis.del(threadKey).catch(() => undefined); + const rawClaims = await redis.eval( + CLAIM_PR_REVIEW_ACTIONS_FOR_THREAD_LUA, + 1, + threadKey, + PR_REVIEW_ACTION_PREFIX, + ); const claimed: PendingPrReviewAction[] = []; - for (const nonce of nonces) { - const raw = await redis.eval( - CLAIM_PR_REVIEW_ACTION_LUA, - 1, - getPrReviewActionKey(nonce), - ); - + for (const raw of Array.isArray(rawClaims) ? rawClaims : []) { if (typeof raw !== 'string') { continue; }