From 6bfb86f7bd6829104326a3bcc7ef9ceae089d9fc Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Fri, 7 Aug 2026 02:46:03 -0400 Subject: [PATCH] fix(repo): harden SSH fork upstream backfill --- ...runtime-fork-upstream-ssh-backfill.test.ts | 200 +++++++++++++++--- src/main/runtime/orca-runtime.ts | 144 ++++++++++--- 2 files changed, 289 insertions(+), 55 deletions(-) diff --git a/src/main/runtime/orca-runtime-fork-upstream-ssh-backfill.test.ts b/src/main/runtime/orca-runtime-fork-upstream-ssh-backfill.test.ts index 2cc6e92b246..cd8af11ecd2 100644 --- a/src/main/runtime/orca-runtime-fork-upstream-ssh-backfill.test.ts +++ b/src/main/runtime/orca-runtime-fork-upstream-ssh-backfill.test.ts @@ -1,14 +1,20 @@ /** * A fork on an SSH host used to keep `upstream === undefined` until someone opened * its settings page — so its GitHub Project rows stayed hidden and its fork badge - * never rendered (#12967). Connect is the first moment the probe can succeed, so - * the backfill runs there: best-effort, sequential, and never awaited by callers. + * never rendered (#12967). Connected is the first published state with a ready + * provider, so the backfill runs there: best-effort, bounded, generation-safe, + * and never awaited by callers. */ -import { describe, expect, it, vi, beforeEach } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { Repo } from '../../shared/types' import type { SshConnectionState } from '../../shared/ssh-types' import { OrcaRuntimeService } from './orca-runtime' import { getRepoUpstream } from '../github/client' +import { + getSshGitProvider, + registerSshGitProvider, + unregisterSshGitProvider +} from '../providers/ssh-git-dispatch' const getRepoUpstreamMock = vi.hoisted(() => vi.fn()) @@ -18,6 +24,7 @@ vi.mock('../github/client', async (importOriginal) => ({ })) const UPSTREAM = { owner: 'stablyai', repo: 'orca' } +const registeredConnectionIds = new Set() function makeRepo(overrides: Partial): Repo { return { @@ -58,6 +65,19 @@ function sshState(targetId: string, status: SshConnectionState['status']): SshCo return { targetId, status, error: null, reconnectAttempt: 0 } } +function notifyConnected(runtime: OrcaRuntimeService, targetId: string): void { + registeredConnectionIds.add(targetId) + if (!getSshGitProvider(targetId)) { + registerSshGitProvider(targetId, {} as never) + } + runtime.notifySshStateChanged(targetId, sshState(targetId, 'connected')) +} + +function notifyDisconnected(runtime: OrcaRuntimeService, targetId: string): void { + unregisterSshGitProvider(targetId) + runtime.notifySshStateChanged(targetId, sshState(targetId, 'disconnected')) +} + // Why: the backfill is deliberately fire-and-forget, so tests drain the queue // rather than awaiting a handle the production caller never has. async function drainBackfill(): Promise { @@ -71,12 +91,19 @@ beforeEach(() => { getRepoUpstreamMock.mockResolvedValue(null) }) +afterEach(() => { + for (const connectionId of registeredConnectionIds) { + unregisterSshGitProvider(connectionId) + } + registeredConnectionIds.clear() +}) + describe('fork upstream backfill for SSH repos', () => { it('resolves the upstream when the connection comes up', async () => { getRepoUpstreamMock.mockResolvedValue(UPSTREAM) const { runtime, repos } = createRuntime([makeRepo({ id: 'ssh-fork', connectionId: 'ssh-1' })]) - runtime.notifySshStateChanged('ssh-1', sshState('ssh-1', 'connected')) + notifyConnected(runtime, 'ssh-1') await drainBackfill() expect(getRepoUpstream).toHaveBeenCalledWith('/srv/orca', 'ssh-1') @@ -99,7 +126,7 @@ describe('fork upstream backfill for SSH repos', () => { }) ]) - runtime.notifySshStateChanged('ssh-1', sshState('ssh-1', 'connected')) + notifyConnected(runtime, 'ssh-1') await drainBackfill() expect(repos[0].repoIcon).toMatchObject({ type: 'image', source: 'github' }) @@ -115,63 +142,186 @@ describe('fork upstream backfill for SSH repos', () => { makeRepo({ id: 'resolved', path: '/srv/c', connectionId: 'ssh-1', upstream: null }) ]) - runtime.notifySshStateChanged('ssh-1', sshState('ssh-1', 'connected')) + notifyConnected(runtime, 'ssh-1') await drainBackfill() expect(getRepoUpstream).not.toHaveBeenCalled() }) - it('probes each repo once, not once per reconnect', async () => { - const { runtime } = createRuntime([makeRepo({ id: 'ssh-fork', connectionId: 'ssh-1' })]) + it('leaves best-effort null unresolved without retrying every reconnect', async () => { + const { runtime, repos, updateRepo } = createRuntime([ + makeRepo({ id: 'ssh-fork', connectionId: 'ssh-1' }) + ]) - runtime.notifySshStateChanged('ssh-1', sshState('ssh-1', 'connected')) + notifyConnected(runtime, 'ssh-1') runtime.notifySshStateChanged('ssh-1', sshState('ssh-1', 'connected')) await drainBackfill() - runtime.notifySshStateChanged('ssh-1', sshState('ssh-1', 'disconnected')) - runtime.notifySshStateChanged('ssh-1', sshState('ssh-1', 'connected')) + notifyDisconnected(runtime, 'ssh-1') + notifyConnected(runtime, 'ssh-1') await drainBackfill() expect(getRepoUpstream).toHaveBeenCalledTimes(1) + expect(repos[0].upstream).toBeUndefined() + expect(updateRepo).not.toHaveBeenCalled() }) it('retries on the next connect when the probe threw', async () => { getRepoUpstreamMock.mockRejectedValueOnce(new Error('ssh channel closed')) const { runtime, repos } = createRuntime([makeRepo({ id: 'ssh-fork', connectionId: 'ssh-1' })]) - runtime.notifySshStateChanged('ssh-1', sshState('ssh-1', 'connected')) + notifyConnected(runtime, 'ssh-1') await drainBackfill() expect(repos[0].upstream).toBeUndefined() getRepoUpstreamMock.mockResolvedValue(UPSTREAM) - runtime.notifySshStateChanged('ssh-1', sshState('ssh-1', 'connected')) + notifyDisconnected(runtime, 'ssh-1') + notifyConnected(runtime, 'ssh-1') await drainBackfill() expect(repos[0].upstream).toEqual(UPSTREAM) }) - it('never overlaps probes when several connections come up at once', async () => { + it('retries a newer provider generation when reconnect wins the failure race', async () => { + let rejectProbe: ((error: Error) => void) | undefined + getRepoUpstreamMock + .mockImplementationOnce( + () => + new Promise((_, reject) => { + rejectProbe = reject + }) + ) + .mockResolvedValueOnce(UPSTREAM) + const { runtime, repos } = createRuntime([makeRepo({ id: 'ssh-fork', connectionId: 'ssh-1' })]) + + notifyConnected(runtime, 'ssh-1') + await drainBackfill() + notifyDisconnected(runtime, 'ssh-1') + notifyConnected(runtime, 'ssh-1') + rejectProbe?.(new Error('old provider disconnected')) + await drainBackfill() + + expect(getRepoUpstream).toHaveBeenCalledTimes(2) + expect(repos[0].upstream).toEqual(UPSTREAM) + }) + + it('discards a stale null returned after the provider changed', async () => { + let resolveProbe: ((value: null) => void) | undefined + getRepoUpstreamMock + .mockImplementationOnce( + () => + new Promise((resolve) => { + resolveProbe = resolve + }) + ) + .mockResolvedValueOnce(UPSTREAM) + const { runtime, repos } = createRuntime([makeRepo({ id: 'ssh-fork', connectionId: 'ssh-1' })]) + + notifyConnected(runtime, 'ssh-1') + await drainBackfill() + notifyDisconnected(runtime, 'ssh-1') + notifyConnected(runtime, 'ssh-1') + resolveProbe?.(null) + await drainBackfill() + + expect(getRepoUpstream).toHaveBeenCalledTimes(2) + expect(repos[0].upstream).toEqual(UPSTREAM) + }) + + it('lets a healthy host pass a stalled host while bounding concurrent probes', async () => { let inFlight = 0 let maxInFlight = 0 - getRepoUpstreamMock.mockImplementation(async () => { + let releaseA: (() => void) | undefined + let releaseB: (() => void) | undefined + getRepoUpstreamMock.mockImplementation(async (path) => { inFlight += 1 maxInFlight = Math.max(maxInFlight, inFlight) - await Promise.resolve() - inFlight -= 1 - return null + try { + if (path === '/srv/a') { + await new Promise((resolve) => { + releaseA = resolve + }) + } + if (path === '/srv/b') { + await new Promise((resolve) => { + releaseB = resolve + }) + } + return UPSTREAM + } finally { + inFlight -= 1 + } }) - const { runtime } = createRuntime([ + const { runtime, repos } = createRuntime([ makeRepo({ id: 'a', path: '/srv/a', connectionId: 'ssh-1' }), makeRepo({ id: 'b', path: '/srv/b', connectionId: 'ssh-2' }), makeRepo({ id: 'c', path: '/srv/c', connectionId: 'ssh-3' }) ]) - runtime.notifySshStateChanged('ssh-1', sshState('ssh-1', 'connected')) - runtime.notifySshStateChanged('ssh-2', sshState('ssh-2', 'connected')) - runtime.notifySshStateChanged('ssh-3', sshState('ssh-3', 'connected')) + notifyConnected(runtime, 'ssh-1') + notifyConnected(runtime, 'ssh-2') + notifyConnected(runtime, 'ssh-3') await drainBackfill() - expect(getRepoUpstream).toHaveBeenCalledTimes(3) - expect(maxInFlight).toBe(1) + expect(getRepoUpstream).toHaveBeenCalledTimes(2) + expect(getRepoUpstream).not.toHaveBeenCalledWith('/srv/c', 'ssh-3') + expect(maxInFlight).toBe(2) + + releaseB?.() + await drainBackfill() + + expect(repos[1].upstream).toEqual(UPSTREAM) + expect(repos[2].upstream).toEqual(UPSTREAM) + expect(repos[0].upstream).toBeUndefined() + expect(maxInFlight).toBe(2) + + releaseA?.() + await drainBackfill() + + expect(repos[0].upstream).toEqual(UPSTREAM) + }) + + it('does not persist a stale null when a queued provider disconnects', async () => { + let releaseA: (() => void) | undefined + let releaseB: (() => void) | undefined + getRepoUpstreamMock.mockImplementation((path) => { + if (path === '/srv/a') { + return new Promise((resolve) => { + releaseA = () => resolve(null) + }) + } + if (path === '/srv/b') { + return new Promise((resolve) => { + releaseB = () => resolve(null) + }) + } + return Promise.resolve(UPSTREAM) + }) + const { runtime, repos } = createRuntime([ + makeRepo({ id: 'a', path: '/srv/a', connectionId: 'ssh-1' }), + makeRepo({ id: 'b', path: '/srv/b', connectionId: 'ssh-2' }), + makeRepo({ id: 'c', path: '/srv/c', connectionId: 'ssh-3' }) + ]) + + notifyConnected(runtime, 'ssh-1') + notifyConnected(runtime, 'ssh-2') + await drainBackfill() + expect(getRepoUpstream).toHaveBeenCalledTimes(2) + + notifyConnected(runtime, 'ssh-3') + notifyDisconnected(runtime, 'ssh-3') + releaseA?.() + await drainBackfill() + + expect(repos[2].upstream).toBeUndefined() + expect(getRepoUpstream).not.toHaveBeenCalledWith('/srv/c', 'ssh-3') + + notifyConnected(runtime, 'ssh-3') + await drainBackfill() + + expect(repos[2].upstream).toEqual(UPSTREAM) + + releaseB?.() + await drainBackfill() }) it('returns from the connect notification without waiting on the probe', async () => { @@ -184,7 +334,7 @@ describe('fork upstream backfill for SSH repos', () => { ) const { runtime, repos } = createRuntime([makeRepo({ id: 'ssh-fork', connectionId: 'ssh-1' })]) - runtime.notifySshStateChanged('ssh-1', sshState('ssh-1', 'connected')) + notifyConnected(runtime, 'ssh-1') expect(getRepoUpstream).not.toHaveBeenCalled() await drainBackfill() diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 2befd211dc3..c1849dcab4e 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -2787,6 +2787,13 @@ function getSetupRunnerCommandPlatformForLaunch( return getSetupRunnerCommandPlatformForPath(setup?.runnerScriptPath ?? '', fallbackPlatform) } +type ForkUpstreamSshScope = { + connectionId: string + providerGeneration: number +} + +const FORK_UPSTREAM_BACKFILL_CONCURRENCY = 2 + export class OrcaRuntimeService { private readonly runtimeId = randomUUID() private readonly startedAt = Date.now() @@ -2920,10 +2927,11 @@ export class OrcaRuntimeService { >() private worktreeLifecycleListeners = new Set<(event: RuntimeWorktreeLifecycleEvent) => void>() private forkBackfillStarted = false - // Why: one chain keeps every backfill pass sequential, so a burst of SSH - // connects can never fan out `gh repo view` calls past the shared semaphore. - private forkUpstreamBackfillQueue: Promise = Promise.resolve() + private forkUpstreamBackfillActivePasses = 0 + private forkUpstreamBackfillWaiters: (() => void)[] = [] private forkUpstreamBackfilledConnectionIds = new Set() + private forkUpstreamBackfillReadyGenerationByConnectionId = new Map() + private forkUpstreamBackfillRunningConnectionIds = new Set() private agentBrowserBridge: AgentBrowserBridge | null = null private offscreenBrowserBackend: BrowserBackend | null = null private emulatorBridge: EmulatorBridge | null = null @@ -5300,6 +5308,7 @@ export class OrcaRuntimeService { this.invalidateSshWorktreeScanCache(targetId) if (state.status !== 'connected') { this.cancelLegacyWorkerTerminalRecoveryRetry(`ssh:${targetId}`) + this.forkUpstreamBackfillReadyGenerationByConnectionId.delete(targetId) } else { this.backfillForkUpstreamsForConnection(targetId) } @@ -19400,7 +19409,9 @@ export class OrcaRuntimeService { // startup for repos whose host is already reachable; SSH repos wait for their // connection instead (their host may not be up yet). private backfillForkUpstreams(): Promise { - return this.queueForkUpstreamBackfill((repo) => !repo.connectionId) + return this.queueForkUpstreamBackfill(() => + this.runForkUpstreamBackfill((repo) => !repo.connectionId) + ) } // Why: an SSH fork used to stay `upstream === undefined` until someone opened @@ -19408,36 +19419,99 @@ export class OrcaRuntimeService { // no visible cause (#12967). Connect is the first moment the probe can succeed. // Fire-and-forget so nothing on the connection path waits on the network. private backfillForkUpstreamsForConnection(connectionId: string): void { - if (this.forkUpstreamBackfilledConnectionIds.has(connectionId)) { + if ( + this.forkUpstreamBackfilledConnectionIds.has(connectionId) || + !getSshGitProvider(connectionId) + ) { return } - this.forkUpstreamBackfilledConnectionIds.add(connectionId) - void this.queueForkUpstreamBackfill((repo) => repo.connectionId === connectionId) - .then((complete) => { - // Why: a pass interrupted by a probe failure left `upstream` unset, so let - // the next connect retry rather than stranding it until the app restarts. - if (!complete) { - this.forkUpstreamBackfilledConnectionIds.delete(connectionId) + this.forkUpstreamBackfillReadyGenerationByConnectionId.set( + connectionId, + getSshGitProviderGeneration(connectionId) + ) + if (this.forkUpstreamBackfillRunningConnectionIds.has(connectionId)) { + return + } + this.forkUpstreamBackfillRunningConnectionIds.add(connectionId) + void this.drainForkUpstreamBackfillForConnection(connectionId) + .catch(() => undefined) + .finally(() => { + this.forkUpstreamBackfillRunningConnectionIds.delete(connectionId) + if ( + !this.forkUpstreamBackfilledConnectionIds.has(connectionId) && + this.forkUpstreamBackfillReadyGenerationByConnectionId.has(connectionId) + ) { + this.backfillForkUpstreamsForConnection(connectionId) } }) - .catch(() => { - this.forkUpstreamBackfilledConnectionIds.delete(connectionId) - }) } - // Why: chained, never concurrent — `getRepoUpstream` shells out to `gh` behind a - // process-global semaphore, so passes queue instead of fanning out. - private queueForkUpstreamBackfill(selects: (repo: Repo) => boolean): Promise { - const pass = this.forkUpstreamBackfillQueue.then(() => this.runForkUpstreamBackfill(selects)) - this.forkUpstreamBackfillQueue = pass.then( - () => undefined, - () => undefined + // Why: bound migration subprocesses without letting one stalled host block every other host. + private async queueForkUpstreamBackfill(run: () => Promise): Promise { + await this.acquireForkUpstreamBackfillPermit() + try { + return await run() + } finally { + this.releaseForkUpstreamBackfillPermit() + } + } + + private acquireForkUpstreamBackfillPermit(): Promise { + if (this.forkUpstreamBackfillActivePasses < FORK_UPSTREAM_BACKFILL_CONCURRENCY) { + this.forkUpstreamBackfillActivePasses += 1 + return Promise.resolve() + } + return new Promise((resolve) => this.forkUpstreamBackfillWaiters.push(resolve)) + } + + private releaseForkUpstreamBackfillPermit(): void { + const next = this.forkUpstreamBackfillWaiters.shift() + if (next) { + next() + return + } + this.forkUpstreamBackfillActivePasses -= 1 + } + + private async drainForkUpstreamBackfillForConnection(connectionId: string): Promise { + while (!this.forkUpstreamBackfilledConnectionIds.has(connectionId)) { + const providerGeneration = + this.forkUpstreamBackfillReadyGenerationByConnectionId.get(connectionId) + if (providerGeneration === undefined) { + return + } + const scope = { connectionId, providerGeneration } + const complete = await this.queueForkUpstreamBackfill(() => + this.runForkUpstreamBackfill((repo) => repo.connectionId === connectionId, scope) + ) + if (complete) { + this.forkUpstreamBackfilledConnectionIds.add(connectionId) + this.forkUpstreamBackfillReadyGenerationByConnectionId.delete(connectionId) + return + } + if ( + this.forkUpstreamBackfillReadyGenerationByConnectionId.get(connectionId) === + providerGeneration + ) { + this.forkUpstreamBackfillReadyGenerationByConnectionId.delete(connectionId) + } + } + } + + private isForkUpstreamBackfillScopeCurrent(scope: ForkUpstreamSshScope): boolean { + return ( + this.forkUpstreamBackfillReadyGenerationByConnectionId.get(scope.connectionId) === + scope.providerGeneration && + getSshGitProviderGeneration(scope.connectionId) === scope.providerGeneration && + getSshGitProvider(scope.connectionId) !== undefined ) - return pass } /** Resolves false when any probe failed, so the caller can retry that scope later. */ - private async runForkUpstreamBackfill(selects: (repo: Repo) => boolean): Promise { + private async runForkUpstreamBackfill( + selects: (repo: Repo) => boolean, + sshScope?: ForkUpstreamSshScope + ): Promise { let complete = true try { const store = this.requireStore() @@ -19446,16 +19520,26 @@ export class OrcaRuntimeService { if (repo.upstream !== undefined || repo.kind === 'folder' || !selects(repo)) { continue } + if (sshScope && !this.isForkUpstreamBackfillScopeCurrent(sshScope)) { + complete = false + break + } let upstream: GitHubOwnerRepo | null try { - // Why: same resolution the settings page uses, so backfilled and - // lazily-resolved upstreams can never disagree for the same repo. - const options = this.getHostedReviewExecutionOptions(repo) - upstream = await (options - ? getRepoUpstream(repo.path, repo.connectionId ?? null, options) - : getRepoUpstream(repo.path, repo.connectionId ?? null)) + upstream = await getRepoUpstream(repo.path, repo.connectionId ?? null) } catch { complete = false + if (sshScope && !this.isForkUpstreamBackfillScopeCurrent(sshScope)) { + break + } + continue + } + if (sshScope && !this.isForkUpstreamBackfillScopeCurrent(sshScope)) { + complete = false + break + } + // Why: SSH null may hide a transient Git/gh failure, so only persist a positive result. + if (sshScope && !upstream) { continue } const updates: Partial = { upstream: upstream ?? null }