fix(github): preserve shared producer settlement and selected demand

Co-authored-by: Jinjing <6427696+AmethystLiang@users.noreply.github.com>
This commit is contained in:
Brennan Benson
2026-11-13 05:16:35 -08:00
co-authored by Jinjing
parent 448111eda3
commit 3f34a8bb47
19 changed files with 540 additions and 90 deletions
@@ -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
@@ -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 ||
@@ -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
@@ -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 },
@@ -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
@@ -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') {
@@ -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<PRRefreshQueueEntry['watch']>
@@ -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
}
}
@@ -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()
@@ -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<PRRefreshOutcome>()
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<PRRefreshOutcome> | 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<PRRefreshOutcome>()
const successor = Promise.withResolvers<PRRefreshOutcome>()
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)
}
}
)
@@ -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<number | null>()
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
})
})
@@ -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.
/// <reference path="../../../../types/build-constants.d.ts" />
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)
}
@@ -87,7 +87,8 @@ export class RuntimeHostedReviewCommands {
userRefresh?: boolean,
clientKind?: DeviceScope,
signal?: AbortSignal,
acceptRevalidationOnly = false
acceptRevalidationOnly = false,
selected = false
): Promise<PRSnapshotReply> {
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,
@@ -28,7 +28,8 @@ export const GITHUB_PULL_REQUEST_METHODS = [
params.userRefresh,
clientKind,
signal,
params.acceptRevalidationOnly
params.acceptRevalidationOnly,
params.selected
)
}),
defineMethod({
@@ -71,6 +71,7 @@ export type GitHubSlice = {
fallbackPRSource?: GitHubPRFallbackSource | null
reason?: GitHubPRRefreshReason
userRefresh?: boolean
selected?: boolean
onReadOutcome?: (outcome: PRSnapshotReply) => void
signal?: AbortSignal
}
@@ -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()
}
}
})
@@ -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 }
@@ -200,6 +200,7 @@ export function getVisibleHostedReviewRefreshTargets(
force: true,
signal,
reason: (intent ? intent === 'discovery' : force) ? 'active' : 'swr',
selected,
onReadOutcome: (next) => {
outcome = next
},
@@ -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()
@@ -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(),