diff --git a/src/main/ipc/worktree-base-directory-poller.ts b/src/main/ipc/worktree-base-directory-poller.ts index 31a068146c8..76ae84f228e 100644 --- a/src/main/ipc/worktree-base-directory-poller.ts +++ b/src/main/ipc/worktree-base-directory-poller.ts @@ -57,6 +57,8 @@ export function createWorktreePollerWindowVisibility( export type WorktreeBasePollerOptions = { pollIntervalMs?: number + gitCommonIdlePollIntervalMs?: number + gitCommonBackstopIntervalMs?: number platform?: NodeJS.Platform visibility?: WorktreePollerWindowVisibility /** Test hook: called whenever a full snapshot scan runs (vs. a gated skip). */ @@ -72,6 +74,8 @@ export type WorktreeBasePollerOptions = { // calls per tick. 2s is fast enough for external `git worktree add/remove`; // Orca's own worktree operations notify the renderer directly. export const WORKTREE_BASE_POLL_INTERVAL_MS = 2_000 +export const WORKTREE_GIT_COMMON_IDLE_POLL_INTERVAL_MS = 10_000 +export const WORKTREE_GIT_COMMON_BACKSTOP_INTERVAL_MS = 30_000 // Why: the mtime gate is an optimization, not a correctness boundary — some // filesystems have coarse dir timestamps, and pending `.git` markers expire. @@ -343,13 +347,18 @@ export async function startWorktreeBaseDirectoryPoller( const platform = options.platform ?? process.platform const visibility = options.visibility ?? alwaysVisible if (target.kind === 'git-common') { + const intervalScale = pollIntervalMs / WORKTREE_BASE_POLL_INTERVAL_MS return startGitCommonWatch( target, onEvents, pollIntervalMs, platform, visibility, - options.onFullScan + options.onFullScan, + options.gitCommonIdlePollIntervalMs ?? + WORKTREE_GIT_COMMON_IDLE_POLL_INTERVAL_MS * intervalScale, + options.gitCommonBackstopIntervalMs ?? + WORKTREE_GIT_COMMON_BACKSTOP_INTERVAL_MS * intervalScale ) } return startBasePoller(target, getRepos, onEvents, pollIntervalMs, visibility, options.onFullScan) diff --git a/src/main/ipc/worktree-git-common-poll-cadence.ts b/src/main/ipc/worktree-git-common-poll-cadence.ts new file mode 100644 index 00000000000..9e0d326af33 --- /dev/null +++ b/src/main/ipc/worktree-git-common-poll-cadence.ts @@ -0,0 +1,159 @@ +import type { + WorktreeBaseSubscription, + WorktreePollerWindowVisibility +} from './worktree-base-directory-poller' + +const UNCHANGED_POLLS_BEFORE_IDLE = 3 + +export type GitCommonPollingCadence = { + activeIntervalMs: number + idleIntervalMs: number + indexBackstopIntervalMs?: number +} + +export type AdaptiveGitCommonPollSubscription = WorktreeBaseSubscription & { + resetCadence: () => void +} + +export async function tryTakeGitCommonPollBaseline( + takeSnapshot: () => Promise +): Promise { + try { + return await takeSnapshot() + } catch { + return null + } +} + +export function startAdaptiveGitCommonPoller(args: { + cadence: GitCommonPollingCadence + visibility: WorktreePollerWindowVisibility + poll: (forceFullScan: boolean) => Promise<{ changed: boolean }> +}): AdaptiveGitCommonPollSubscription { + const { cadence, visibility, poll } = args + let disposed = false + let ticking = false + let unchangedPolls = 0 + let timer: ReturnType | null = null + let pendingForceFullScan = false + let forceRetryNeeded = false + let activityGeneration = 0 + let lastActivityAt: number | null = null + let indexBackstopAt = + cadence.indexBackstopIntervalMs === undefined + ? Number.POSITIVE_INFINITY + : performance.now() + cadence.indexBackstopIntervalMs + + const clearTimer = (): void => { + if (timer) { + clearTimeout(timer) + timer = null + } + } + + const armTimer = (delayMs: number): void => { + clearTimer() + timer = setTimeout(() => { + timer = null + void tick() + }, delayMs) + timer.unref?.() + } + + const resetCadence = (): void => { + if (disposed) { + return + } + unchangedPolls = 0 + activityGeneration++ + lastActivityAt = performance.now() + clearTimer() + if (!ticking && visibility.isWindowVisible()) { + armTimer(cadence.activeIntervalMs) + } + } + + const tick = async (): Promise => { + clearTimer() + if (disposed || !visibility.isWindowVisible() || ticking) { + return + } + ticking = true + const startedAt = performance.now() + const startingActivityGeneration = activityGeneration + const requestedForceFullScan = pendingForceFullScan + pendingForceFullScan = false + const backstopDue = startedAt >= indexBackstopAt + const shouldForceFullScan = requestedForceFullScan || forceRetryNeeded || backstopDue + let succeeded = false + let changed = false + try { + const result = await poll(shouldForceFullScan) + succeeded = true + changed = result.changed + if (shouldForceFullScan && cadence.indexBackstopIntervalMs !== undefined) { + indexBackstopAt = performance.now() + cadence.indexBackstopIntervalMs + } + forceRetryNeeded = false + } catch { + forceRetryNeeded ||= shouldForceFullScan + } finally { + ticking = false + } + if (disposed) { + return + } + + const activityDuringPoll = activityGeneration !== startingActivityGeneration + if (!activityDuringPoll) { + unchangedPolls = !succeeded || changed ? 0 : unchangedPolls + 1 + } + if (pendingForceFullScan) { + void tick() + return + } + if (!visibility.isWindowVisible()) { + return + } + if (forceRetryNeeded) { + armTimer(cadence.activeIntervalMs) + return + } + const now = performance.now() + if (activityDuringPoll && lastActivityAt !== null) { + armTimer(Math.max(0, cadence.activeIntervalMs - (now - lastActivityAt))) + return + } + const intervalMs = + unchangedPolls >= UNCHANGED_POLLS_BEFORE_IDLE + ? cadence.idleIntervalMs + : cadence.activeIntervalMs + const cadenceDelay = Math.max(0, intervalMs - (now - startedAt)) + const backstopDelay = Math.max(0, indexBackstopAt - now) + armTimer(Math.min(cadenceDelay, backstopDelay)) + } + + const unsubscribeVisibility = visibility.onWindowBecameVisible(() => { + if (disposed) { + return + } + unchangedPolls = 0 + activityGeneration++ + lastActivityAt = performance.now() + pendingForceFullScan = true + clearTimer() + if (!ticking) { + void tick() + } + }) + + armTimer(cadence.activeIntervalMs) + return { + resetCadence, + unsubscribe: async () => { + disposed = true + clearTimer() + unsubscribeVisibility() + } + } +} diff --git a/src/main/ipc/worktree-git-common-polling-cadence.test.ts b/src/main/ipc/worktree-git-common-polling-cadence.test.ts new file mode 100644 index 00000000000..2c9829b3300 --- /dev/null +++ b/src/main/ipc/worktree-git-common-polling-cadence.test.ts @@ -0,0 +1,176 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { WorktreePollerWindowVisibility } from './worktree-base-directory-poller' +import { startAdaptiveGitCommonPoller } from './worktree-git-common-poll-cadence' + +function createVisibilityHarness(): { + source: WorktreePollerWindowVisibility + hide: () => void + show: () => void +} { + let visible = true + let listener: (() => void) | null = null + return { + source: { + isWindowVisible: () => visible, + onWindowBecameVisible: (nextListener) => { + listener = nextListener + return () => { + if (listener === nextListener) { + listener = null + } + } + } + }, + hide: () => { + visible = false + }, + show: () => { + visible = true + listener?.() + } + } +} + +function controlledPromise(): { + promise: Promise + resolve: (value: T) => void +} { + let resolve!: (value: T) => void + const promise = new Promise((nextResolve) => { + resolve = nextResolve + }) + return { promise, resolve } +} + +afterEach(() => { + vi.useRealTimers() +}) + +describe('adaptive git-common poll cadence', () => { + it('backs off after three unchanged polls and returns fast after a change', async () => { + vi.useFakeTimers() + const changes = [false, false, false, true, false] + const poll = vi.fn(async () => ({ changed: changes.shift() ?? false })) + const subscription = startAdaptiveGitCommonPoller({ + cadence: { activeIntervalMs: 100, idleIntervalMs: 500 }, + visibility: createVisibilityHarness().source, + poll + }) + + await vi.advanceTimersByTimeAsync(300) + expect(poll).toHaveBeenCalledTimes(3) + await vi.advanceTimersByTimeAsync(499) + expect(poll).toHaveBeenCalledTimes(3) + await vi.advanceTimersByTimeAsync(1) + expect(poll).toHaveBeenCalledTimes(4) + await vi.advanceTimersByTimeAsync(99) + expect(poll).toHaveBeenCalledTimes(4) + await vi.advanceTimersByTimeAsync(1) + expect(poll).toHaveBeenCalledTimes(5) + + await subscription.unsubscribe() + }) + + it('caps an idle delay at the monotonic index backstop deadline', async () => { + vi.useFakeTimers() + const forced: boolean[] = [] + const subscription = startAdaptiveGitCommonPoller({ + cadence: { + activeIntervalMs: 100, + idleIntervalMs: 500, + indexBackstopIntervalMs: 700 + }, + visibility: createVisibilityHarness().source, + poll: async (forceFullScan) => { + forced.push(forceFullScan) + return { changed: false } + } + }) + + await vi.advanceTimersByTimeAsync(699) + expect(forced).toEqual([false, false, false]) + await vi.advanceTimersByTimeAsync(1) + expect(forced).toEqual([false, false, false, true]) + + await subscription.unsubscribe() + }) + + it('retries a failed overdue scan at the nonzero active cadence', async () => { + vi.useFakeTimers() + const forced: boolean[] = [] + let failFirstForcedScan = true + const subscription = startAdaptiveGitCommonPoller({ + cadence: { + activeIntervalMs: 100, + idleIntervalMs: 500, + indexBackstopIntervalMs: 300 + }, + visibility: createVisibilityHarness().source, + poll: async (forceFullScan) => { + forced.push(forceFullScan) + if (forceFullScan && failFirstForcedScan) { + failFirstForcedScan = false + throw new Error('transient stat failure') + } + return { changed: false } + } + }) + + await vi.advanceTimersByTimeAsync(300) + expect(forced).toEqual([false, false, true]) + await vi.advanceTimersByTimeAsync(99) + expect(forced).toEqual([false, false, true]) + await vi.advanceTimersByTimeAsync(1) + expect(forced).toEqual([false, false, true, true]) + + await subscription.unsubscribe() + }) + + it('forces every visibility resume and coalesces one behind an in-flight poll', async () => { + vi.useFakeTimers() + const visibility = createVisibilityHarness() + const first = controlledPromise<{ changed: boolean }>() + const forced: boolean[] = [] + const poll = vi.fn((forceFullScan: boolean) => { + forced.push(forceFullScan) + return forced.length === 1 ? first.promise : Promise.resolve({ changed: false }) + }) + const subscription = startAdaptiveGitCommonPoller({ + cadence: { activeIntervalMs: 100, idleIntervalMs: 500 }, + visibility: visibility.source, + poll + }) + + await vi.advanceTimersByTimeAsync(100) + expect(forced).toEqual([false]) + visibility.show() + expect(forced).toEqual([false]) + + first.resolve({ changed: false }) + await vi.advanceTimersByTimeAsync(0) + expect(forced).toEqual([false, true]) + + await subscription.unsubscribe() + }) + + it('replaces an outstanding idle timer when native activity resets cadence', async () => { + vi.useFakeTimers() + const poll = vi.fn(async () => ({ changed: false })) + const subscription = startAdaptiveGitCommonPoller({ + cadence: { activeIntervalMs: 100, idleIntervalMs: 500 }, + visibility: createVisibilityHarness().source, + poll + }) + + await vi.advanceTimersByTimeAsync(300) + expect(poll).toHaveBeenCalledTimes(3) + await vi.advanceTimersByTimeAsync(50) + subscription.resetCadence() + await vi.advanceTimersByTimeAsync(99) + expect(poll).toHaveBeenCalledTimes(3) + await vi.advanceTimersByTimeAsync(1) + expect(poll).toHaveBeenCalledTimes(4) + + await subscription.unsubscribe() + }) +}) diff --git a/src/main/ipc/worktree-git-common-polling.ts b/src/main/ipc/worktree-git-common-polling.ts index ed0043b3c5b..9cad411b1f5 100644 --- a/src/main/ipc/worktree-git-common-polling.ts +++ b/src/main/ipc/worktree-git-common-polling.ts @@ -5,6 +5,11 @@ import type { WorktreeBaseSubscription, WorktreePollerWindowVisibility } from './worktree-base-directory-poller' +import { + startAdaptiveGitCommonPoller, + tryTakeGitCommonPollBaseline, + type GitCommonPollingCadence +} from './worktree-git-common-poll-cadence' // Shared with the darwin primary-metadata poll so platforms cannot drift. // `logs/HEAD` catches head moves; `config.worktree` carries the sparse flag. @@ -18,23 +23,27 @@ export const PRIMARY_CHECKOUT_METADATA_FILES = [ const LINKED_WORKTREE_STRUCTURAL_METADATA_FILES = ['HEAD', 'gitdir', 'locked', 'config.worktree'] const LINKED_WORKTREE_INDEX_FILE = 'index' const LINKED_WORKTREE_HEAD_LOG_FILE = join('logs', 'HEAD') -// Why: the entry-dir signature gate can miss same-granule index rewrites on -// coarse-mtime filesystems; a periodic ungated re-stat bounds that miss the -// same way the base poller's backstop rescan does. -const INDEX_BACKSTOP_TICKS = 15 function statSignature(s: { mtimeMs: number; ctimeMs: number; ino: number }): string { return `${s.mtimeMs}:${s.ctimeMs}:${s.ino}` } +function isMissingPathError(error: unknown): boolean { + const code = (error as NodeJS.ErrnoException).code + return code === 'ENOENT' || code === 'ENOTDIR' +} + async function dirSignature(path: string): Promise { try { // Why: keep `size` — on a coarse-timestamp filesystem a same-granule directory // allocation change would otherwise slip the readdir gate to the backstop. const s = await stat(path) return `${statSignature(s)}:${s.size}` - } catch { - return 'missing' + } catch (error) { + if (isMissingPathError(error)) { + return 'missing' + } + throw error } } @@ -42,8 +51,11 @@ async function fileSignature(path: string): Promise { try { const s = await stat(path) return s.isFile() ? `${statSignature(s)}:${s.size}` : null - } catch { - return null + } catch (error) { + if (isMissingPathError(error)) { + return null + } + throw error } } @@ -144,15 +156,11 @@ async function snapshotGitCommon( .filter((entry) => entry.isDirectory()) .map((entry) => join(worktreesDir, entry.name)) } catch (error) { - if ((error as NodeJS.ErrnoException).code === 'ENOENT') { + if (isMissingPathError(error)) { // Dir genuinely absent (no linked worktrees, or all removed) → authoritative empty listing. entryPaths = [] } else { - // Why: a TRANSIENT readdir failure (EIO/ESTALE/EMFILE, network/SSH hiccup) must not masquerade as - // "every worktree removed" — that would emit false delete events (and false creates next tick). - // Reuse the known entries so per-entry stats still run; a real removal surfaces as that entry's own - // stat miss (handled in snapshotGitCommonEntry), and the next successful readdir catches any add. - entryPaths = previous ? [...previous.entries.keys()] : [] + throw error } } @@ -256,45 +264,40 @@ function diffGitCommon( export async function startGitCommonPolling( commonDirPath: string, onEvents: (events: WorktreeBasePollEvent[]) => void, - pollIntervalMs: number, + cadence: GitCommonPollingCadence, visibility: WorktreePollerWindowVisibility, onFullScan?: () => void, includePrimary = true ): Promise { let disposed = false - let ticking = false - let tickCount = 0 - let snapshot = await snapshotGitCommon(commonDirPath, undefined, includePrimary) - let timer: ReturnType | null = null - let parkedWhileHidden = false - - const tick = async (forceFullScan = false): Promise => { - timer = null - if (disposed) { - return - } - if (!visibility.isWindowVisible()) { - parkedWhileHidden = true - return - } - if (ticking) { - return - } - ticking = true - // Why: measure from tick start so cadence is start-to-start, not gap-after-completion (which would - // land each visible refresh a full scan-duration late every tick). - const startedAt = Date.now() - tickCount++ - const shouldForceFullScan = forceFullScan || tickCount % INDEX_BACKSTOP_TICKS === 0 - try { + let snapshot = await tryTakeGitCommonPollBaseline(() => + snapshotGitCommon(commonDirPath, undefined, includePrimary) + ) + const polling = startAdaptiveGitCommonPoller({ + cadence, + visibility, + poll: async (forceFullScan) => { const next = await snapshotGitCommon( commonDirPath, - snapshot, + snapshot ?? undefined, includePrimary, - shouldForceFullScan + forceFullScan ) if (disposed) { - return + return { changed: false } + } + if (!snapshot) { + snapshot = next + onEvents([ + { type: 'update', path: join(commonDirPath, 'worktrees') }, + ...(includePrimary + ? PRIMARY_CHECKOUT_METADATA_FILES.map((name) => ({ + type: 'update' as const, + path: join(commonDirPath, name) + })) + : []) + ]) + return { changed: true } } if (next.didFullScan) { onFullScan?.() @@ -304,44 +307,14 @@ export async function startGitCommonPolling( if (events.length > 0) { onEvents(events) } - } catch { - // Transient fs error: keep the previous snapshot and retry next tick. - } finally { - ticking = false + return { changed: events.length > 0 } } - if (!disposed) { - // Why: clamp to [0, pollIntervalMs]. Date.now() is not monotonic — a backward wall-clock jump (NTP) would - // otherwise make elapsed negative and push the next tick out by the adjustment (suppressing refreshes for - // minutes); the upper clamp caps the wait at one interval, the lower clamp keeps a long scan from going negative. - const nextDelay = Math.max( - 0, - Math.min(pollIntervalMs, pollIntervalMs - (Date.now() - startedAt)) - ) - timer = setTimeout(() => void tick(), nextDelay) - timer.unref?.() - } - } - - const unsubscribeVisibility = visibility.onWindowBecameVisible(() => { - if (disposed || !parkedWhileHidden) { - return - } - parkedWhileHidden = false - // Why: a linked index can change without its parent dir signature moving; - // force the leaf read when diffing the retained pre-hide snapshot. - void tick(true) }) - timer = setTimeout(() => void tick(), pollIntervalMs) - timer.unref?.() - return { unsubscribe: async () => { disposed = true - if (timer) { - clearTimeout(timer) - } - unsubscribeVisibility() + await polling.unsubscribe() } } } diff --git a/src/main/ipc/worktree-git-common-watch.test.ts b/src/main/ipc/worktree-git-common-watch.test.ts index 6dab9765130..ed3fc74741f 100644 --- a/src/main/ipc/worktree-git-common-watch.test.ts +++ b/src/main/ipc/worktree-git-common-watch.test.ts @@ -21,14 +21,23 @@ vi.mock('./parcel-watcher-process', () => ({ })) // Records every stat target so a test can assert which paths a parked poll stopped touching. -const { statCalls } = vi.hoisted(() => ({ statCalls: [] as string[] })) +const { statCalls, transientStatFailures } = vi.hoisted(() => ({ + statCalls: [] as string[], + transientStatFailures: new Map() +})) vi.mock('node:fs/promises', async (importOriginal) => { const actual = await importOriginal() return { ...actual, - stat: (...args: Parameters) => { - statCalls.push(String(args[0])) + stat: async (...args: Parameters) => { + const path = String(args[0]) + statCalls.push(path) + const remainingFailures = transientStatFailures.get(path) ?? 0 + if (remainingFailures > 0) { + transientStatFailures.set(path, remainingFailures - 1) + throw Object.assign(new Error('transient stat failure'), { code: 'EIO' }) + } return actual.stat(...args) } } @@ -89,6 +98,7 @@ describe('worktree git-common narrow watch (darwin)', () => { await Promise.all(cleanups.splice(0).map((cleanup) => cleanup())) childSubscriptions = [] statCalls.length = 0 + transientStatFailures.clear() subscribeMock.mockReset() }) @@ -146,6 +156,46 @@ describe('worktree git-common narrow watch (darwin)', () => { expect(received.flat()).toContainEqual({ type: 'create', path: entryPath }) }) + it('installs the narrow stream and retries a transient primary baseline failure', async () => { + installSubscribeMock() + const commonDir = await makeCommonDir(true) + const headPath = join(commonDir, 'HEAD') + transientStatFailures.set(headPath, 1) + const received: WorktreeBasePollEvent[][] = [] + await startWatch(commonDir, received) + + expect(subscribeMock).toHaveBeenCalledTimes(1) + await vi.waitFor(() => { + expect(received.flat()).toContainEqual({ type: 'update', path: headPath }) + }) + }) + + it('resets the idle primary poll when the native stream reports activity', async () => { + installSubscribeMock() + const commonDir = await makeCommonDir(true) + const fullScans: number[] = [] + const watch = await startGitCommonWatch( + makeTarget(commonDir), + () => {}, + POLL_MS, + 'darwin', + alwaysVisible, + () => fullScans.push(performance.now()), + 1_000, + 3_000 + ) + cleanups.push(() => watch.unsubscribe()) + + await vi.waitFor(() => expect(fullScans.length).toBeGreaterThanOrEqual(3)) + const scansBeforeNativeEvent = fullScans.length + childSubscriptions[0].callback(null, [ + { type: 'update', path: join(commonDir, 'worktrees', 'native-change', 'HEAD') } + ]) + await vi.waitFor(() => expect(fullScans.length).toBeGreaterThan(scansBeforeNativeEvent), { + timeout: 250 + }) + }) + it('tears down and re-arms when the watched root is deleted', async () => { installSubscribeMock() const commonDir = await makeCommonDir(true) @@ -392,6 +442,7 @@ describe('worktree git-common polling gate (non-darwin)', () => { afterEach(async () => { await Promise.all(cleanups.splice(0).map((cleanup) => cleanup())) + transientStatFailures.clear() }) async function makePollingCommonDir(): Promise { @@ -447,6 +498,19 @@ describe('worktree git-common polling gate (non-darwin)', () => { expect(received.flat()).toHaveLength(0) }) + it('installs the poller and resyncs after a transient initial baseline failure', async () => { + const commonDir = await makePollingCommonDir() + const worktreesDir = join(commonDir, 'worktrees') + transientStatFailures.set(worktreesDir, 1) + const received: WorktreeBasePollEvent[][] = [] + + await startPollingWatch(commonDir, received) + + await vi.waitFor(() => { + expect(received.flat()).toContainEqual({ type: 'update', path: worktreesDir }) + }) + }) + it('detects linked worktree add and remove from the every-tick readdir', async () => { // Why: the worktrees-dir listing runs every tick (not gated on its stat signature), so an add/remove // surfaces within one poll interval even on a coarse-mtime filesystem whose dir signature would not diff --git a/src/main/ipc/worktree-git-common-watch.ts b/src/main/ipc/worktree-git-common-watch.ts index 217e8ed308a..8c3af3c5309 100644 --- a/src/main/ipc/worktree-git-common-watch.ts +++ b/src/main/ipc/worktree-git-common-watch.ts @@ -7,6 +7,12 @@ import type { WorktreeBaseSubscription, WorktreePollerWindowVisibility } from './worktree-base-directory-poller' +import { + startAdaptiveGitCommonPoller, + tryTakeGitCommonPollBaseline, + type AdaptiveGitCommonPollSubscription, + type GitCommonPollingCadence +} from './worktree-git-common-poll-cadence' import { PRIMARY_CHECKOUT_METADATA_FILES, startGitCommonPolling @@ -30,6 +36,11 @@ import { // Why: branch switches and commits made in the primary checkout rewrite these // top-level files (linked-worktree equivalents live under `worktrees/`). // Deliberately excludes FETCH_HEAD-style churn that carries no status change. +function isMissingPathError(error: unknown): boolean { + const code = (error as NodeJS.ErrnoException).code + return code === 'ENOENT' || code === 'ENOTDIR' +} + async function snapshotPrimaryCheckoutMetadata( commonDirPath: string ): Promise> { @@ -38,8 +49,10 @@ async function snapshotPrimaryCheckoutMetadata( const filePath = join(commonDirPath, name) try { mtimes.set(filePath, (await stat(filePath)).mtimeMs) - } catch { - // Missing file (e.g. no packed-refs yet) diffs into a create later. + } catch (error) { + if (!isMissingPathError(error)) { + throw error + } } } return mtimes @@ -69,79 +82,41 @@ function diffMtimeMap( async function startSnapshotDiffPoller( takeSnapshot: () => Promise>, onEvents: (events: WorktreeBasePollEvent[]) => void, - pollIntervalMs: number, + cadence: GitCommonPollingCadence, visibility: WorktreePollerWindowVisibility, - onFullScan?: () => void -): Promise { + onFullScan: (() => void) | undefined, + onBaselineRecovered: () => void +): Promise { let disposed = false - let ticking = false - let snapshot = await takeSnapshot() - let timer: ReturnType | null = null - let parkedWhileHidden = false - - const tick = async (): Promise => { - timer = null - if (disposed) { - return - } - if (!visibility.isWindowVisible()) { - parkedWhileHidden = true - return - } - if (ticking) { - return - } - ticking = true - // Why: measure from tick start so cadence is start-to-start, not gap-after-completion (which would - // land each visible refresh a full scan-duration late every tick). - const startedAt = Date.now() - onFullScan?.() - try { + let snapshot = await tryTakeGitCommonPollBaseline(takeSnapshot) + const polling = startAdaptiveGitCommonPoller({ + cadence: { activeIntervalMs: cadence.activeIntervalMs, idleIntervalMs: cadence.idleIntervalMs }, + visibility, + poll: async () => { + onFullScan?.() const next = await takeSnapshot() if (disposed) { - return + return { changed: false } + } + if (!snapshot) { + snapshot = next + onBaselineRecovered() + return { changed: true } } const events = diffMtimeMap(snapshot, next) snapshot = next if (events.length > 0) { onEvents(events) } - } catch { - // Transient fs error: keep the previous snapshot and retry next tick. - } finally { - ticking = false + return { changed: events.length > 0 } } - if (!disposed) { - // Why: clamp to [0, pollIntervalMs]. Date.now() is not monotonic — a backward wall-clock jump (NTP) would - // otherwise make elapsed negative and push the next tick out by the adjustment (suppressing refreshes for - // minutes); the upper clamp caps the wait at one interval, the lower clamp keeps a long scan from going negative. - const nextDelay = Math.max( - 0, - Math.min(pollIntervalMs, pollIntervalMs - (Date.now() - startedAt)) - ) - timer = setTimeout(() => void tick(), nextDelay) - timer.unref?.() - } - } - - const unsubscribeVisibility = visibility.onWindowBecameVisible(() => { - if (disposed || !parkedWhileHidden) { - return - } - parkedWhileHidden = false - void tick() }) - timer = setTimeout(() => void tick(), pollIntervalMs) - timer.unref?.() - return { + resetCadence: polling.resetCadence, unsubscribe: async () => { disposed = true - if (timer) { - clearTimeout(timer) - } - unsubscribeVisibility() + await polling.unsubscribe() } } } @@ -150,7 +125,8 @@ async function startGitCommonNarrowWatch( target: WorktreeBaseWatchTarget, onEvents: (events: WorktreeBasePollEvent[]) => void, pollIntervalMs: number, - visibility: WorktreePollerWindowVisibility + visibility: WorktreePollerWindowVisibility, + onActivity: () => void ): Promise { const worktreesDir = join(target.path, 'worktrees') let disposed = false @@ -175,6 +151,7 @@ async function startGitCommonNarrowWatch( const installed = await trySubscribe() if (installed && !disposed) { stopExistencePoll() + onActivity() // The dir appearing means a first linked worktree was just // registered; surface it so the repo's worktree list refreshes. onEvents([{ type: 'create', path: worktreesDir }]) @@ -254,11 +231,13 @@ async function startGitCommonNarrowWatch( return } if (error) { + onActivity() onEvents([{ type: 'update', path: worktreesDir }]) teardownAndRearm() return } if (events.length > 0) { + onActivity() const rootGone = events.some( (event) => event.type === 'delete' && event.path === worktreesDir ) @@ -274,6 +253,7 @@ async function startGitCommonNarrowWatch( // resubscribe gap; report a structural change so worktrees re-sync. onInterruption: () => { if (!disposed && active) { + onActivity() onEvents([{ type: 'update', path: worktreesDir }]) } } @@ -316,24 +296,42 @@ export async function startGitCommonWatch( pollIntervalMs: number, platform: NodeJS.Platform, visibility: WorktreePollerWindowVisibility, - onFullScan?: () => void + onFullScan?: () => void, + idlePollIntervalMs = pollIntervalMs * 5, + indexBackstopIntervalMs = pollIntervalMs * 15 ): Promise { + const cadence: GitCommonPollingCadence = { + activeIntervalMs: pollIntervalMs, + idleIntervalMs: idlePollIntervalMs, + indexBackstopIntervalMs + } if (platform === 'darwin') { - const [narrowWatch, primaryMetadataPoll] = await Promise.all([ - startGitCommonNarrowWatch(target, onEvents, pollIntervalMs, visibility), - startSnapshotDiffPoller( - () => snapshotPrimaryCheckoutMetadata(target.path), - onEvents, - pollIntervalMs, - visibility, - onFullScan - ) - ]) + const primaryMetadataPoll = await startSnapshotDiffPoller( + () => snapshotPrimaryCheckoutMetadata(target.path), + onEvents, + cadence, + visibility, + onFullScan, + () => + onEvents( + PRIMARY_CHECKOUT_METADATA_FILES.map((name) => ({ + type: 'update', + path: join(target.path, name) + })) + ) + ) + const narrowWatch = await startGitCommonNarrowWatch( + target, + onEvents, + pollIntervalMs, + visibility, + primaryMetadataPoll.resetCadence + ) return { unsubscribe: async () => { await Promise.all([narrowWatch.unsubscribe(), primaryMetadataPoll.unsubscribe()]) } } } - return startGitCommonPolling(target.path, onEvents, pollIntervalMs, visibility, onFullScan) + return startGitCommonPolling(target.path, onEvents, cadence, visibility, onFullScan) } diff --git a/src/renderer/src/App.tsx b/src/renderer/src/App.tsx index 1d90f1cf5e0..0745a25f490 100644 --- a/src/renderer/src/App.tsx +++ b/src/renderer/src/App.tsx @@ -146,7 +146,12 @@ import { timeRendererStartupStep, timeRendererStartupSyncStep } from './startup/startup-diagnostics' -import { reconnectSshTargetForRendererStartup } from './startup/ssh-startup-reconnect' +import { + partitionSshStartupReconnectTargets, + reconnectSshTargetsForRendererStartup, + resolveSshStartupActiveWorkspaceId, + type SshStartupReconnectBatchResult +} from './startup/ssh-startup-reconnect' import { shouldRenderPetOverlay } from './components/pet/pet-overlay-visibility' import { applyDocumentTheme } from './lib/document-theme' import { getSystemPrefersDark } from './lib/terminal-theme' @@ -196,6 +201,7 @@ import { type ExecutionHostId } from '../../shared/execution-host' import { mapWithConcurrency } from '../../shared/map-with-concurrency' +import { getConnectionIdFromState } from './lib/connection-owner-resolution' import { ModifierDoubleTapDetector, toModifierDoubleTapEvent @@ -465,6 +471,7 @@ function App(): React.JSX.Element { fetchBrowserSessionProfiles: s.fetchBrowserSessionProfiles, reconnectPersistedTerminals: s.reconnectPersistedTerminals, setDeferredSshReconnectTargets: s.setDeferredSshReconnectTargets, + removeDeferredSshReconnectTarget: s.removeDeferredSshReconnectTarget, setSshConnectionState: s.setSshConnectionState, hydratePersistedUI: s.hydratePersistedUI, setHydrationSucceeded: s.setHydrationSucceeded, @@ -998,59 +1005,77 @@ function App(): React.JSX.Element { const eagerTargets = targets.filter((t) => !t.needsPassphrase) const deferredTargets = targets.filter((t) => t.needsPassphrase) - if (deferredTargets.length > 0) { - actions.setDeferredSshReconnectTargets(deferredTargets.map((t) => t.targetId)) + const startupState = useAppStore.getState() + const activeWorkspaceId = resolveSshStartupActiveWorkspaceId({ + activeWorkspaceKey: startupState.activeWorkspaceKey, + activeWorktreeId: startupState.activeWorktreeId + }) + const activeConnectionId = getConnectionIdFromState(startupState, activeWorkspaceId) + const { criticalTargetIds, backgroundTargetIds } = + partitionSshStartupReconnectTargets({ + targetIds: eagerTargets.map(({ targetId }) => targetId), + activeTargetIds: + typeof activeConnectionId === 'string' ? [activeConnectionId] : [], + activeTabId: sessionRead.session.activeTabId, + remoteSessionIdsByTabId: sessionRead.session.remoteSessionIdsByTabId + }) + actions.setDeferredSshReconnectTargets([ + ...new Set(targets.map(({ targetId }) => targetId)) + ]) + + const reconnectTargets = (targetIds: readonly string[]) => + reconnectSshTargetsForRendererStartup({ + targetIds, + budgetMs: SSH_RECONNECT_TIMEOUT_MS, + signal: abortController.signal, + connect: (targetId) => window.api.ssh.connect({ targetId }), + publishState: actions.setSshConnectionState, + onFailure: (targetId, error) => { + console.warn(`SSH auto-reconnect failed for ${targetId}:`, error) + } + }) + const finalizeCompletedTargets = async ( + results: readonly SshStartupReconnectBatchResult[] + ): Promise => { + for (const { targetId, outcome } of results) { + if (outcome !== 'completed' || abortController.signal.aborted) { + continue + } + // Why: older/wrapped providers may return no connect state; poll once before releasing the deferred pane gate. + try { + const state = await window.api.ssh.getState({ targetId }) + if (state?.status === 'connected') { + actions.setSshConnectionState(targetId, state) + } + } catch { + /* best-effort */ + } + if (!abortController.signal.aborted) { + actions.removeDeferredSshReconnectTarget(targetId) + } + } } - // Why: treat timed-out eager targets as deferred so their PTYs reattach on tab focus (ssh.connect keeps running in main and likely finishes by then). - const timedOutTargets: string[] = [] - await timeRendererStartupStep( + const criticalResults = await timeRendererStartupStep( 'ssh-reconnect', - () => - Promise.all( - eagerTargets.map(async ({ targetId }) => { - const result = await reconnectSshTargetForRendererStartup({ - targetId, - timeoutMs: SSH_RECONNECT_TIMEOUT_MS, - connect: (id) => window.api.ssh.connect({ targetId: id }), - publishState: actions.setSshConnectionState, - onFailure: (id, error) => { - console.warn(`SSH auto-reconnect failed for ${id}:`, error) - } - }) - if (result.timedOut) { - timedOutTargets.push(targetId) - } - }) - ), + () => reconnectTargets(criticalTargetIds), { - eagerTargets: eagerTargets.length, + criticalTargets: criticalTargetIds.length, + backgroundTargets: backgroundTargetIds.length, deferredTargets: deferredTargets.length } ) - if (timedOutTargets.length > 0) { - actions.setDeferredSshReconnectTargets([ - ...deferredTargets.map((t) => t.targetId), - ...timedOutTargets - ]) - } - - // Why: older/wrapped providers may return no state from connect; poll main once as a compatibility fallback before terminal restoration. - for (const { targetId } of eagerTargets) { - if (timedOutTargets.includes(targetId)) { - continue - } - try { - const state = await window.api.ssh.getState({ targetId }) - console.warn( - `[ssh-restore] Polled state for ${targetId}: status=${state?.status}` - ) - if (state?.status === 'connected') { - actions.setSshConnectionState(targetId, state) - } - } catch { - /* best-effort */ - } + await finalizeCompletedTargets(criticalResults) + if ( + criticalResults.every(({ outcome }) => outcome === 'completed') && + backgroundTargetIds.length > 0 && + !abortController.signal.aborted + ) { + void timeRendererStartupStep('ssh-reconnect-background', () => + reconnectTargets(backgroundTargetIds) + ) + .then(finalizeCompletedTargets) + .catch((error) => console.warn('SSH background reconnect failed:', error)) } } catch (err) { console.warn('SSH startup reconnect failed:', err) diff --git a/src/renderer/src/startup/ssh-startup-reconnect.test.ts b/src/renderer/src/startup/ssh-startup-reconnect.test.ts index 44872a97f6e..ab54d202ac0 100644 --- a/src/renderer/src/startup/ssh-startup-reconnect.test.ts +++ b/src/renderer/src/startup/ssh-startup-reconnect.test.ts @@ -1,6 +1,12 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import type { SshConnectionState, SshProviderEpoch } from '../../../shared/ssh-types' -import { reconnectSshTargetForRendererStartup } from './ssh-startup-reconnect' +import { + partitionSshStartupReconnectTargets, + reconnectSshTargetForRendererStartup, + reconnectSshTargetsForRendererStartup, + resolveSshStartupActiveWorkspaceId, + SshStartupReconnectScheduler +} from './ssh-startup-reconnect' const connectedState: SshConnectionState = { targetId: 'ssh-1', @@ -12,6 +18,24 @@ const connectedState: SshConnectionState = { remotePlatform: 'linux' } +function stateFor(targetId: string): SshConnectionState { + return { ...connectedState, targetId } +} + +function controlledPromise(): { + promise: Promise + resolve: (value: T) => void + reject: (error: unknown) => void +} { + let resolve!: (value: T) => void + let reject!: (error: unknown) => void + const promise = new Promise((nextResolve, nextReject) => { + resolve = nextResolve + reject = nextReject + }) + return { promise, resolve, reject } +} + afterEach(() => { vi.useRealTimers() }) @@ -76,3 +100,225 @@ describe('reconnectSshTargetForRendererStartup', () => { expect(onFailure).toHaveBeenCalledWith('ssh-1', expect.any(Error)) }) }) + +describe('partitionSshStartupReconnectTargets', () => { + it('unwraps an active worktree key before connection-owner resolution', () => { + expect( + resolveSshStartupActiveWorkspaceId({ + activeWorkspaceKey: 'worktree:repo-1::/remote/project', + activeWorktreeId: 'stale-worktree' + }) + ).toBe('repo-1::/remote/project') + expect( + resolveSshStartupActiveWorkspaceId({ + activeWorkspaceKey: 'folder:folder-1', + activeWorktreeId: null + }) + ).toBe('folder:folder-1') + }) + + it('prioritizes active targets, then persisted sessions, in stable target order', () => { + expect( + partitionSshStartupReconnectTargets({ + targetIds: ['idle-a', 'session-b', 'active-c', 'session-d', 'active-c'], + activeTargetIds: ['active-c', 'missing'], + activeTabId: 'tab-active', + remoteSessionIdsByTabId: { + 'tab-other': 'ssh:session-b@@pty-1', + 'tab-active': 'ssh:session-d@@pty-2', + malformed: 'relay-pty-without-target', + stale: 'ssh:missing@@pty-3' + } + }) + ).toEqual({ + criticalTargetIds: ['active-c', 'session-d'], + backgroundTargetIds: ['session-b', 'idle-a'] + }) + }) +}) + +describe('reconnectSshTargetsForRendererStartup', () => { + it('bounds raw concurrent attempts and starts queued targets as slots settle', async () => { + const scheduler = new SshStartupReconnectScheduler(2) + const controls = new Map>>() + let active = 0 + let peak = 0 + const starts: string[] = [] + const resultPromise = reconnectSshTargetsForRendererStartup({ + targetIds: ['ssh-1', 'ssh-2', 'ssh-3', 'ssh-4'], + budgetMs: 1_000, + signal: new AbortController().signal, + scheduler, + connect: (targetId) => { + starts.push(targetId) + active++ + peak = Math.max(peak, active) + const control = controlledPromise() + controls.set(targetId, control) + return control.promise.finally(() => { + active-- + }) + }, + publishState: vi.fn(), + onFailure: vi.fn() + }) + + await vi.waitFor(() => expect(starts).toEqual(['ssh-1', 'ssh-2'])) + controls.get('ssh-1')!.resolve(stateFor('ssh-1')) + await vi.waitFor(() => expect(starts).toEqual(['ssh-1', 'ssh-2', 'ssh-3'])) + controls.get('ssh-2')!.resolve(stateFor('ssh-2')) + await vi.waitFor(() => expect(starts).toEqual(['ssh-1', 'ssh-2', 'ssh-3', 'ssh-4'])) + controls.get('ssh-3')!.resolve(stateFor('ssh-3')) + controls.get('ssh-4')!.resolve(stateFor('ssh-4')) + + await expect(resultPromise).resolves.toEqual( + ['ssh-1', 'ssh-2', 'ssh-3', 'ssh-4'].map((targetId) => ({ + targetId, + outcome: 'completed' + })) + ) + expect(peak).toBe(2) + }) + + it('uses one batch budget and never starts targets still queued at expiry', async () => { + vi.useFakeTimers() + const starts: string[] = [] + const resultPromise = reconnectSshTargetsForRendererStartup({ + targetIds: ['ssh-stalled', 'ssh-queued'], + budgetMs: 1_000, + signal: new AbortController().signal, + scheduler: new SshStartupReconnectScheduler(1), + connect: (targetId) => { + starts.push(targetId) + return new Promise(() => {}) + }, + publishState: vi.fn(), + onFailure: vi.fn() + }) + + await vi.advanceTimersByTimeAsync(1_000) + + await expect(resultPromise).resolves.toEqual([ + { targetId: 'ssh-stalled', outcome: 'timed-out' }, + { targetId: 'ssh-queued', outcome: 'not-started-budget' } + ]) + expect(starts).toEqual(['ssh-stalled']) + }) + + it('cancels queued results and suppresses late state publication', async () => { + const scheduler = new SshStartupReconnectScheduler(1) + const abortController = new AbortController() + const first = controlledPromise() + const starts: string[] = [] + const publishState = vi.fn() + const onFailure = vi.fn() + const resultPromise = reconnectSshTargetsForRendererStartup({ + targetIds: ['ssh-started', 'ssh-queued'], + budgetMs: 1_000, + signal: abortController.signal, + scheduler, + connect: (targetId) => { + starts.push(targetId) + return first.promise + }, + publishState, + onFailure + }) + + await vi.waitFor(() => expect(starts).toEqual(['ssh-started'])) + abortController.abort() + await expect(resultPromise).resolves.toEqual([ + { targetId: 'ssh-started', outcome: 'cancelled' }, + { targetId: 'ssh-queued', outcome: 'cancelled' } + ]) + + first.resolve(stateFor('ssh-started')) + await first.promise + await Promise.resolve() + expect(starts).toEqual(['ssh-started']) + expect(publishState).not.toHaveBeenCalled() + expect(onFailure).not.toHaveBeenCalled() + }) + + it('retains capacity for duplicate in-progress rejections until the first budget ends', async () => { + vi.useFakeTimers() + const scheduler = new SshStartupReconnectScheduler(1) + const starts: string[] = [] + const firstResult = reconnectSshTargetsForRendererStartup({ + targetIds: ['ssh-duplicate'], + budgetMs: 1_000, + signal: new AbortController().signal, + scheduler, + connect: async (targetId) => { + starts.push(targetId) + throw new Error('Connection to duplicate host is already in progress') + }, + publishState: vi.fn(), + onFailure: vi.fn() + }) + + await expect(firstResult).resolves.toEqual([ + { targetId: 'ssh-duplicate', outcome: 'in-progress' } + ]) + const secondResult = reconnectSshTargetsForRendererStartup({ + targetIds: ['ssh-next'], + budgetMs: 2_000, + signal: new AbortController().signal, + scheduler, + connect: async (targetId) => { + starts.push(targetId) + return stateFor(targetId) + }, + publishState: vi.fn(), + onFailure: vi.fn() + }) + + await vi.advanceTimersByTimeAsync(999) + expect(starts).toEqual(['ssh-duplicate']) + await vi.advanceTimersByTimeAsync(1) + await expect(secondResult).resolves.toEqual([{ targetId: 'ssh-next', outcome: 'completed' }]) + expect(starts).toEqual(['ssh-duplicate', 'ssh-next']) + }) + + it('keeps an aborted raw attempt in the pool until its connect promise settles', async () => { + const scheduler = new SshStartupReconnectScheduler(1) + const firstAbort = new AbortController() + const first = controlledPromise() + const starts: string[] = [] + const firstResult = reconnectSshTargetsForRendererStartup({ + targetIds: ['ssh-old'], + budgetMs: 1_000, + signal: firstAbort.signal, + scheduler, + connect: (targetId) => { + starts.push(targetId) + return first.promise + }, + publishState: vi.fn(), + onFailure: vi.fn() + }) + + await vi.waitFor(() => expect(starts).toEqual(['ssh-old'])) + firstAbort.abort() + await expect(firstResult).resolves.toEqual([{ targetId: 'ssh-old', outcome: 'cancelled' }]) + + const nextResult = reconnectSshTargetsForRendererStartup({ + targetIds: ['ssh-new'], + budgetMs: 1_000, + signal: new AbortController().signal, + scheduler, + connect: async (targetId) => { + starts.push(targetId) + return stateFor(targetId) + }, + publishState: vi.fn(), + onFailure: vi.fn() + }) + await Promise.resolve() + expect(starts).toEqual(['ssh-old']) + + first.resolve(stateFor('ssh-old')) + await expect(nextResult).resolves.toEqual([{ targetId: 'ssh-new', outcome: 'completed' }]) + expect(starts).toEqual(['ssh-old', 'ssh-new']) + }) +}) diff --git a/src/renderer/src/startup/ssh-startup-reconnect.ts b/src/renderer/src/startup/ssh-startup-reconnect.ts index d312239a878..b32ea0ccb03 100644 --- a/src/renderer/src/startup/ssh-startup-reconnect.ts +++ b/src/renderer/src/startup/ssh-startup-reconnect.ts @@ -1,9 +1,278 @@ import type { SshConnectionState } from '../../../shared/ssh-types' +import { parseAppSshPtyId } from '../../../shared/ssh-pty-id' +import { getActiveSidebarWorkspaceId } from '../../../shared/workspace-scope' + +export const SSH_STARTUP_RECONNECT_CONCURRENCY = 3 export type SshStartupReconnectResult = { timedOut: boolean } +export type SshStartupReconnectOutcome = + | 'completed' + | 'failed' + | 'timed-out' + | 'in-progress' + | 'not-started-budget' + | 'cancelled' + +export type SshStartupReconnectBatchResult = { + targetId: string + outcome: SshStartupReconnectOutcome +} + +type ScheduledReconnect = { + targetId: string + deadline: number + signal: AbortSignal + connect: () => Promise + publishState: (state: SshConnectionState) => void + onFailure: (error: unknown) => void + resolve: (result: SshStartupReconnectBatchResult) => void + started: boolean + resultSettled: boolean + released: boolean + holdUntilBudget: boolean + budgetTimer: ReturnType | null + removeAbortListener: () => void +} + +function remainingBudgetMs(deadline: number): number { + return Math.max(0, deadline - performance.now()) +} + +function isConnectAlreadyInProgress(error: unknown): boolean { + return error instanceof Error && /Connection to .+ is already in progress/.test(error.message) +} + +export class SshStartupReconnectScheduler { + private activeCount = 0 + private readonly queue: ScheduledReconnect[] = [] + + constructor(private readonly concurrency = SSH_STARTUP_RECONNECT_CONCURRENCY) {} + + schedule(args: { + targetId: string + deadline: number + signal: AbortSignal + connect: () => Promise + publishState: (state: SshConnectionState) => void + onFailure: (error: unknown) => void + }): Promise { + return new Promise((resolve) => { + const task: ScheduledReconnect = { + ...args, + resolve, + started: false, + resultSettled: false, + released: false, + holdUntilBudget: false, + budgetTimer: null, + removeAbortListener: () => {} + } + const finish = (outcome: SshStartupReconnectOutcome): void => { + if (task.resultSettled) { + return + } + task.resultSettled = true + task.resolve({ targetId: task.targetId, outcome }) + } + const onAbort = (): void => { + if (!task.started) { + this.removeQueuedTask(task) + this.clearTaskTimer(task) + task.removeAbortListener() + finish('cancelled') + this.drain() + return + } + finish('cancelled') + } + args.signal.addEventListener('abort', onAbort, { once: true }) + task.removeAbortListener = () => args.signal.removeEventListener('abort', onAbort) + + if (args.signal.aborted) { + onAbort() + return + } + const budgetMs = remainingBudgetMs(args.deadline) + if (budgetMs <= 0) { + task.removeAbortListener() + finish('not-started-budget') + return + } + task.budgetTimer = setTimeout(() => { + task.budgetTimer = null + if (!task.started) { + this.removeQueuedTask(task) + task.removeAbortListener() + finish('not-started-budget') + this.drain() + return + } + if (task.holdUntilBudget) { + this.release(task) + return + } + if (!task.resultSettled && !task.signal.aborted) { + const error = new Error('SSH reconnect timeout') + task.onFailure(error) + finish('timed-out') + } + }, budgetMs) + this.queue.push(task) + this.drain() + }) + } + + private drain(): void { + while (this.activeCount < this.concurrency && this.queue.length > 0) { + const task = this.queue.shift()! + if (task.signal.aborted || remainingBudgetMs(task.deadline) <= 0) { + this.clearTaskTimer(task) + task.removeAbortListener() + if (!task.resultSettled) { + task.resolve({ + targetId: task.targetId, + outcome: task.signal.aborted ? 'cancelled' : 'not-started-budget' + }) + task.resultSettled = true + } + continue + } + this.start(task) + } + } + + private start(task: ScheduledReconnect): void { + task.started = true + this.activeCount++ + void Promise.resolve() + .then(task.connect) + .then( + (state) => { + if (!task.resultSettled && !task.signal.aborted) { + if (state) { + task.publishState(state) + } + task.resultSettled = true + task.resolve({ targetId: task.targetId, outcome: 'completed' }) + } + this.release(task) + }, + (error: unknown) => { + if (isConnectAlreadyInProgress(error) && remainingBudgetMs(task.deadline) > 0) { + task.holdUntilBudget = true + if (!task.resultSettled) { + task.resultSettled = true + task.resolve({ targetId: task.targetId, outcome: 'in-progress' }) + } + return + } + if (!task.resultSettled && !task.signal.aborted) { + task.onFailure(error) + task.resultSettled = true + task.resolve({ targetId: task.targetId, outcome: 'failed' }) + } + this.release(task) + } + ) + } + + private release(task: ScheduledReconnect): void { + if (task.released) { + return + } + task.released = true + this.clearTaskTimer(task) + task.removeAbortListener() + this.activeCount-- + this.drain() + } + + private removeQueuedTask(task: ScheduledReconnect): void { + const index = this.queue.indexOf(task) + if (index >= 0) { + this.queue.splice(index, 1) + } + } + + private clearTaskTimer(task: ScheduledReconnect): void { + if (task.budgetTimer) { + clearTimeout(task.budgetTimer) + task.budgetTimer = null + } + } +} + +const startupReconnectScheduler = new SshStartupReconnectScheduler() + +export function resolveSshStartupActiveWorkspaceId(args: { + activeWorkspaceKey: string | null + activeWorktreeId: string | null +}): string | null { + return getActiveSidebarWorkspaceId(args.activeWorkspaceKey, args.activeWorktreeId) +} + +export function partitionSshStartupReconnectTargets(args: { + targetIds: readonly string[] + activeTargetIds: readonly string[] + activeTabId: string | null + remoteSessionIdsByTabId?: Readonly> +}): { criticalTargetIds: string[]; backgroundTargetIds: string[] } { + const targetIds = [...new Set(args.targetIds)] + const eligible = new Set(targetIds) + const activeTargets = new Set(args.activeTargetIds.filter((id) => eligible.has(id))) + const sessionTargets = new Set() + for (const [tabId, sessionId] of Object.entries(args.remoteSessionIdsByTabId ?? {})) { + const targetId = parseAppSshPtyId(sessionId)?.connectionId + if (!targetId || !eligible.has(targetId)) { + continue + } + sessionTargets.add(targetId) + if (tabId === args.activeTabId) { + activeTargets.add(targetId) + } + } + const criticalTargetIds = targetIds.filter((targetId) => activeTargets.has(targetId)) + return { + criticalTargetIds, + backgroundTargetIds: [ + ...targetIds.filter( + (targetId) => !activeTargets.has(targetId) && sessionTargets.has(targetId) + ), + ...targetIds.filter( + (targetId) => !activeTargets.has(targetId) && !sessionTargets.has(targetId) + ) + ] + } +} + +export function reconnectSshTargetsForRendererStartup(args: { + targetIds: readonly string[] + budgetMs: number + signal: AbortSignal + connect: (targetId: string) => Promise + publishState: (targetId: string, state: SshConnectionState) => void + onFailure: (targetId: string, error: unknown) => void + scheduler?: SshStartupReconnectScheduler +}): Promise { + const deadline = performance.now() + args.budgetMs + const scheduler = args.scheduler ?? startupReconnectScheduler + return Promise.all( + args.targetIds.map((targetId) => + scheduler.schedule({ + targetId, + deadline, + signal: args.signal, + connect: () => args.connect(targetId), + publishState: (state) => args.publishState(targetId, state), + onFailure: (error) => args.onFailure(targetId, error) + }) + ) + ) +} + export async function reconnectSshTargetForRendererStartup(args: { targetId: string timeoutMs: number @@ -18,8 +287,6 @@ export async function reconnectSshTargetForRendererStartup(args: { timeoutId = setTimeout(() => reject(new Error('SSH reconnect timeout')), timeoutMs) }) const state = await Promise.race([connect(targetId), timeout]) - // Why: the state-change IPC can trail connect's resolution. Publish the - // authoritative result before restored terminals inspect renderer state. if (state) { publishState(targetId, state) }