mirror of
https://github.com/stablyai/orca.git
synced 2026-10-20 16:03:20 +00:00
perf: bound SSH startup reconnects and idle git polling
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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<T>(
|
||||
takeSnapshot: () => Promise<T>
|
||||
): Promise<T | null> {
|
||||
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<typeof setTimeout> | 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<void> => {
|
||||
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()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<T>(): {
|
||||
promise: Promise<T>
|
||||
resolve: (value: T) => void
|
||||
} {
|
||||
let resolve!: (value: T) => void
|
||||
const promise = new Promise<T>((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()
|
||||
})
|
||||
})
|
||||
@@ -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<string> {
|
||||
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<string | null> {
|
||||
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<WorktreeBaseSubscription> {
|
||||
let disposed = false
|
||||
let ticking = false
|
||||
let tickCount = 0
|
||||
let snapshot = await snapshotGitCommon(commonDirPath, undefined, includePrimary)
|
||||
let timer: ReturnType<typeof setTimeout> | null = null
|
||||
let parkedWhileHidden = false
|
||||
|
||||
const tick = async (forceFullScan = false): Promise<void> => {
|
||||
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()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<string, number>()
|
||||
}))
|
||||
|
||||
vi.mock('node:fs/promises', async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof NodeFsPromises>()
|
||||
return {
|
||||
...actual,
|
||||
stat: (...args: Parameters<typeof actual.stat>) => {
|
||||
statCalls.push(String(args[0]))
|
||||
stat: async (...args: Parameters<typeof actual.stat>) => {
|
||||
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<string> {
|
||||
@@ -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
|
||||
|
||||
@@ -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<Map<string, number>> {
|
||||
@@ -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<Map<string, number>>,
|
||||
onEvents: (events: WorktreeBasePollEvent[]) => void,
|
||||
pollIntervalMs: number,
|
||||
cadence: GitCommonPollingCadence,
|
||||
visibility: WorktreePollerWindowVisibility,
|
||||
onFullScan?: () => void
|
||||
): Promise<WorktreeBaseSubscription> {
|
||||
onFullScan: (() => void) | undefined,
|
||||
onBaselineRecovered: () => void
|
||||
): Promise<AdaptiveGitCommonPollSubscription> {
|
||||
let disposed = false
|
||||
let ticking = false
|
||||
let snapshot = await takeSnapshot()
|
||||
let timer: ReturnType<typeof setTimeout> | null = null
|
||||
let parkedWhileHidden = false
|
||||
|
||||
const tick = async (): Promise<void> => {
|
||||
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<WorktreeBaseSubscription> {
|
||||
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<WorktreeBaseSubscription> {
|
||||
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)
|
||||
}
|
||||
|
||||
+72
-47
@@ -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<void> => {
|
||||
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)
|
||||
|
||||
@@ -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<T>(): {
|
||||
promise: Promise<T>
|
||||
resolve: (value: T) => void
|
||||
reject: (error: unknown) => void
|
||||
} {
|
||||
let resolve!: (value: T) => void
|
||||
let reject!: (error: unknown) => void
|
||||
const promise = new Promise<T>((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<string, ReturnType<typeof controlledPromise<SshConnectionState>>>()
|
||||
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<SshConnectionState>()
|
||||
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<SshConnectionState>()
|
||||
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<SshConnectionState>()
|
||||
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'])
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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<SshConnectionState | null>
|
||||
publishState: (state: SshConnectionState) => void
|
||||
onFailure: (error: unknown) => void
|
||||
resolve: (result: SshStartupReconnectBatchResult) => void
|
||||
started: boolean
|
||||
resultSettled: boolean
|
||||
released: boolean
|
||||
holdUntilBudget: boolean
|
||||
budgetTimer: ReturnType<typeof setTimeout> | 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<SshConnectionState | null>
|
||||
publishState: (state: SshConnectionState) => void
|
||||
onFailure: (error: unknown) => void
|
||||
}): Promise<SshStartupReconnectBatchResult> {
|
||||
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<Record<string, string>>
|
||||
}): { 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<string>()
|
||||
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<SshConnectionState | null>
|
||||
publishState: (targetId: string, state: SshConnectionState) => void
|
||||
onFailure: (targetId: string, error: unknown) => void
|
||||
scheduler?: SshStartupReconnectScheduler
|
||||
}): Promise<SshStartupReconnectBatchResult[]> {
|
||||
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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user