mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 08:03:39 +00:00
fix(git): back off host probes a wedged runtime never answers
Crash f2521868: a wedged WSL distro cost 431 `git remote get-url` calls in one hour, 414 of them dying on the 30s deadline, peaking at 15 concurrent `wsl.exe` children — and it escalated rather than settling, because the probe *settles* on its deadline, so the in-flight coalescer drops its entry and the next poll re-issues immediately. There was no negative cache, no circuit breaker, and no concurrency bound, so "problem infinitly repeat" was literal. Add per-host admission control at `readRemoteUrl`, the one function every forge identity probe funnels through. One unanswered probe changes nothing; two drop the host to a single slot and shed the overflow; three open a breaker that backs off 30s, 60s, then 120s. Any answer — including a repo reporting no such remote — clears the whole budget, so recovery is unattended and bounded by one cap interval. State is keyed by the runtime that executes git (native, WSL distro, SSH connection generation), so a dead distro never gates a host that is answering. Transitions leave durable crash breadcrumbs; failures do not.
This commit is contained in:
OrcaWin
@@ -0,0 +1,250 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
clearCrashBreadcrumbsForTest,
|
||||
getCrashBreadcrumbSnapshot
|
||||
} from '../crash-reporting/crash-breadcrumb-store'
|
||||
import {
|
||||
_getGitHostProbeState,
|
||||
_resetGitHostProbeBreaker,
|
||||
GIT_HOST_PROBE_BASE_COOLDOWN_MS,
|
||||
GIT_HOST_PROBE_HEALTHY_CONCURRENCY,
|
||||
GIT_HOST_PROBE_MAX_COOLDOWN_MS,
|
||||
gitProbeHostKey,
|
||||
isGitHostProbeBlockedError,
|
||||
runGuardedGitHostProbe
|
||||
} from './git-host-probe-breaker'
|
||||
|
||||
const isUnavailable = (error: unknown): boolean =>
|
||||
error instanceof Error && /timed out/i.test(error.message)
|
||||
|
||||
function timeout(): Error {
|
||||
return new Error('wsl.exe timed out.')
|
||||
}
|
||||
|
||||
function deferred<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>((settle, fail) => {
|
||||
resolve = settle
|
||||
reject = fail
|
||||
})
|
||||
return { promise, resolve, reject }
|
||||
}
|
||||
|
||||
/** Drives one probe to a settled outcome and reports whether it ran at all. */
|
||||
async function probeOnce(
|
||||
hostKey: string,
|
||||
run: () => Promise<string>
|
||||
): Promise<'answered' | 'unavailable' | 'blocked'> {
|
||||
try {
|
||||
await runGuardedGitHostProbe(hostKey, run, isUnavailable)
|
||||
return 'answered'
|
||||
} catch (error) {
|
||||
return isGitHostProbeBlockedError(error) ? 'blocked' : 'unavailable'
|
||||
}
|
||||
}
|
||||
|
||||
describe('git host probe breaker', () => {
|
||||
beforeEach(() => {
|
||||
_resetGitHostProbeBreaker()
|
||||
clearCrashBreadcrumbsForTest()
|
||||
vi.useFakeTimers()
|
||||
vi.setSystemTime(1_000_000)
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
_resetGitHostProbeBreaker()
|
||||
})
|
||||
|
||||
it('scopes state to the runtime that executes git, never globally', () => {
|
||||
expect(gitProbeHostKey({})).toBe('native')
|
||||
expect(gitProbeHostKey({ wslDistro: 'Ubuntu-24.04' })).toBe('wsl:Ubuntu-24.04')
|
||||
expect(gitProbeHostKey({ connectionId: 'ssh-1', connectionGeneration: 3 })).toBe('ssh:ssh-1:3')
|
||||
// A reconnect mints a new key, so a fresh transport never serves the old
|
||||
// one's cooldown.
|
||||
expect(gitProbeHostKey({ connectionId: 'ssh-1', connectionGeneration: 4 })).not.toBe(
|
||||
gitProbeHostKey({ connectionId: 'ssh-1', connectionGeneration: 3 })
|
||||
)
|
||||
})
|
||||
|
||||
it('leaves a single failure alone so a cold host is not punished for a blip', async () => {
|
||||
const run = vi.fn(async () => {
|
||||
throw timeout()
|
||||
})
|
||||
expect(await probeOnce('wsl:Ubuntu', run)).toBe('unavailable')
|
||||
|
||||
const ok = vi.fn(async () => 'git@github.com:acme/orca.git')
|
||||
expect(await probeOnce('wsl:Ubuntu', ok)).toBe('answered')
|
||||
expect(ok).toHaveBeenCalledTimes(1)
|
||||
expect(_getGitHostProbeState('wsl:Ubuntu')).toBeNull()
|
||||
})
|
||||
|
||||
it('opens after three consecutive unanswered probes and stops spawning', async () => {
|
||||
const run = vi.fn(async () => {
|
||||
throw timeout()
|
||||
})
|
||||
for (let attempt = 0; attempt < 3; attempt += 1) {
|
||||
expect(await probeOnce('wsl:Ubuntu', run)).toBe('unavailable')
|
||||
}
|
||||
expect(run).toHaveBeenCalledTimes(3)
|
||||
|
||||
for (let poll = 0; poll < 50; poll += 1) {
|
||||
expect(await probeOnce('wsl:Ubuntu', run)).toBe('blocked')
|
||||
}
|
||||
expect(run).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
|
||||
it('escalates the cooldown per failed trial and caps it', async () => {
|
||||
const run = async (): Promise<string> => {
|
||||
throw timeout()
|
||||
}
|
||||
for (let attempt = 0; attempt < 3; attempt += 1) {
|
||||
await probeOnce('wsl:Ubuntu', run)
|
||||
}
|
||||
|
||||
const cooldowns: number[] = []
|
||||
for (let trial = 0; trial < 4; trial += 1) {
|
||||
const state = _getGitHostProbeState('wsl:Ubuntu')
|
||||
cooldowns.push((state?.blockedUntilMs ?? 0) - Date.now())
|
||||
vi.setSystemTime(state?.blockedUntilMs ?? 0)
|
||||
expect(await probeOnce('wsl:Ubuntu', run)).toBe('unavailable')
|
||||
}
|
||||
|
||||
expect(cooldowns).toEqual([
|
||||
GIT_HOST_PROBE_BASE_COOLDOWN_MS,
|
||||
GIT_HOST_PROBE_BASE_COOLDOWN_MS * 2,
|
||||
GIT_HOST_PROBE_MAX_COOLDOWN_MS,
|
||||
GIT_HOST_PROBE_MAX_COOLDOWN_MS
|
||||
])
|
||||
})
|
||||
|
||||
it('admits exactly one trial probe once the cooldown expires', async () => {
|
||||
const run = vi.fn(async () => {
|
||||
throw timeout()
|
||||
})
|
||||
for (let attempt = 0; attempt < 3; attempt += 1) {
|
||||
await probeOnce('wsl:Ubuntu', run)
|
||||
}
|
||||
run.mockClear()
|
||||
|
||||
const trial = deferred<string>()
|
||||
vi.setSystemTime(Date.now() + GIT_HOST_PROBE_BASE_COOLDOWN_MS)
|
||||
const trialProbe = probeOnce('wsl:Ubuntu', () => trial.promise)
|
||||
await Promise.resolve()
|
||||
|
||||
expect(await probeOnce('wsl:Ubuntu', run)).toBe('blocked')
|
||||
expect(run).not.toHaveBeenCalled()
|
||||
|
||||
trial.resolve('git@github.com:acme/orca.git')
|
||||
expect(await trialProbe).toBe('answered')
|
||||
})
|
||||
|
||||
it('serializes and sheds once a host has failed twice, but never before', async () => {
|
||||
const first = deferred<string>()
|
||||
const second = deferred<string>()
|
||||
|
||||
// Healthy: concurrent probes run side by side.
|
||||
const healthyA = probeOnce('wsl:Ubuntu', () => first.promise)
|
||||
const healthyB = probeOnce('wsl:Ubuntu', () => second.promise)
|
||||
await Promise.resolve()
|
||||
first.reject(timeout())
|
||||
second.reject(timeout())
|
||||
expect(await healthyA).toBe('unavailable')
|
||||
expect(await healthyB).toBe('unavailable')
|
||||
|
||||
// Degraded: a second concurrent probe is shed instead of queued.
|
||||
const held = deferred<string>()
|
||||
const outstanding = probeOnce('wsl:Ubuntu', () => held.promise)
|
||||
await Promise.resolve()
|
||||
const shedRun = vi.fn(async () => 'unused')
|
||||
expect(await probeOnce('wsl:Ubuntu', shedRun)).toBe('blocked')
|
||||
expect(shedRun).not.toHaveBeenCalled()
|
||||
|
||||
held.resolve('git@github.com:acme/orca.git')
|
||||
expect(await outstanding).toBe('answered')
|
||||
})
|
||||
|
||||
it('caps healthy concurrency and queues the overflow rather than failing it', async () => {
|
||||
const gates = Array.from({ length: GIT_HOST_PROBE_HEALTHY_CONCURRENCY + 3 }, () =>
|
||||
deferred<string>()
|
||||
)
|
||||
const started: number[] = []
|
||||
const probes = gates.map((gate, index) =>
|
||||
probeOnce('native', () => {
|
||||
started.push(index)
|
||||
return gate.promise
|
||||
})
|
||||
)
|
||||
await Promise.resolve()
|
||||
|
||||
expect(started).toHaveLength(GIT_HOST_PROBE_HEALTHY_CONCURRENCY)
|
||||
for (const gate of gates) {
|
||||
gate.resolve('git@github.com:acme/orca.git')
|
||||
}
|
||||
expect(await Promise.all(probes)).toEqual(gates.map(() => 'answered'))
|
||||
expect(started).toHaveLength(gates.length)
|
||||
})
|
||||
|
||||
it('treats a host answering "no such remote" as proof it is alive', async () => {
|
||||
const missing = async (): Promise<string> => {
|
||||
throw new Error("fatal: No such remote 'origin'")
|
||||
}
|
||||
for (let attempt = 0; attempt < 5; attempt += 1) {
|
||||
expect(await probeOnce('wsl:Ubuntu', missing)).toBe('unavailable')
|
||||
}
|
||||
expect(_getGitHostProbeState('wsl:Ubuntu')).toBeNull()
|
||||
})
|
||||
|
||||
it('keeps a wedged distro from gating a different host', async () => {
|
||||
const run = async (): Promise<string> => {
|
||||
throw timeout()
|
||||
}
|
||||
for (let attempt = 0; attempt < 3; attempt += 1) {
|
||||
await probeOnce('wsl:Ubuntu', run)
|
||||
}
|
||||
expect(await probeOnce('wsl:Ubuntu', run)).toBe('blocked')
|
||||
|
||||
const native = vi.fn(async () => 'git@github.com:acme/orca.git')
|
||||
expect(await probeOnce('native', native)).toBe('answered')
|
||||
expect(await probeOnce('ssh:conn-1:0', native)).toBe('answered')
|
||||
expect(native).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('leaves a durable breadcrumb for the outage and for the recovery', async () => {
|
||||
const wedged = async (): Promise<string> => {
|
||||
throw timeout()
|
||||
}
|
||||
for (let attempt = 0; attempt < 3; attempt += 1) {
|
||||
await probeOnce('wsl:Ubuntu', wedged)
|
||||
}
|
||||
|
||||
const opened = getCrashBreadcrumbSnapshot().filter(
|
||||
(crumb) => crumb.name === 'git_host_probe_breaker_open'
|
||||
)
|
||||
expect(opened).toHaveLength(1)
|
||||
expect(opened[0]?.data).toMatchObject({ host: 'wsl:Ubuntu', unansweredProbes: 3 })
|
||||
|
||||
// A wedged host must not turn one breadcrumb per failure into the noise it
|
||||
// was meant to explain.
|
||||
for (let trial = 0; trial < 6; trial += 1) {
|
||||
vi.setSystemTime(Date.now() + GIT_HOST_PROBE_MAX_COOLDOWN_MS)
|
||||
await probeOnce('wsl:Ubuntu', wedged)
|
||||
}
|
||||
expect(
|
||||
getCrashBreadcrumbSnapshot().filter((crumb) => crumb.name.startsWith('git_host_probe')).length
|
||||
).toBeLessThanOrEqual(3)
|
||||
|
||||
vi.setSystemTime(Date.now() + GIT_HOST_PROBE_MAX_COOLDOWN_MS)
|
||||
expect(await probeOnce('wsl:Ubuntu', async () => 'git@github.com:acme/orca.git')).toBe(
|
||||
'answered'
|
||||
)
|
||||
expect(
|
||||
getCrashBreadcrumbSnapshot().some((crumb) => crumb.name === 'git_host_probe_recovered')
|
||||
).toBe(true)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,310 @@
|
||||
import {
|
||||
recordCoalescedDurableCrashBreadcrumb,
|
||||
recordDurableCrashBreadcrumb
|
||||
} from '../crash-reporting/durable-crash-breadcrumb'
|
||||
|
||||
/**
|
||||
* Per-host admission control for the git host probes behind forge detection
|
||||
* (crash f2521868).
|
||||
*
|
||||
* Why: the remote-URL probe's deadline makes a wedged call *settle* rather than
|
||||
* hang, so the in-flight coalescer drops its entry and the very next poll
|
||||
* re-issues — at whatever concurrency the pollers happen to have. In the
|
||||
* reported hour that was 431 calls, 414 of them dying on the deadline, peaking
|
||||
* at 15 concurrent `wsl.exe` children whose contention stretched a 30s deadline
|
||||
* into a 120s span, so the storm fed itself and never recovered. Nothing here
|
||||
* changes how a probe is run; it decides how hard a host that is not answering
|
||||
* may still be pushed, and it always keeps a way back.
|
||||
*
|
||||
* State is scoped to the host that executes git — the native host, one WSL
|
||||
* distro, one SSH connection generation — so a dead distro can never gate the
|
||||
* repos of a host that is fine.
|
||||
*/
|
||||
|
||||
/**
|
||||
* One unanswered probe is a blip: a cold WSL interop call, a dropped relay
|
||||
* frame. Two means the host has spent a minute answering nothing, so stop
|
||||
* fanning out. Three means it is wedged, and probing it on demand only feeds
|
||||
* the loop. Tripping any earlier would punish WSL's genuinely slow cold start,
|
||||
* which is exactly when a user opening Orca deserves a real answer.
|
||||
*/
|
||||
export const GIT_HOST_PROBE_SERIALIZE_AFTER = 2
|
||||
export const GIT_HOST_PROBE_OPEN_AFTER = 3
|
||||
|
||||
/**
|
||||
* The incident peaked at 15 concurrent children, so a ceiling only bounds
|
||||
* anything if it sits under that. A burst wider than this queues rather than
|
||||
* fails — on a healthy host each probe is a sub-second config read, so the
|
||||
* waves cost far less than one contended spawn. A degraded host gets one slot.
|
||||
*/
|
||||
export const GIT_HOST_PROBE_HEALTHY_CONCURRENCY = 8
|
||||
|
||||
/**
|
||||
* A probe costs one full deadline, so retrying sooner than that spends more
|
||||
* wall time inside the host than outside it. The ceiling is the longest span a
|
||||
* single wedged call was observed to hold the host: past that, waiting longer
|
||||
* buys no further idle time and only delays an unattended recovery.
|
||||
*/
|
||||
export const GIT_HOST_PROBE_BASE_COOLDOWN_MS = 30_000
|
||||
export const GIT_HOST_PROBE_MAX_COOLDOWN_MS = 120_000
|
||||
|
||||
/** Past this the fan-out is pathological; shed rather than buffer it. */
|
||||
const MAX_QUEUED_PER_HOST = 64
|
||||
/** Reconnect churn mints a key per SSH generation, so bound retained hosts. */
|
||||
const MAX_TRACKED_HOSTS = 64
|
||||
/** A two-hour outage must still be legible in a 30-entry breadcrumb ring. */
|
||||
const STILL_OPEN_BREADCRUMB_INTERVAL_MS = 5 * 60_000
|
||||
/** Keeps `2 ** openCount` away from Infinity on a host down for days. */
|
||||
const MAX_COOLDOWN_DOUBLINGS = 20
|
||||
|
||||
type HostProbeState = {
|
||||
consecutiveUnavailable: number
|
||||
openCount: number
|
||||
blockedUntilMs: number
|
||||
inFlight: number
|
||||
queued: number
|
||||
waiters: (() => void)[]
|
||||
reportedOpen: boolean
|
||||
}
|
||||
|
||||
type Admission = 'closed' | 'half-open' | 'blocked' | 'queue'
|
||||
|
||||
export type GitHostProbeBlockedError = Error & { gitHostProbeBlocked: true }
|
||||
|
||||
export type GitProbeHostParts = {
|
||||
connectionId?: string | null
|
||||
connectionGeneration?: number
|
||||
wslDistro?: string
|
||||
}
|
||||
|
||||
const hostStates = new Map<string, HostProbeState>()
|
||||
|
||||
/** Scopes probe state to the runtime that actually executes git. */
|
||||
export function gitProbeHostKey(parts: GitProbeHostParts): string {
|
||||
if (parts.connectionId) {
|
||||
// Why: a reconnect retires the failing transport, so the new generation is a
|
||||
// new key and starts trusted instead of serving out the old one's cooldown.
|
||||
return `ssh:${parts.connectionId}:${parts.connectionGeneration ?? 0}`
|
||||
}
|
||||
return parts.wslDistro ? `wsl:${parts.wslDistro}` : 'native'
|
||||
}
|
||||
|
||||
export function isGitHostProbeBlockedError(error: unknown): error is GitHostProbeBlockedError {
|
||||
return (
|
||||
typeof error === 'object' &&
|
||||
error !== null &&
|
||||
(error as { gitHostProbeBlocked?: unknown }).gitHostProbeBlocked === true
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs `probe` under this host's failure budget. `isUnavailableError` decides
|
||||
* which rejections are the host failing to answer — anything else, including a
|
||||
* repo saying it has no such remote, is proof the host is alive and clears the
|
||||
* budget. Rejects without running `probe` when the host is being backed off.
|
||||
*/
|
||||
export async function runGuardedGitHostProbe<T>(
|
||||
hostKey: string,
|
||||
probe: () => Promise<T>,
|
||||
isUnavailableError: (error: unknown) => boolean
|
||||
): Promise<T> {
|
||||
const state = getHostState(hostKey)
|
||||
const halfOpen = await admit(hostKey, state)
|
||||
try {
|
||||
const result = await probe()
|
||||
recordAnswered(hostKey, state)
|
||||
return result
|
||||
} catch (error) {
|
||||
if (isUnavailableError(error)) {
|
||||
recordUnavailable(hostKey, state, halfOpen)
|
||||
} else {
|
||||
recordAnswered(hostKey, state)
|
||||
}
|
||||
throw error
|
||||
} finally {
|
||||
release(hostKey, state)
|
||||
}
|
||||
}
|
||||
|
||||
/** Resolves to true when this probe is the trial that decides whether the host is back. */
|
||||
async function admit(hostKey: string, state: HostProbeState): Promise<boolean> {
|
||||
for (;;) {
|
||||
const decision = tryAdmit(state)
|
||||
if (decision === 'blocked') {
|
||||
throw createBlockedError(hostKey, state)
|
||||
}
|
||||
if (decision !== 'queue') {
|
||||
return decision === 'half-open'
|
||||
}
|
||||
if (state.queued >= MAX_QUEUED_PER_HOST) {
|
||||
throw createBlockedError(hostKey, state)
|
||||
}
|
||||
state.queued += 1
|
||||
try {
|
||||
await new Promise<void>((wake) => state.waiters.push(wake))
|
||||
} finally {
|
||||
state.queued -= 1
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function tryAdmit(state: HostProbeState): Admission {
|
||||
const now = Date.now()
|
||||
if (state.blockedUntilMs > now) {
|
||||
return 'blocked'
|
||||
}
|
||||
if (state.blockedUntilMs > 0) {
|
||||
// Cooled down: exactly one trial probe gets to decide, alone.
|
||||
if (state.inFlight > 0) {
|
||||
return 'blocked'
|
||||
}
|
||||
state.inFlight += 1
|
||||
return 'half-open'
|
||||
}
|
||||
const ceiling =
|
||||
state.consecutiveUnavailable >= GIT_HOST_PROBE_SERIALIZE_AFTER
|
||||
? 1
|
||||
: GIT_HOST_PROBE_HEALTHY_CONCURRENCY
|
||||
if (state.inFlight >= ceiling) {
|
||||
// Why: a caller queued behind a probe that is already failing would wait a
|
||||
// whole deadline to learn what shedding tells it now, and its poll returns.
|
||||
return ceiling === 1 ? 'blocked' : 'queue'
|
||||
}
|
||||
state.inFlight += 1
|
||||
return 'closed'
|
||||
}
|
||||
|
||||
function recordAnswered(hostKey: string, state: HostProbeState): void {
|
||||
const unansweredProbes = state.consecutiveUnavailable
|
||||
const wasOpen = state.reportedOpen
|
||||
state.consecutiveUnavailable = 0
|
||||
state.openCount = 0
|
||||
state.blockedUntilMs = 0
|
||||
state.reportedOpen = false
|
||||
if (wasOpen) {
|
||||
recordDurableCrashBreadcrumb('git_host_probe_recovered', { host: hostKey, unansweredProbes })
|
||||
}
|
||||
}
|
||||
|
||||
function recordUnavailable(hostKey: string, state: HostProbeState, halfOpen: boolean): void {
|
||||
state.consecutiveUnavailable += 1
|
||||
if (halfOpen) {
|
||||
openBreaker(hostKey, state)
|
||||
return
|
||||
}
|
||||
// Why: probes admitted before the breaker opened settle together, and letting
|
||||
// each escalate would jump straight to the ceiling on the first failure wave.
|
||||
if (state.blockedUntilMs > 0 || state.consecutiveUnavailable < GIT_HOST_PROBE_OPEN_AFTER) {
|
||||
return
|
||||
}
|
||||
openBreaker(hostKey, state)
|
||||
}
|
||||
|
||||
function openBreaker(hostKey: string, state: HostProbeState): void {
|
||||
state.openCount += 1
|
||||
state.blockedUntilMs = Date.now() + cooldownMs(state.openCount)
|
||||
const data = {
|
||||
host: hostKey,
|
||||
unansweredProbes: state.consecutiveUnavailable,
|
||||
cooldownMs: state.blockedUntilMs - Date.now()
|
||||
}
|
||||
if (!state.reportedOpen) {
|
||||
state.reportedOpen = true
|
||||
recordDurableCrashBreadcrumb('git_host_probe_breaker_open', data)
|
||||
return
|
||||
}
|
||||
recordCoalescedDurableCrashBreadcrumb({
|
||||
name: 'git_host_probe_breaker_still_open',
|
||||
data,
|
||||
coalesceKey: `git-host-probe:${hostKey}`,
|
||||
minIntervalMs: STILL_OPEN_BREADCRUMB_INTERVAL_MS
|
||||
})
|
||||
}
|
||||
|
||||
function cooldownMs(openCount: number): number {
|
||||
const doublings = Math.min(Math.max(0, openCount - 1), MAX_COOLDOWN_DOUBLINGS)
|
||||
return Math.min(GIT_HOST_PROBE_BASE_COOLDOWN_MS * 2 ** doublings, GIT_HOST_PROBE_MAX_COOLDOWN_MS)
|
||||
}
|
||||
|
||||
function release(hostKey: string, state: HostProbeState): void {
|
||||
state.inFlight -= 1
|
||||
const waiters = state.waiters.splice(0)
|
||||
for (const wake of waiters) {
|
||||
wake()
|
||||
}
|
||||
// Why: a host with nothing left to remember costs nothing to forget, which is
|
||||
// what keeps the healthy case from retaining an entry per repo host forever.
|
||||
if (hostStates.get(hostKey) === state && isForgettable(state)) {
|
||||
hostStates.delete(hostKey)
|
||||
}
|
||||
}
|
||||
|
||||
function isForgettable(state: HostProbeState): boolean {
|
||||
return isIdle(state) && state.consecutiveUnavailable === 0 && state.blockedUntilMs === 0
|
||||
}
|
||||
|
||||
function isIdle(state: HostProbeState): boolean {
|
||||
return state.inFlight === 0 && state.queued === 0 && state.waiters.length === 0
|
||||
}
|
||||
|
||||
function createBlockedError(hostKey: string, state: HostProbeState): GitHostProbeBlockedError {
|
||||
const remainingMs = state.blockedUntilMs - Date.now()
|
||||
const message =
|
||||
remainingMs > 0
|
||||
? `Git host ${hostKey} did not answer ${state.consecutiveUnavailable} consecutive probes; suppressed for ~${Math.ceil(remainingMs / 1000)}s.`
|
||||
: `Git host ${hostKey} did not answer ${state.consecutiveUnavailable} consecutive probes; shed while an earlier probe is still outstanding.`
|
||||
return Object.assign(new Error(message), { gitHostProbeBlocked: true as const })
|
||||
}
|
||||
|
||||
function getHostState(hostKey: string): HostProbeState {
|
||||
const existing = hostStates.get(hostKey)
|
||||
if (existing) {
|
||||
return existing
|
||||
}
|
||||
const state: HostProbeState = {
|
||||
consecutiveUnavailable: 0,
|
||||
openCount: 0,
|
||||
blockedUntilMs: 0,
|
||||
inFlight: 0,
|
||||
queued: 0,
|
||||
waiters: [],
|
||||
reportedOpen: false
|
||||
}
|
||||
hostStates.set(hostKey, state)
|
||||
evictIdleHostStates()
|
||||
return state
|
||||
}
|
||||
|
||||
function evictIdleHostStates(): void {
|
||||
if (hostStates.size <= MAX_TRACKED_HOSTS) {
|
||||
return
|
||||
}
|
||||
for (const [key, state] of hostStates) {
|
||||
if (hostStates.size <= MAX_TRACKED_HOSTS) {
|
||||
return
|
||||
}
|
||||
// Never drop a host with live slot accounting; its release would orphan.
|
||||
if (isIdle(state)) {
|
||||
hostStates.delete(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** @internal — tests only. */
|
||||
export function _resetGitHostProbeBreaker(): void {
|
||||
hostStates.clear()
|
||||
}
|
||||
|
||||
/** @internal — tests only. */
|
||||
export function _getGitHostProbeState(
|
||||
hostKey: string
|
||||
): { consecutiveUnavailable: number; blockedUntilMs: number; inFlight: number } | null {
|
||||
const state = hostStates.get(hostKey)
|
||||
return state
|
||||
? {
|
||||
consecutiveUnavailable: state.consecutiveUnavailable,
|
||||
blockedUntilMs: state.blockedUntilMs,
|
||||
inFlight: state.inFlight
|
||||
}
|
||||
: null
|
||||
}
|
||||
@@ -0,0 +1,191 @@
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const { getSshGitProviderMock, gitExecFileAsyncMock } = vi.hoisted(() => ({
|
||||
getSshGitProviderMock: vi.fn(),
|
||||
gitExecFileAsyncMock: vi.fn()
|
||||
}))
|
||||
|
||||
vi.mock('../providers/ssh-git-dispatch', () => ({
|
||||
getSshGitProvider: getSshGitProviderMock,
|
||||
getSshGitProviderGeneration: () => 0,
|
||||
SSH_GIT_PROVIDER_UNAVAILABLE_MESSAGE: 'SSH Git provider unavailable'
|
||||
}))
|
||||
|
||||
vi.mock('./runner', () => ({ gitExecFileAsync: gitExecFileAsyncMock }))
|
||||
|
||||
import {
|
||||
_resetGitHostProbeBreaker,
|
||||
GIT_HOST_PROBE_HEALTHY_CONCURRENCY
|
||||
} from './git-host-probe-breaker'
|
||||
import {
|
||||
assertRemoteUrlReadable,
|
||||
isTransientGitProbeError,
|
||||
readRemoteUrl,
|
||||
REMOTE_URL_PROBE_TIMEOUT_MS
|
||||
} from './remote-url-probe'
|
||||
|
||||
/**
|
||||
* Crash f2521868: a wedged WSL distro answered nothing for two hours while every
|
||||
* git-backed panel stayed stale. In the final hour that cost 431 `git remote
|
||||
* get-url` calls, 414 of them dying on the deadline, peaking at 15 concurrent
|
||||
* `wsl.exe` children — and it got worse over time, because each probe settles on
|
||||
* its deadline, so the in-flight coalescer drops it and the next poll re-issues.
|
||||
*/
|
||||
const WEDGED_DISTRO = 'Ubuntu-24.04'
|
||||
const WSL_REPO = `\\\\wsl$\\${WEDGED_DISTRO}\\home\\dev\\orca`
|
||||
const POLL_INTERVAL_MS = 10_000
|
||||
const INCIDENT_WINDOW_MS = 60 * 60_000
|
||||
const REPORTED_PEAK_CONCURRENCY = 15
|
||||
|
||||
let virtualNowMs = 0
|
||||
|
||||
function advanceTo(nextMs: number): void {
|
||||
virtualNowMs = nextMs
|
||||
vi.setSystemTime(virtualNowMs)
|
||||
}
|
||||
|
||||
function wslTimeout(): Error {
|
||||
return new Error('Error: wsl.exe timed out.')
|
||||
}
|
||||
|
||||
/** Every spawn burns its full deadline, which is what the report recorded. */
|
||||
function wedgeHost(spawns: number[]): void {
|
||||
gitExecFileAsyncMock.mockImplementation(async () => {
|
||||
spawns.push(virtualNowMs)
|
||||
advanceTo(virtualNowMs + REMOTE_URL_PROBE_TIMEOUT_MS)
|
||||
throw wslTimeout()
|
||||
})
|
||||
}
|
||||
|
||||
async function pollWslRepo(): Promise<'answered' | 'unavailable'> {
|
||||
try {
|
||||
await assertRemoteUrlReadable({ repoPath: WSL_REPO, wslDistro: WEDGED_DISTRO })
|
||||
return 'answered'
|
||||
} catch {
|
||||
return 'unavailable'
|
||||
}
|
||||
}
|
||||
|
||||
describe('remote URL probe against a wedged host (crash f2521868)', () => {
|
||||
beforeEach(() => {
|
||||
getSshGitProviderMock.mockReset()
|
||||
gitExecFileAsyncMock.mockReset()
|
||||
_resetGitHostProbeBreaker()
|
||||
vi.useFakeTimers()
|
||||
advanceTo(1_700_000_000_000)
|
||||
})
|
||||
|
||||
it('stops hammering a wedged distro, then fully restores it on one success', async () => {
|
||||
const spawns: number[] = []
|
||||
wedgeHost(spawns)
|
||||
|
||||
const startedAtMs = virtualNowMs
|
||||
while (virtualNowMs - startedAtMs < INCIDENT_WINDOW_MS) {
|
||||
const pollStartedAtMs = virtualNowMs
|
||||
expect(await pollWslRepo()).toBe('unavailable')
|
||||
// A blocked poll costs nothing, so the next one lands on its own cadence.
|
||||
advanceTo(Math.max(virtualNowMs, pollStartedAtMs + POLL_INTERVAL_MS))
|
||||
}
|
||||
|
||||
// The report's hour was 431 calls; a poll every 10s over the same hour now
|
||||
// spawns 27, because the backoff converges to one deadline per 150s.
|
||||
expect(spawns.length).toBe(27)
|
||||
|
||||
const spawnsInBucket = (bucket: number): number =>
|
||||
spawns.filter(
|
||||
(at) =>
|
||||
at >= startedAtMs + bucket * 10 * 60_000 && at < startedAtMs + (bucket + 1) * 10 * 60_000
|
||||
).length
|
||||
// The report's buckets escalated (44 -> 68 -> 125) before the user gave up.
|
||||
// "problem infinitly repeat" is disproved by a tail no worse than the head.
|
||||
expect([0, 1, 2, 3, 4, 5].map(spawnsInBucket)).toEqual([7, 4, 4, 4, 4, 4])
|
||||
|
||||
// Recovery is unattended: no restart, no user action, just the host answering.
|
||||
gitExecFileAsyncMock.mockReset()
|
||||
gitExecFileAsyncMock.mockResolvedValue({ stdout: 'git@github.com:acme/orca.git\n' })
|
||||
advanceTo(virtualNowMs + 2 * 60_000)
|
||||
expect(await pollWslRepo()).toBe('answered')
|
||||
|
||||
gitExecFileAsyncMock.mockClear()
|
||||
for (let poll = 0; poll < 5; poll += 1) {
|
||||
expect(await pollWslRepo()).toBe('answered')
|
||||
}
|
||||
expect(gitExecFileAsyncMock).toHaveBeenCalledTimes(5)
|
||||
})
|
||||
|
||||
it('collapses the fan-out the report peaked at, and restores it after recovery', async () => {
|
||||
let concurrent = 0
|
||||
let peakConcurrent = 0
|
||||
const outstanding: (() => void)[] = []
|
||||
gitExecFileAsyncMock.mockImplementation(async () => {
|
||||
concurrent += 1
|
||||
peakConcurrent = Math.max(peakConcurrent, concurrent)
|
||||
await new Promise<void>((resume) => outstanding.push(resume))
|
||||
concurrent -= 1
|
||||
throw wslTimeout()
|
||||
})
|
||||
|
||||
const firstWave = Array.from({ length: REPORTED_PEAK_CONCURRENCY }, () => pollWslRepo())
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
expect(peakConcurrent).toBe(GIT_HOST_PROBE_HEALTHY_CONCURRENCY)
|
||||
while (outstanding.length > 0) {
|
||||
outstanding.pop()?.()
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
}
|
||||
expect(await Promise.all(firstWave)).toEqual(firstWave.map(() => 'unavailable'))
|
||||
// Every later probe in the wave was serialized behind the failing host.
|
||||
expect(peakConcurrent).toBe(GIT_HOST_PROBE_HEALTHY_CONCURRENCY)
|
||||
|
||||
const spawns: number[] = []
|
||||
wedgeHost(spawns)
|
||||
const secondWave = Array.from({ length: REPORTED_PEAK_CONCURRENCY }, () => pollWslRepo())
|
||||
expect(await Promise.all(secondWave)).toEqual(secondWave.map(() => 'unavailable'))
|
||||
expect(spawns).toEqual([])
|
||||
|
||||
gitExecFileAsyncMock.mockReset()
|
||||
gitExecFileAsyncMock.mockResolvedValue({ stdout: 'git@github.com:acme/orca.git\n' })
|
||||
advanceTo(virtualNowMs + 5 * 60_000)
|
||||
expect(await pollWslRepo()).toBe('answered')
|
||||
|
||||
gitExecFileAsyncMock.mockClear()
|
||||
const healthyWave = Array.from({ length: REPORTED_PEAK_CONCURRENCY }, () => pollWslRepo())
|
||||
expect(await Promise.all(healthyWave)).toEqual(healthyWave.map(() => 'answered'))
|
||||
expect(gitExecFileAsyncMock).toHaveBeenCalledTimes(REPORTED_PEAK_CONCURRENCY)
|
||||
})
|
||||
|
||||
it('keeps the wedged distro from gating repos on hosts that are answering', async () => {
|
||||
const spawns: number[] = []
|
||||
wedgeHost(spawns)
|
||||
for (let attempt = 0; attempt < 3; attempt += 1) {
|
||||
expect(await pollWslRepo()).toBe('unavailable')
|
||||
}
|
||||
spawns.length = 0
|
||||
expect(await pollWslRepo()).toBe('unavailable')
|
||||
expect(spawns).toEqual([])
|
||||
|
||||
// A native repo, and a repo on a different distro, still probe immediately.
|
||||
gitExecFileAsyncMock.mockReset()
|
||||
gitExecFileAsyncMock.mockResolvedValue({ stdout: 'git@github.com:acme/orca.git\n' })
|
||||
await expect(readRemoteUrl({ repoPath: 'C:/repos/orca' }, 'origin')).resolves.toContain(
|
||||
'github.com'
|
||||
)
|
||||
await expect(
|
||||
readRemoteUrl(
|
||||
{ repoPath: '\\\\wsl$\\Debian\\home\\dev\\orca', wslDistro: 'Debian' },
|
||||
'origin'
|
||||
)
|
||||
).resolves.toContain('github.com')
|
||||
})
|
||||
|
||||
it('reports a suppressed probe as unavailable, never as a cacheable answer', async () => {
|
||||
const spawns: number[] = []
|
||||
wedgeHost(spawns)
|
||||
for (let attempt = 0; attempt < 3; attempt += 1) {
|
||||
await pollWslRepo()
|
||||
}
|
||||
|
||||
await expect(
|
||||
readRemoteUrl({ repoPath: WSL_REPO, wslDistro: WEDGED_DISTRO }, 'origin')
|
||||
).rejects.toSatisfy(isTransientGitProbeError)
|
||||
})
|
||||
})
|
||||
@@ -7,11 +7,13 @@ const { getSshGitProviderMock, gitExecFileAsyncMock } = vi.hoisted(() => ({
|
||||
|
||||
vi.mock('../providers/ssh-git-dispatch', () => ({
|
||||
getSshGitProvider: getSshGitProviderMock,
|
||||
getSshGitProviderGeneration: () => 0,
|
||||
SSH_GIT_PROVIDER_UNAVAILABLE_MESSAGE: 'SSH Git provider unavailable'
|
||||
}))
|
||||
|
||||
vi.mock('./runner', () => ({ gitExecFileAsync: gitExecFileAsyncMock }))
|
||||
|
||||
import { _resetGitHostProbeBreaker } from './git-host-probe-breaker'
|
||||
import {
|
||||
assertRemoteUrlReadable,
|
||||
isTransientGitProbeError,
|
||||
@@ -23,6 +25,7 @@ describe('remote URL probe', () => {
|
||||
beforeEach(() => {
|
||||
getSshGitProviderMock.mockReset()
|
||||
gitExecFileAsyncMock.mockReset()
|
||||
_resetGitHostProbeBreaker()
|
||||
})
|
||||
|
||||
it('bounds local and WSL remote reads with the shared timeout', async () => {
|
||||
|
||||
@@ -1,7 +1,14 @@
|
||||
import { parseWslUncPath } from '../../shared/wsl-paths'
|
||||
import {
|
||||
getSshGitProvider,
|
||||
getSshGitProviderGeneration,
|
||||
SSH_GIT_PROVIDER_UNAVAILABLE_MESSAGE
|
||||
} from '../providers/ssh-git-dispatch'
|
||||
import {
|
||||
gitProbeHostKey,
|
||||
isGitHostProbeBlockedError,
|
||||
runGuardedGitHostProbe
|
||||
} from './git-host-probe-breaker'
|
||||
import { gitExecFileAsync } from './runner'
|
||||
import { isStableMissingGitRemoteError } from './stable-missing-git-remote-error'
|
||||
|
||||
@@ -25,27 +32,56 @@ export type RemoteUrlProbeContext = {
|
||||
wslDistro?: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Which host executes this probe. A repo reached over a `\wsl$` UNC path names
|
||||
* its distro nowhere else, and keying it as the native host would let a dead
|
||||
* distro back off the probes of local repos that are answering fine.
|
||||
*/
|
||||
function localProbeHostKey(context: RemoteUrlProbeContext): string {
|
||||
const wslDistro = context.wslDistro ?? parseWslUncPath(context.repoPath)?.distro
|
||||
return gitProbeHostKey(wslDistro ? { wslDistro } : {})
|
||||
}
|
||||
|
||||
/** Reads a remote URL, or null when the repo's SSH runtime is not connected. */
|
||||
export async function readRemoteUrl(
|
||||
context: RemoteUrlProbeContext,
|
||||
remoteName: string
|
||||
): Promise<string | null> {
|
||||
if (context.connectionId) {
|
||||
const provider = getSshGitProvider(context.connectionId)
|
||||
const connectionId = context.connectionId
|
||||
const provider = getSshGitProvider(connectionId)
|
||||
if (!provider) {
|
||||
// Costs no git, so there is nothing here for the host's budget to learn.
|
||||
return null
|
||||
}
|
||||
const { stdout } = await provider.exec(['remote', 'get-url', remoteName], context.repoPath, {
|
||||
signal: AbortSignal.timeout(REMOTE_URL_PROBE_TIMEOUT_MS)
|
||||
})
|
||||
return stdout
|
||||
return runGuardedGitHostProbe(
|
||||
gitProbeHostKey({
|
||||
connectionId,
|
||||
connectionGeneration: getSshGitProviderGeneration(connectionId)
|
||||
}),
|
||||
async () => {
|
||||
const { stdout } = await provider.exec(
|
||||
['remote', 'get-url', remoteName],
|
||||
context.repoPath,
|
||||
{ signal: AbortSignal.timeout(REMOTE_URL_PROBE_TIMEOUT_MS) }
|
||||
)
|
||||
return stdout
|
||||
},
|
||||
isTransientGitProbeError
|
||||
)
|
||||
}
|
||||
const { stdout } = await gitExecFileAsync(['remote', 'get-url', remoteName], {
|
||||
cwd: context.repoPath,
|
||||
timeout: REMOTE_URL_PROBE_TIMEOUT_MS,
|
||||
...(context.wslDistro ? { wslDistro: context.wslDistro } : {})
|
||||
})
|
||||
return stdout
|
||||
return runGuardedGitHostProbe(
|
||||
localProbeHostKey(context),
|
||||
async () => {
|
||||
const { stdout } = await gitExecFileAsync(['remote', 'get-url', remoteName], {
|
||||
cwd: context.repoPath,
|
||||
timeout: REMOTE_URL_PROBE_TIMEOUT_MS,
|
||||
...(context.wslDistro ? { wslDistro: context.wslDistro } : {})
|
||||
})
|
||||
return stdout
|
||||
},
|
||||
isTransientGitProbeError
|
||||
)
|
||||
}
|
||||
|
||||
const TRANSIENT_PROBE_PATTERNS = [
|
||||
@@ -62,6 +98,11 @@ const TRANSIENT_PROBE_PATTERNS = [
|
||||
* report it as "no review": it is an unavailable result, not a negative one.
|
||||
*/
|
||||
export function isTransientGitProbeError(error: unknown): boolean {
|
||||
// Why: a probe the host's failure budget refused never reached the host, so it
|
||||
// stands in for exactly the deadline kill it was issued instead of.
|
||||
if (isGitHostProbeBlockedError(error)) {
|
||||
return true
|
||||
}
|
||||
// Why: an abort — this probe's deadline, or a caller cancelling — carries no
|
||||
// message a pattern could match, but it is the emptiest answer of all.
|
||||
if (
|
||||
|
||||
Reference in New Issue
Block a user