fix(repo): harden SSH fork upstream backfill

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<string>()
 
 function makeRepo(overrides: Partial<Repo>): 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<void> {
@@ -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<null>((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<void>((resolve) => {
+            releaseA = resolve
+          })
+        }
+        if (path === '/srv/b') {
+          await new Promise<void>((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<null>((resolve) => {
+          releaseA = () => resolve(null)
+        })
+      }
+      if (path === '/srv/b') {
+        return new Promise<null>((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()

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<void> = Promise.resolve()
+  private forkUpstreamBackfillActivePasses = 0
+  private forkUpstreamBackfillWaiters: (() => void)[] = []
   private forkUpstreamBackfilledConnectionIds = new Set<string>()
+  private forkUpstreamBackfillReadyGenerationByConnectionId = new Map<string, number>()
+  private forkUpstreamBackfillRunningConnectionIds = new Set<string>()
   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<boolean> {
-    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<boolean> {
-    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<boolean>): Promise<boolean> {
+    await this.acquireForkUpstreamBackfillPermit()
+    try {
+      return await run()
+    } finally {
+      this.releaseForkUpstreamBackfillPermit()
+    }
+  }
+
+  private acquireForkUpstreamBackfillPermit(): Promise<void> {
+    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<void> {
+    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<boolean> {
+  private async runForkUpstreamBackfill(
+    selects: (repo: Repo) => boolean,
+    sshScope?: ForkUpstreamSshScope
+  ): Promise<boolean> {
     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<Repo> = { upstream: upstream ?? null }