From 3f34a8bb47d177aaa07afc5d0a9dda80920abcfd Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 13 Nov 2026 05:16:35 -0800 Subject: [PATCH] fix(github): preserve shared producer settlement and selected demand Co-authored-by: Jinjing <6427696+AmethystLiang@users.noreply.github.com> --- .../client/lookup/pr-for-branch-outcome.ts | 11 +- .../refresh/hosted-pr-read-jobs.ts | 43 +++-- .../pr-refresh-coordinator-test-mocks.ts | 4 +- .../refresh/pr-refresh-immediate-read.ts | 31 ++-- .../refresh/pr-refresh-pacing.ts | 13 +- .../refresh/pr-refresh-queue-drainer.ts | 48 +++--- .../refresh/pr-refresh-watch-jobs.ts | 3 +- ...nvironments-subscription-lifecycle.test.ts | 11 +- ...osted-pr-producer-settlement-test-cases.ts | 119 +++++++++++++- .../reviews/hosted-pr-selected-demand.test.ts | 142 ++++++++++++++++ ...runtime-hosted-pr-snapshot-test-harness.ts | 24 ++- .../reviews/runtime-hosted-review-commands.ts | 6 +- .../github/github-pull-request-methods.ts | 3 +- .../src/store/github/github-slice-contract.ts | 1 + .../github-pr-hosted-receipt-loop.test.ts | 151 ++++++++++++++++-- .../pull-requests/pull-request-execution.ts | 9 +- .../visible-hosted-review-refresh-targets.ts | 1 + .../web-runtime-pr-subscription-queue.test.ts | 9 +- .../github-pull-request-params.ts | 1 + 19 files changed, 540 insertions(+), 90 deletions(-) create mode 100644 src/main/runtime/integrations/reviews/hosted-pr-selected-demand.test.ts diff --git a/src/main/github/client/lookup/pr-for-branch-outcome.ts b/src/main/github/client/lookup/pr-for-branch-outcome.ts index d42827627c7..a3a59d4c9b7 100644 --- a/src/main/github/client/lookup/pr-for-branch-outcome.ts +++ b/src/main/github/client/lookup/pr-for-branch-outcome.ts @@ -152,7 +152,7 @@ export async function preparePRBranchRead( runCoalescedProbe( reads, coalescingKey, - async () => { + async (ownsKey) => { const start = startPRLookup() const retained = !narrowed && @@ -206,8 +206,8 @@ export async function preparePRBranchRead( }) } return settle && - (!retained || readTimedPROutcome(ownerKey, key)?.outcome === outcome) - ? settle(outcome) + (retained ? readTimedPROutcome(ownerKey, key)?.outcome === outcome : ownsKey()) + ? settle(outcome, { retained }) : outcome } finally { if (retained) { @@ -267,7 +267,10 @@ export async function preparePRBranchRead( ) } -export type PRReadSettlement = (outcome: PRRefreshOutcome) => PRRefreshOutcome +export type PRReadSettlement = ( + outcome: PRRefreshOutcome, + producer: { retained: boolean } +) => PRRefreshOutcome export type PreparedPRBranchRead = { key: string diff --git a/src/main/github/pull-requests/refresh/hosted-pr-read-jobs.ts b/src/main/github/pull-requests/refresh/hosted-pr-read-jobs.ts index 48f3ad9da7b..c1456b49692 100644 --- a/src/main/github/pull-requests/refresh/hosted-pr-read-jobs.ts +++ b/src/main/github/pull-requests/refresh/hosted-pr-read-jobs.ts @@ -19,6 +19,7 @@ export type HostedPRReadJob = { prepared: PreparedPRBranchRead expiresAt: number running?: boolean + promoteSelected: () => void listeners: Set<(outcome: PRRefreshOutcome | null) => void> finish: (outcome: PRRefreshOutcome | null) => void } @@ -142,7 +143,13 @@ export class HostedPRReadJobs { return outcome } const key = `hosted::${prepared.key}` - return this.retry.settleProducerError(outcome, key, candidate, prepared, this.queue) + return this.retry.settleProducerError( + outcome, + key, + { ...candidate, isSelected: false }, + prepared, + this.queue + ) }) : prepared.read()) if (!outcome) { @@ -194,9 +201,26 @@ export class HostedPRReadJobs { } const listeners: HostedPRReadJob['listeners'] = new Set() const releaseHold = prepared.hold() + const expiresAt = Date.now() + JOB_LEASE_MS + const entry: PRRefreshQueueEntry = { + key, + candidate, + aliases: new Map(), + reason: 'swr', + priority: candidate.isSelected ? 80 : prepared.cached()?.quietIntervalMs ? 20 : 40, + dueAt: Math.min(expiresAt, Math.max(Date.now(), gate)), + queuedAt: this.queue.nextOrder() + } job = { prepared, - expiresAt: Date.now() + JOB_LEASE_MS, + promoteSelected: () => { + if (!entry.candidate.isSelected) { + entry.candidate = { ...entry.candidate, isSelected: true } + entry.priority = 80 + this.wake() + } + }, + expiresAt, listeners, finish: (outcome) => { releaseHold() @@ -211,18 +235,13 @@ export class HostedPRReadJobs { } } this.jobs.set(key, job) - this.queue.set(key, { - key, - candidate, - aliases: new Map(), - reason: 'swr', - priority: prepared.cached()?.quietIntervalMs ? 20 : 40, - dueAt: Math.min(job.expiresAt, Math.max(Date.now(), gate)), - queuedAt: this.queue.nextOrder(), - hosted: job - }) + entry.hosted = job + this.queue.set(key, entry) this.wake() } + if (candidate.isSelected && !job.running) { + job.promoteSelected() + } if ( !acceptRevalidationOnly || this.waiters >= MAX_WAITERS || diff --git a/src/main/github/pull-requests/refresh/pr-refresh-coordinator-test-mocks.ts b/src/main/github/pull-requests/refresh/pr-refresh-coordinator-test-mocks.ts index 225b1c354ef..847419f90d9 100644 --- a/src/main/github/pull-requests/refresh/pr-refresh-coordinator-test-mocks.ts +++ b/src/main/github/pull-requests/refresh/pr-refresh-coordinator-test-mocks.ts @@ -79,11 +79,11 @@ export function clientModuleMock(mocks: PRRefreshCoordinatorMocks): ClientModule withExecution: (operation) => operation(), read: async (settle) => { const outcome = await mocks.getPRForBranchOutcomeMock(...args) - return settle ? settle(outcome) : outcome + return settle ? settle(outcome, { retained: true }) : outcome }, readRetained: async (settle) => { const outcome = await mocks.getPRForBranchOutcomeMock(...args) - return settle ? settle(outcome) : outcome + return settle ? settle(outcome, { retained: true }) : outcome } } return prepared diff --git a/src/main/github/pull-requests/refresh/pr-refresh-immediate-read.ts b/src/main/github/pull-requests/refresh/pr-refresh-immediate-read.ts index 956ef659bf4..97cacf66025 100644 --- a/src/main/github/pull-requests/refresh/pr-refresh-immediate-read.ts +++ b/src/main/github/pull-requests/refresh/pr-refresh-immediate-read.ts @@ -115,22 +115,22 @@ export async function runImmediatePRRefresh( ...hostedReviewOptionArgs(candidate, reason, options) ) try { - const outcome = await prepared.read((outcome) => { - if ( - outcome.kind !== 'upstream-error' || - !prepared.isCurrent() || - !queue.ownsRequest(key, requestSequence) || - !visibility.has(key) - ) { + const outcome = await prepared.read((outcome, producer) => { + if (!prepared.isCurrent()) { return outcome } - return retry.settleProducerError( - outcome, - key, - visibility.candidate(key, candidate), - prepared, - queue - ) + const settled = + producer.retained && outcome.kind === 'upstream-error' && visibility.has(key) + ? retry.settleProducerError( + outcome, + key, + visibility.candidate(key, candidate), + prepared, + queue + ) + : outcome + events.observe(candidate, settled, aliases) + return settled }) if (!queue.ownsRequest(key, requestSequence)) { events.broadcast({ aliases, reason, outcome, requestStartedAt }, requestSequence) @@ -144,9 +144,6 @@ export async function runImmediatePRRefresh( retry.nextVisibleErrorRetryAt(key, visibility.candidate(key, candidate)) broadcastOutcome = retry.withErrorSchedule(outcome, plannedRetryAt) } - if (prepared.didRead()) { - events.observe(candidate, outcome, aliases) - } retry.noteManualGate(key, broadcastOutcome) events.broadcast( { aliases, reason, outcome: broadcastOutcome, requestStartedAt }, diff --git a/src/main/github/pull-requests/refresh/pr-refresh-pacing.ts b/src/main/github/pull-requests/refresh/pr-refresh-pacing.ts index 2013a5e05ae..0353b30c4b4 100644 --- a/src/main/github/pull-requests/refresh/pr-refresh-pacing.ts +++ b/src/main/github/pull-requests/refresh/pr-refresh-pacing.ts @@ -7,6 +7,13 @@ const BACKGROUND_BUDGET_MAX = 20 const ACTIVE_BURST_WINDOW_MS = 30_000 const ACTIVE_BURST_MAX = 3 +export function isSelectedBudgetedPRRefresh(entry: PRRefreshQueueEntry): boolean { + return ( + entry.candidate.isSelected === true && + (entry.reason === 'visible' || (entry.reason === 'swr' && entry.hosted !== undefined)) + ) +} + export function usesActiveRefreshPacing(entry: PRRefreshQueueEntry): boolean { // Selected admission survives deselection while queued; periodic follow-ups clear the flag. return ( @@ -31,8 +38,7 @@ export class PRRefreshPacing { } noteBackgroundStart(entry: PRRefreshQueueEntry): void { - this.lastBudgetedStartWasSelected = - entry.reason === 'visible' && entry.candidate.isSelected === true + this.lastBudgetedStartWasSelected = isSelectedBudgetedPRRefresh(entry) const now = Date.now() this.lastBackgroundStartAt = now this.backgroundStarts.push(now) @@ -70,8 +76,7 @@ export class PRRefreshPacing { private priority(entry: PRRefreshQueueEntry): number { // Reserve the next budgeted slot for other ready reviews after a selected list read. return this.lastBudgetedStartWasSelected && - entry.reason === 'visible' && - entry.candidate.isSelected === true && + isSelectedBudgetedPRRefresh(entry) && entry.bypassBackgroundBudget !== true ? 0 : entry.priority diff --git a/src/main/github/pull-requests/refresh/pr-refresh-queue-drainer.ts b/src/main/github/pull-requests/refresh/pr-refresh-queue-drainer.ts index 8923f4def98..11373b21bb9 100644 --- a/src/main/github/pull-requests/refresh/pr-refresh-queue-drainer.ts +++ b/src/main/github/pull-requests/refresh/pr-refresh-queue-drainer.ts @@ -1,8 +1,8 @@ -import type { - GitHubPRRefreshCandidate, - PRRefreshOutcome -} from '../../../../shared/github/pull-requests/refresh/pull-request-refresh-types' -import { preparePRBranchRead } from '../../client/lookup/pr-for-branch-outcome' +import type { GitHubPRRefreshCandidate } from '../../../../shared/github/pull-requests/refresh/pull-request-refresh-types' +import { + preparePRBranchRead, + type PRReadSettlement +} from '../../client/lookup/pr-for-branch-outcome' import { hostedReviewOptionArgs, isBackground, @@ -179,7 +179,8 @@ export class PRRefreshQueueDrainer { if (pausedUntil !== null) { this.queue.protectBackgroundUntil(next.key, pausedUntil) const dueAt = Math.min(pausedUntil, next.hosted?.expiresAt ?? Infinity) - this.queue.set(next.key, { ...next, dueAt }) + next.dueAt = dueAt + this.queue.set(next.key, next) if (!next.hosted) { this.events.broadcast({ aliases, @@ -224,22 +225,26 @@ export class PRRefreshQueueDrainer { watchPRKey: next.watch?.prKey }) )) - const settle = (outcome: PRRefreshOutcome): PRRefreshOutcome => { - if ( - outcome.kind !== 'upstream-error' || - !prepared.isCurrent() || - !this.queue.ownsRequest(next.key, requestSequence) || - !(next.hosted || this.visibility.has(next.key)) - ) { + const settle: PRReadSettlement = (outcome, producer) => { + if (!prepared.isCurrent()) { return outcome } - return this.retry.settleProducerError( - outcome, - next.key, - this.visibility.candidate(next.key, next.candidate), - prepared, - this.queue - ) + const settled = + producer.retained && + outcome.kind === 'upstream-error' && + (next.hosted || this.visibility.has(next.key)) + ? this.retry.settleProducerError( + outcome, + next.key, + next.hosted + ? { ...next.candidate, isSelected: false } + : this.visibility.candidate(next.key, next.candidate), + prepared, + this.queue + ) + : outcome + this.events.observe(next.candidate, settled, aliases) + return settled } const outcome = await (joinedRead ?? (next.hosted ? prepared.readRetained(settle) : prepared.read(settle))) @@ -272,9 +277,6 @@ export class PRRefreshQueueDrainer { plannedRetryAt === undefined ? outcome : this.retry.withErrorSchedule(outcome, plannedRetryAt) - if (!joinedRead && prepared.didRead()) { - this.events.observe(next.candidate, outcome, aliases) - } this.retry.noteManualGate(next.key, broadcastOutcome) if (next.hosted) { if (outcome.kind !== 'upstream-error') { diff --git a/src/main/github/pull-requests/refresh/pr-refresh-watch-jobs.ts b/src/main/github/pull-requests/refresh/pr-refresh-watch-jobs.ts index 2eda3aed927..66d6737f246 100644 --- a/src/main/github/pull-requests/refresh/pr-refresh-watch-jobs.ts +++ b/src/main/github/pull-requests/refresh/pr-refresh-watch-jobs.ts @@ -6,6 +6,7 @@ import { prChecksDone } from '../../../../shared/github/pull-requests/refresh/pr export { prChecksDone } from '../../../../shared/github/pull-requests/refresh/pr-quiet-refresh' import { reviewRefreshIntervalMs } from '../../../../shared/review-refresh-policy' import type { PRRefreshQueue, PRRefreshQueueEntry } from './pr-refresh-queue' +import { isSelectedBudgetedPRRefresh } from './pr-refresh-pacing' import type { PRRefreshRetryState } from './pr-refresh-retry-state' export type PRWatchJob = NonNullable @@ -97,7 +98,7 @@ export class PRWatchJobRunner { noteBudgetedStart(entry: PRRefreshQueueEntry): void { // Selected list reads have their own reserved turn; preserve the watch/list alternation. - if (entry.reason !== 'visible' || entry.candidate.isSelected !== true) { + if (!isSelectedBudgetedPRRefresh(entry)) { this.lastBudgetedStartWasWatch = entry.watch !== undefined } } diff --git a/src/main/ipc/runtime-environments/runtime-environments-subscription-lifecycle.test.ts b/src/main/ipc/runtime-environments/runtime-environments-subscription-lifecycle.test.ts index 7b0fbc78b8c..bf5b52e2f22 100644 --- a/src/main/ipc/runtime-environments/runtime-environments-subscription-lifecycle.test.ts +++ b/src/main/ipc/runtime-environments/runtime-environments-subscription-lifecycle.test.ts @@ -145,14 +145,21 @@ describe('registerRuntimeEnvironmentHandlers', () => { { selector: string method: string - params: { reason?: string } + params: { reason?: string; selected?: boolean } subscriptionId: string }, { subscriptionId: string; requestId: string } >('runtimeEnvironments:subscribe') const start = (subscriptionId: string, reason: string, method = 'github.prForBranch') => - subscribe({ sender }, { selector: 'desk', subscriptionId, method, params: { reason } }) + subscribe( + { sender }, + { selector: 'desk', subscriptionId, method, params: { reason, selected: true } } + ) await start('timer-1', 'swr') + expect(subscribeRemoteRuntimeRequestMock.mock.calls[0][2]).toMatchObject({ + reason: 'swr', + selected: true + }) await start('timer-2', 'swr') const third = start('timer-3', 'swr') const cancelled = expect(third).rejects.toThrow() diff --git a/src/main/runtime/integrations/reviews/hosted-pr-producer-settlement-test-cases.ts b/src/main/runtime/integrations/reviews/hosted-pr-producer-settlement-test-cases.ts index 440ca4f15bd..c2d1ea21fec 100644 --- a/src/main/runtime/integrations/reviews/hosted-pr-producer-settlement-test-cases.ts +++ b/src/main/runtime/integrations/reviews/hosted-pr-producer-settlement-test-cases.ts @@ -65,11 +65,119 @@ export function registerPRProducerSettlementTests({ remove() }) - it.each(['queued', 'immediate', 'foreground'] as const)( + it.each(['queued', 'immediate'] as const)( + 'keeps the %s producer policy when another active consumer joins the same task', + async (origin) => { + const { + addPRRefreshOutcomeObserver, + reportVisiblePRRefreshCandidates, + enqueuePRRefresh, + refreshPRNow + } = await import('../../../github/pull-requests/refresh/pr-refresh-coordinator') + const { setPRRefreshRendererPort } = + await import('../../../github/pull-requests/refresh/pr-refresh-renderer-port') + setPRRefreshRendererPort({ isRendererOpen: () => true, sendToTrustedUI: () => {} }) + const candidate = makeCandidate({ + cacheKey: '/repo::joined-active', + repoId: 'repo', + branch: 'joined-active', + currentHeadOid: 'head-a', + isSelected: true + }) + const provider = Promise.withResolvers() + mocks.resolve.mockReturnValueOnce(provider.promise) + const observe = vi.fn() + const remove = addPRRefreshOutcomeObserver(observe) + try { + mocks.gate.mockResolvedValue(Date.now() + 60_000) + reportVisiblePRRefreshCandidates([candidate], 1, 1) + await vi.advanceTimersByTimeAsync(0) + expect(mocks.resolve).not.toHaveBeenCalled() + mocks.gate.mockResolvedValue(null) + let producer: Promise | undefined + if (origin === 'queued') { + enqueuePRRefresh(candidate, 'active', 80) + } else { + producer = refreshPRNow(candidate, 'active') + } + await vi.advanceTimersByTimeAsync(0) + expect(mocks.resolve).toHaveBeenCalledTimes(1) + const sibling = refreshPRNow(candidate, 'active') + await vi.advanceTimersByTimeAsync(0) + const hosted = read('joined-active') + await vi.advanceTimersByTimeAsync(0) + expect(mocks.resolve).toHaveBeenCalledTimes(1) + provider.reject(new Error('network offline')) + const result = await hosted + expect(result).toMatchObject({ + kind: 'upstream-error', + nextAutoRetryAt: expect.any(Number) + }) + expect(await sibling).toEqual(result) + if (producer) { + expect(await producer).toEqual(result) + } + expect(await read('joined-active')).toEqual(result) + expect(mocks.resolve).toHaveBeenCalledTimes(1) + expect(observe).toHaveBeenCalledTimes(1) + } finally { + remove() + setPRRefreshRendererPort(null) + } + } + ) + + it.each([false, true])( + 'observes one joined success while fencing a genuine successor (userRefresh=%s)', + async (userRefresh) => { + const { addPRRefreshOutcomeObserver, refreshPRNow } = + await import('../../../github/pull-requests/refresh/pr-refresh-coordinator') + const candidate = makeCandidate({ + cacheKey: '/repo::joined-success', + repoId: 'repo', + branch: 'joined-success', + currentHeadOid: 'head-a' + }) + const first = Promise.withResolvers() + const successor = Promise.withResolvers() + mocks.resolve.mockReturnValueOnce(first.promise).mockReturnValueOnce(successor.promise) + const observe = vi.fn() + const remove = addPRRefreshOutcomeObserver(observe) + try { + const producer = refreshPRNow(candidate, 'active') + await vi.advanceTimersByTimeAsync(0) + const sibling = refreshPRNow(candidate, 'active', { userRefresh }) + await vi.advanceTimersByTimeAsync(0) + expect(mocks.resolve).toHaveBeenCalledTimes(userRefresh ? 2 : 1) + const joined = read('joined-success') + await vi.advanceTimersByTimeAsync(0) + first.resolve(found()) + await producer + if (userRefresh) { + expect(observe).not.toHaveBeenCalled() + successor.resolve(found()) + } + const result = await sibling + expect(await joined).toEqual(result) + expect(result).toMatchObject({ kind: 'found' }) + expect(observe).toHaveBeenCalledTimes(1) + } finally { + remove() + } + } + ) + + it.each(['queued', 'immediate', 'headless-immediate', 'foreground'] as const)( 'settles the %s producer before its hosted pending error consumer', async (origin) => { const { addPRRefreshOutcomeObserver, reportVisiblePRRefreshCandidates, refreshPRNow } = await import('../../../github/pull-requests/refresh/pr-refresh-coordinator') + const { setPRRefreshRendererPort } = + await import('../../../github/pull-requests/refresh/pr-refresh-renderer-port') + if (origin === 'queued' || origin === 'immediate') { + setPRRefreshRendererPort({ isRendererOpen: () => true, sendToTrustedUI: () => {} }) + } + const immediate = origin === 'immediate' || origin === 'headless-immediate' const candidate = makeCandidate({ cacheKey: '/repo::origin-error', repoId: 'repo', @@ -86,17 +194,19 @@ export function registerPRProducerSettlementTests({ if (origin === 'foreground') { producer = read('origin-error', 'active') } else { - if (origin === 'immediate') { + if (immediate) { mocks.gate.mockResolvedValue(Date.now() + 60_000) } reportVisiblePRRefreshCandidates([candidate], 1, 1) await vi.advanceTimersByTimeAsync(0) - if (origin === 'immediate') { + if (immediate) { + expect(mocks.resolve).not.toHaveBeenCalled() mocks.gate.mockResolvedValue(null) producer = refreshPRNow(candidate, 'manual') } } await vi.advanceTimersByTimeAsync(0) + expect(mocks.resolve).toHaveBeenCalledTimes(1) const joined = read('origin-error') await vi.advanceTimersByTimeAsync(0) provider.reject(new Error('network offline')) @@ -105,7 +215,7 @@ export function registerPRProducerSettlementTests({ if (!('kind' in result) || result.kind !== 'upstream-error') { throw new Error('expected the actual provider failure') } - if (origin === 'foreground') { + if (origin === 'foreground' || origin === 'headless-immediate') { expect(result.nextAutoRetryAt).toBeUndefined() } else { expect(result.nextAutoRetryAt).toEqual(expect.any(Number)) @@ -118,6 +228,7 @@ export function registerPRProducerSettlementTests({ expect(observe).toHaveBeenCalledTimes(1) } finally { remove() + setPRRefreshRendererPort(null) } } ) diff --git a/src/main/runtime/integrations/reviews/hosted-pr-selected-demand.test.ts b/src/main/runtime/integrations/reviews/hosted-pr-selected-demand.test.ts new file mode 100644 index 00000000000..91869d0c012 --- /dev/null +++ b/src/main/runtime/integrations/reviews/hosted-pr-selected-demand.test.ts @@ -0,0 +1,142 @@ +import { expect, it, vi } from 'vitest' +import { commands, found, mocks, read, rpcRead } from './runtime-hosted-pr-snapshot-test-harness' +import { PrForBranch } from '../../../../shared/rpc-contract/github-pull-request-params' + +it('promotes a shared selected demand without losing other rows or renewing its attempt', async () => { + const branches = ['ordinary-a', 'ordinary-b', 'selected-a', 'selected-b'] + await Promise.all(branches.map((branch) => read(branch, 'active'))) + await vi.advanceTimersByTimeAsync(60_000) + const gate = Date.now() + 60_000 + mocks.gate.mockResolvedValue(gate) + const starts: string[] = [] + mocks.resolve.mockImplementation(async ({ branchName }) => { + starts.push(branchName) + return found() + }) + const demand = (branch: string, selected = false) => + rpcRead({ + repo: 'repo', + branch, + currentHeadOid: 'head-a', + reason: 'swr', + acceptRevalidationOnly: true, + selected + }) + const waiting = branches.map((branch) => demand(branch)) + await vi.advanceTimersByTimeAsync(25_000) + await Promise.all(waiting) + expect(starts).toEqual([]) + await demand('selected-a', true) + await demand('selected-b', true) + await demand('selected-a', false) + mocks.gate.mockResolvedValue(null) + await vi.advanceTimersByTimeAsync(65_000) + expect(starts).toEqual(['selected-a', 'ordinary-a', 'selected-b', 'ordinary-b']) + const completed = starts.length + await vi.advanceTimersByTimeAsync(6 * 60_000) + expect(starts.length).toBe(completed) + + // A selected request stays finite even when its only waiter detaches. + await read('expires', 'active') + await vi.advanceTimersByTimeAsync(60_000) + mocks.gate.mockResolvedValue(Date.now() + 7 * 60_000) + const abort = new AbortController() + const expires = commands.getRepoPRForBranch( + 'repo', + 'expires', + null, + null, + false, + 'head-a', + 'swr', + false, + undefined, + abort.signal, + true, + true + ) + const cancelled = expect(expires).rejects.toThrow('cancelled') + await vi.advanceTimersByTimeAsync(0) + abort.abort() + await cancelled + await vi.advanceTimersByTimeAsync(4 * 60_000) + await demand('expires', true) + const beforeExpiry = starts.length + await vi.advanceTimersByTimeAsync(4 * 60_000) + expect(starts.length).toBe(beforeExpiry) +}) + +it('keeps selected preference optional for old snapshot request parsers', () => { + const params = { repo: 'repo', branch: 'main', reason: 'swr', selected: true } + expect(PrForBranch.parse(params)).toMatchObject({ selected: true }) + expect(PrForBranch.omit({ selected: true }).parse(params)).toEqual({ + repo: 'repo', + branch: 'main', + reason: 'swr' + }) + expect(PrForBranch.parse({ repo: 'repo', branch: 'main' })).toEqual({ + repo: 'repo', + branch: 'main' + }) +}) + +it('keeps a selected join during an awaited quota probe and its later paused admission', async () => { + await read('quiet-selected', 'active') + await vi.advanceTimersByTimeAsync(60_000) + const confirmation = read('quiet-selected') + await vi.advanceTimersByTimeAsync(0) + await confirmation + const branches = ['ordinary-a', 'ordinary-b', 'ordinary-c'] + await Promise.all(branches.map((branch) => read(branch, 'active'))) + await vi.advanceTimersByTimeAsync(300_000) + const quotaProbe = Promise.withResolvers() + const resetAt = Date.now() + 60_000 + mocks.gate.mockResolvedValue(resetAt).mockReturnValueOnce(quotaProbe.promise) + const starts: string[] = [] + mocks.resolve.mockImplementation(async ({ branchName }) => { + starts.push(branchName) + return found() + }) + const demand = (branch: string, selected = false) => + rpcRead({ + repo: 'repo', + branch, + currentHeadOid: 'head-a', + reason: 'swr', + acceptRevalidationOnly: true, + selected + }) + const first = demand('quiet-selected') + await vi.advanceTimersByTimeAsync(0) + expect(mocks.gate).toHaveBeenCalled() + expect(starts).toEqual([]) + const joined = demand('quiet-selected', true) + const ordinary = branches.map((branch) => demand(branch)) + await vi.advanceTimersByTimeAsync(0) + quotaProbe.resolve(resetAt) + await vi.advanceTimersByTimeAsync(25_000) + await Promise.all([first, joined, ...ordinary]) + mocks.gate.mockResolvedValue(null) + await vi.advanceTimersByTimeAsync(65_000) + expect(starts).toEqual(['quiet-selected', 'ordinary-a', 'ordinary-b', 'ordinary-c']) +}) + +it('uses selected demand only for admission priority, preserving the shared retry interval', async () => { + await read('retry-policy', 'active') + await vi.advanceTimersByTimeAsync(60_000) + mocks.resolve.mockRejectedValueOnce(new Error('offline')) + const requestedAt = Date.now() + const response = rpcRead({ + repo: 'repo', + branch: 'retry-policy', + currentHeadOid: 'head-a', + reason: 'swr', + acceptRevalidationOnly: true, + selected: true + }) + await vi.advanceTimersByTimeAsync(0) + expect(await response).toMatchObject({ + kind: 'upstream-error', + nextAutoRetryAt: requestedAt + 120_000 + }) +}) diff --git a/src/main/runtime/integrations/reviews/runtime-hosted-pr-snapshot-test-harness.ts b/src/main/runtime/integrations/reviews/runtime-hosted-pr-snapshot-test-harness.ts index ddcf28e20df..d7a5534d638 100644 --- a/src/main/runtime/integrations/reviews/runtime-hosted-pr-snapshot-test-harness.ts +++ b/src/main/runtime/integrations/reviews/runtime-hosted-pr-snapshot-test-harness.ts @@ -1,9 +1,13 @@ +// oxlint-disable-next-line typescript/triple-slash-reference -- Renderer integration tests also compile this main fixture with its ambient build constants. +/// import { afterEach, beforeEach, vi } from 'vitest' import { makePR } from '../../../github/pull-requests/refresh/pr-refresh-coordinator-test-harness' import type { PRRefreshOutcome } from '../../../../shared/github/pull-requests/refresh/pull-request-refresh-types' import type { RuntimeHostedReviewCommands } from './runtime-hosted-review-commands' +import type { RpcContext } from '../../rpc/core' -export const mocks = vi.hoisted(() => ({ resolve: vi.fn(), account: vi.fn(), gate: vi.fn() })) +const mocks = vi.hoisted(() => ({ resolve: vi.fn(), account: vi.fn(), gate: vi.fn() })) +export { mocks } vi.mock('../../../github/accounts/github-project-account', () => ({ githubPRPolicyProjects: () => ({ source: null, selection: String(mocks.account()) }), resolveGitHubProjectAccount: mocks.account @@ -61,6 +65,8 @@ beforeEach(async () => { mocks.resolve.mockReset().mockImplementation(async () => found()) mocks.account.mockReset().mockReturnValue(null) mocks.gate.mockReset().mockResolvedValue(null) + // Resolve the real handler before fake-clock assertions begin. + await import('../../rpc/methods/source-control/github/github-pull-request-methods') const { RuntimeHostedReviewCommands: Commands } = await import('./runtime-hosted-review-commands') const repo = { id: 'repo', @@ -81,3 +87,19 @@ afterEach(() => { vi.clearAllTimers() vi.useRealTimers() }) + +export async function rpcRead(params: unknown, signal?: AbortSignal) { + const { GITHUB_PULL_REQUEST_METHODS } = + await import('../../rpc/methods/source-control/github/github-pull-request-methods') + const { PrForBranch } = await import('../../../../shared/rpc-contract/github-pull-request-params') + const { repo, branch, ...options } = PrForBranch.parse(params) + const parsed = { repo, branch, ...options } + const runtime = { getRepoPRForBranch: commands.getRepoPRForBranch.bind(commands) } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: this handler only needs the bound real PR command and signal. + const context = { runtime, signal } as unknown as RpcContext + const method = GITHUB_PULL_REQUEST_METHODS.find((entry) => entry.name === 'github.prForBranch') + if (!method) { + throw new Error('PR RPC method is missing') + } + return method.handler(parsed, context) +} diff --git a/src/main/runtime/integrations/reviews/runtime-hosted-review-commands.ts b/src/main/runtime/integrations/reviews/runtime-hosted-review-commands.ts index 0a7a06b8dbf..b3a0c4810c7 100644 --- a/src/main/runtime/integrations/reviews/runtime-hosted-review-commands.ts +++ b/src/main/runtime/integrations/reviews/runtime-hosted-review-commands.ts @@ -87,7 +87,8 @@ export class RuntimeHostedReviewCommands { userRefresh?: boolean, clientKind?: DeviceScope, signal?: AbortSignal, - acceptRevalidationOnly = false + acceptRevalidationOnly = false, + selected = false ): Promise { const deadline = Date.now() + HOSTED_PR_COMPLETION_WAIT_MS const repo = await this.deps.resolveRepo(repoSelector) @@ -134,7 +135,8 @@ export class RuntimeHostedReviewCommands { currentHeadOid: lookupOptions.currentHeadOid ?? null, linkedPRNumber: linkedPRNumber ?? null, fallbackPRNumber: linkedPRNumber == null ? (fallbackPRNumber ?? null) : null, - localGitOptions: lookupOptions.localGitExecOptions + localGitOptions: lookupOptions.localGitExecOptions, + isSelected: reason === 'swr' && userRefresh !== true && selected } const prepared = await preparePRBranchRead( repo.path, diff --git a/src/main/runtime/rpc/methods/source-control/github/github-pull-request-methods.ts b/src/main/runtime/rpc/methods/source-control/github/github-pull-request-methods.ts index a334f135a9e..f54e6e7fce8 100644 --- a/src/main/runtime/rpc/methods/source-control/github/github-pull-request-methods.ts +++ b/src/main/runtime/rpc/methods/source-control/github/github-pull-request-methods.ts @@ -28,7 +28,8 @@ export const GITHUB_PULL_REQUEST_METHODS = [ params.userRefresh, clientKind, signal, - params.acceptRevalidationOnly + params.acceptRevalidationOnly, + params.selected ) }), defineMethod({ diff --git a/src/renderer/src/store/github/github-slice-contract.ts b/src/renderer/src/store/github/github-slice-contract.ts index 0da1373622c..450eaa75f38 100644 --- a/src/renderer/src/store/github/github-slice-contract.ts +++ b/src/renderer/src/store/github/github-slice-contract.ts @@ -71,6 +71,7 @@ export type GitHubSlice = { fallbackPRSource?: GitHubPRFallbackSource | null reason?: GitHubPRRefreshReason userRefresh?: boolean + selected?: boolean onReadOutcome?: (outcome: PRSnapshotReply) => void signal?: AbortSignal } diff --git a/src/renderer/src/store/github/pull-requests/github-pr-hosted-receipt-loop.test.ts b/src/renderer/src/store/github/pull-requests/github-pr-hosted-receipt-loop.test.ts index 7deed4b9501..6401353c688 100644 --- a/src/renderer/src/store/github/pull-requests/github-pr-hosted-receipt-loop.test.ts +++ b/src/renderer/src/store/github/pull-requests/github-pr-hosted-receipt-loop.test.ts @@ -3,7 +3,8 @@ import { commands, fixture, found, - mocks + mocks, + rpcRead } from '../../../../../main/runtime/integrations/reviews/runtime-hosted-pr-snapshot-test-harness' import type { PRRefreshOutcome } from '../../../../../shared/github/pull-requests/refresh/pull-request-refresh-types' @@ -85,22 +86,11 @@ describe('hosted PR receipt delivery', () => { linkedPRNumber: 12, currentHeadOid: 'head-a', reason: 'swr', - acceptRevalidationOnly: true + acceptRevalidationOnly: true, + selected: true }) reasons.push(params.reason!) - const result = await commands.getRepoPRForBranch( - params.repo, - params.branch, - params.linkedPRNumber, - params.fallbackPRNumber, - params.acceptMergedFallbackPR, - params.currentHeadOid, - params.reason, - params.userRefresh, - undefined, - undefined, - params.acceptRevalidationOnly - ) + const result = await rpcRead(args.params) replies.push(result) return { id: 'rpc', ok: true, result } }) @@ -150,3 +140,134 @@ describe('hosted PR receipt delivery', () => { } }) }) + +it('applies data-bearing retry replies to both clients and sends selection after an unselected wait', async () => { + const seed = await commands.getRepoPRForBranch( + 'repo', + 'shared', + 12, + null, + false, + 'head-a', + 'active' + ) + if (!('kind' in seed) || seed.kind !== 'found') { + throw new Error('expected initial PR') + } + const clients: { select: () => void; title: () => string | undefined; dispose: () => void }[] = [] + const selection: boolean[][] = [[], []] + const receipts: unknown[][] = [[], []] + const reasons: string[] = [] + for (let client = 0; client < 2; client += 1) { + // Separate module graphs retain each renderer's own request coalescer. + vi.resetModules() + const { + createTestStore, + makePRRefreshWorktree, + resetRemoteRuntimeMocks, + runtimeEnvironmentCall + } = await import('../github-slice-test-harness') + const { getVisibleHostedReviewRefreshTargets } = + await import('../../slices/hosted-review/refresh/visible-hosted-review-refresh-targets') + const { createVisibleHostedReviewRefreshScheduler } = + await import('../../slices/hosted-review/refresh/visible-hosted-review-refresh-scheduler') + resetRemoteRuntimeMocks() + runtimeEnvironmentCall.mockImplementation(async (args) => { + if (args.method !== 'github.prForBranch') { + throw new Error('unexpected fixture RPC') + } + const { PrForBranch } = + await import('../../../../../shared/rpc-contract/github-pull-request-params') + const params = PrForBranch.parse(args.params) + const requestClient = 'selector' in args && args.selector === 'env-client-0' ? 0 : 1 + reasons.push(params.reason ?? 'missing') + selection[requestClient].push(params.selected === true) + const result = await rpcRead(args.params) + receipts[requestClient].push(result) + return { id: 'rpc', ok: true, result } + }) + const store = createTestStore() + store.setState({ + repos: [ + { + id: 'repo', + path: '/repo', + kind: 'git', + executionHostId: `runtime:env-client-${client}`, + displayName: 'repo', + badgeColor: 'blue', + addedAt: 0 + } + ], + worktreesByRepo: { + repo: [ + makePRRefreshWorktree({ + id: 'shared-worktree', + repoId: 'repo', + branch: 'shared', + head: 'head-a', + linkedPR: 12, + hostId: `runtime:env-client-${client}`, + runtimeOwnerEnvironmentId: `env-client-${client}` + }) + ] + }, + detectedWorktreesByRepo: {}, + activeWorktreeId: null, + activeWorkspaceExecutionHostId: `runtime:env-client-${client}`, + visibleReviewWorktreeIds: ['shared-worktree'], + prCache: { + [`runtime:env-client-${client}::repo::shared`]: { data: seed.pr, fetchedAt: seed.fetchedAt } + } + }) + const scheduler = createVisibleHostedReviewRefreshScheduler() + const update = () => + scheduler.update(getVisibleHostedReviewRefreshTargets(store.getState(), store.getState)) + const unsubscribe = store.subscribe(update) + update() + scheduler.setVisible(true) + clients.push({ + select: () => store.setState({ activeWorktreeId: 'shared-worktree' }), + title: () => + store.getState().prCache[`runtime:env-client-${client}::repo::shared`]?.data?.title, + fetchedAt: () => + store.getState().prCache[`runtime:env-client-${client}::repo::shared`]?.fetchedAt, + dispose: () => { + unsubscribe() + scheduler.dispose() + } + }) + } + try { + // A second client's fresh result can settle in the same millisecond as the first. + fixture.title = 'changed on another client' + await commands.getRepoPRForBranch('repo', 'shared', 12, null, false, 'head-a', 'active') + mocks.gate.mockResolvedValue(181_000) + await vi.advanceTimersByTimeAsync(120_001) + expect(selection).toEqual([[false], [false]]) + for (let client = 0; client < 2; client += 1) { + expect(receipts[client]).toMatchObject([ + { + kind: 'found', + pr: { title: 'changed on another client' }, + revalidation: { retryAfterMs: expect.any(Number) } + } + ]) + expect(clients[client].title()).toBe('changed on another client') + expect(clients[client].fetchedAt()).toBe(1_000) + clients[client].select() + } + mocks.gate.mockResolvedValue(null) + await vi.advanceTimersByTimeAsync(70_000) + expect(selection).toEqual([ + [false, true], + [false, true] + ]) + expect(reasons).toEqual(['swr', 'swr', 'swr', 'swr']) + expect(mocks.resolve).toHaveBeenCalledTimes(3) + } finally { + for (const client of clients) { + client.dispose() + } + } +}) diff --git a/src/renderer/src/store/github/pull-requests/pull-request-execution.ts b/src/renderer/src/store/github/pull-requests/pull-request-execution.ts index 89e738bc0c7..d67200feccf 100644 --- a/src/renderer/src/store/github/pull-requests/pull-request-execution.ts +++ b/src/renderer/src/store/github/pull-requests/pull-request-execution.ts @@ -105,7 +105,14 @@ export function startPullRequestLookup(args: { linkedPRNumber, currentHeadOid: requestHeadOid, ...(options?.reason ? { reason: options.reason } : {}), - ...(options?.reason === 'swr' ? { acceptRevalidationOnly: true } : {}), + ...(options?.reason === 'swr' + ? { + acceptRevalidationOnly: true, + ...(options.userRefresh !== true && options.selected === true + ? { selected: true } + : {}) + } + : {}), ...(options?.userRefresh === true ? { userRefresh: true } : {}), ...(fallbackPRNumber !== null ? { fallbackPRNumber, acceptMergedFallbackPR: fallbackPRSource !== null } diff --git a/src/renderer/src/store/slices/hosted-review/refresh/visible-hosted-review-refresh-targets.ts b/src/renderer/src/store/slices/hosted-review/refresh/visible-hosted-review-refresh-targets.ts index 9359648f59e..bba964017ea 100644 --- a/src/renderer/src/store/slices/hosted-review/refresh/visible-hosted-review-refresh-targets.ts +++ b/src/renderer/src/store/slices/hosted-review/refresh/visible-hosted-review-refresh-targets.ts @@ -200,6 +200,7 @@ export function getVisibleHostedReviewRefreshTargets( force: true, signal, reason: (intent ? intent === 'discovery' : force) ? 'active' : 'swr', + selected, onReadOutcome: (next) => { outcome = next }, diff --git a/src/renderer/src/web/preload-api/web-runtime-pr-subscription-queue.test.ts b/src/renderer/src/web/preload-api/web-runtime-pr-subscription-queue.test.ts index e5b139903af..4291b757a40 100644 --- a/src/renderer/src/web/preload-api/web-runtime-pr-subscription-queue.test.ts +++ b/src/renderer/src/web/preload-api/web-runtime-pr-subscription-queue.test.ts @@ -45,10 +45,17 @@ it('holds web signal-route slots through responses and releases queued and activ const api = createRuntimeEnvironmentsApi() const start = (subscriptionId: string, reason: string, method = 'github.prForBranch') => api.subscribe( - { selector: 'env', subscriptionId, method, params: { reason }, timeoutMs: 30_000 }, + { + selector: 'env', + subscriptionId, + method, + params: { reason, selected: true }, + timeoutMs: 30_000 + }, { onResponse: () => responseOrder.push('response') } ) const first = await start('timer-1', 'swr') + expect(session.client.subscribe.mock.calls[0][1]).toMatchObject({ reason: 'swr', selected: true }) const second = await start('timer-2', 'swr') const third = start('timer-3', 'swr') const cancelled = expect(third).rejects.toThrow() diff --git a/src/shared/rpc-contract/github-pull-request-params.ts b/src/shared/rpc-contract/github-pull-request-params.ts index 89f5e12be6e..458b06782ac 100644 --- a/src/shared/rpc-contract/github-pull-request-params.ts +++ b/src/shared/rpc-contract/github-pull-request-params.ts @@ -21,6 +21,7 @@ export const PrForBranch = RepoSelector.extend({ reason: OptionalPRRefreshReason, userRefresh: z.boolean().optional(), acceptRevalidationOnly: z.boolean().optional(), + selected: z.boolean().optional(), linkedPRNumber: z.number().int().positive().nullable().optional(), fallbackPRNumber: z.number().int().positive().nullable().optional(), acceptMergedFallbackPR: z.boolean().optional(),