fix(desktop): harden gateway-switch commit ordering for stalled and overlapping switches (#93937)
Review follow-up. The two-phase switch closed the runtime-id leak for a single switch, but two shapes still broke the wipe→publish ordering: Overlapping: click A commits (wipes) and waits for its activation behind the profile-store mutex; click B finishes its dial and wipes too; then A's activation lands and publishes A AFTER B's destructive wipe, and A's endGatewaySwitch() dropped the (boolean) barrier while B was mid-commit. Now the wipe runs INSIDE the serialized section via a `beforeActivate` commit hook on ensureGatewayAgent, synchronously right before the socket is activated; the hook re-checks the click revision and declines when superseded, so a queued-then-superseded switch neither wipes nor activates. The barrier is token-owned (latest switch wins), and the mutex wait is a loop so waiters that wake together can't run interleaved. Stalled: a wedged spawn / ticket mint / IPC (the #93454 class) left the spinner up, swallowed later clicks on the same source, and — inside ensureGatewayAgent — latched the mutex and the barrier. Every await is now bounded (dial, activation, descriptor lookups, setLastUsed) via a shared withTimeout helper extracted from use-gateway-boot. A commit that timed out after the new source was already published counts as committed; one that stalls after the wipe lowers the barrier and repaints the source that is still active. Tests: queued-then-superseded switch, dial that never answers (and retry not swallowed), activation stalled after/before publication, barrier ownership, commit-hook ordering inside the mutex, bounded descriptor lookup releasing the mutex. All fail against the previous revision. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -7,6 +7,7 @@ import { HermesGateway } from '@/hermes'
|
||||
import { translateNow } from '@/i18n'
|
||||
import { desktopDefaultCwd } from '@/lib/desktop-fs'
|
||||
import { reconnectBackoffDelayMs } from '@/lib/reconnect-backoff'
|
||||
import { withTimeout } from '@/lib/with-timeout'
|
||||
import {
|
||||
$desktopBoot,
|
||||
applyDesktopBootProgress,
|
||||
@@ -114,23 +115,6 @@ const BOOT_RETRY_BASE_DELAY_MS = 2_000
|
||||
// already has its own connect timeout.
|
||||
const RECONNECT_ATTEMPT_TIMEOUT_MS = 20_000
|
||||
|
||||
function withTimeout<T>(promise: Promise<T>, ms: number, message: string): Promise<T> {
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
const timer = setTimeout(() => reject(new Error(message)), ms)
|
||||
|
||||
promise.then(
|
||||
value => {
|
||||
clearTimeout(timer)
|
||||
resolve(value)
|
||||
},
|
||||
err => {
|
||||
clearTimeout(timer)
|
||||
reject(err)
|
||||
}
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
/** Registry identity whose runtimes died with the primary connection. */
|
||||
export function primaryRuntimeConnectionId(connection: Pick<HermesConnection, 'connectionId' | 'mode'>): null | string {
|
||||
const connectionId = connection.connectionId?.trim()
|
||||
@@ -493,7 +477,7 @@ export function useGatewayBoot({
|
||||
|
||||
// Barrier up + machine-context reset + session wipe, in one synchronous
|
||||
// step — the shared commit point of every connection switch.
|
||||
beginGatewaySwitch()
|
||||
const switchToken = beginGatewaySwitch()
|
||||
clearReconnectTimer()
|
||||
clearBootRetryTimer()
|
||||
bootRetryAttempt = 0
|
||||
@@ -555,7 +539,7 @@ export function useGatewayBoot({
|
||||
setSessionsLoading(false)
|
||||
}
|
||||
} finally {
|
||||
endGatewaySwitch()
|
||||
endGatewaySwitch(switchToken)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
/** Rejection raised by withTimeout. The bounded work is NOT cancelled — the
|
||||
* caller decides what a straggler that settles later means. */
|
||||
export class TimeoutError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message)
|
||||
this.name = 'TimeoutError'
|
||||
}
|
||||
}
|
||||
|
||||
export function isTimeoutError(error: unknown): error is TimeoutError {
|
||||
return error instanceof TimeoutError
|
||||
}
|
||||
|
||||
/** Settle with `promise`, or reject with a TimeoutError after `ms`. */
|
||||
export function withTimeout<T>(promise: Promise<T>, ms: number, message: string): Promise<T> {
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
const timer = setTimeout(() => reject(new TimeoutError(message)), ms)
|
||||
|
||||
Promise.resolve(promise).then(
|
||||
value => {
|
||||
clearTimeout(timer)
|
||||
resolve(value)
|
||||
},
|
||||
err => {
|
||||
clearTimeout(timer)
|
||||
reject(err)
|
||||
}
|
||||
)
|
||||
})
|
||||
}
|
||||
@@ -3,6 +3,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import type { DesktopConnectionsRegistry } from '@/global'
|
||||
|
||||
import { deferred } from '../test/deferred'
|
||||
|
||||
const $activeGatewayProfile = atom('default')
|
||||
const $newChatProfile = atom<null | string>(null)
|
||||
const $showAllProfiles = atom(false)
|
||||
@@ -19,7 +21,14 @@ const $connection = atom<null | {
|
||||
const $activeSessionId = atom<null | string>(null)
|
||||
const $gatewaySwitching = atom(false)
|
||||
|
||||
const ensureGatewayAgent = vi.fn(async (_connectionId: null | string, _profile: string): Promise<void> => undefined)
|
||||
interface ActivationOptions {
|
||||
beforeActivate?: () => boolean
|
||||
}
|
||||
|
||||
const ensureGatewayAgent = vi.fn(
|
||||
async (_connectionId: null | string, _profile: string, _options?: ActivationOptions): Promise<void> => undefined
|
||||
)
|
||||
|
||||
const openGatewayAgent = vi.fn(async (_connectionId: string, _profile: string): Promise<void> => undefined)
|
||||
const refreshActiveProfile = vi.fn(async () => undefined)
|
||||
const requestFreshSession = vi.fn()
|
||||
@@ -27,15 +36,25 @@ const beforeConnectionSwitch = vi.fn()
|
||||
const wipeSessionListsForGatewaySwitch = vi.fn(() => $activeSessionId.set(null))
|
||||
|
||||
// Test double for the store's commit point with the real one's contract
|
||||
// (barrier → machine-context reset → wipe, synchronously); the real
|
||||
// implementation is covered by gateway-switch.test.ts.
|
||||
// (barrier → machine-context reset → wipe, synchronously; the barrier is
|
||||
// owned by the latest token); the real implementation is covered by
|
||||
// gateway-switch.test.ts.
|
||||
let latestSwitchToken = 0
|
||||
|
||||
const beginGatewaySwitch = vi.fn(() => {
|
||||
$gatewaySwitching.set(true)
|
||||
beforeConnectionSwitch()
|
||||
wipeSessionListsForGatewaySwitch()
|
||||
|
||||
return ++latestSwitchToken
|
||||
})
|
||||
|
||||
const endGatewaySwitch = vi.fn((token?: number) => {
|
||||
if (token === undefined || token === latestSwitchToken) {
|
||||
$gatewaySwitching.set(false)
|
||||
}
|
||||
})
|
||||
|
||||
const endGatewaySwitch = vi.fn(() => $gatewaySwitching.set(false))
|
||||
const recoverActiveSourceAfterFailedGatewaySwitch = vi.fn()
|
||||
|
||||
vi.mock('@/store/session', () => ({ $connection }))
|
||||
@@ -91,7 +110,13 @@ beforeEach(() => {
|
||||
$newChatProfile.set(null)
|
||||
$showAllProfiles.set(false)
|
||||
ensureGatewayAgent.mockReset()
|
||||
ensureGatewayAgent.mockImplementation(async connectionId => {
|
||||
// Mirrors the real door: the commit hook runs right before the activation
|
||||
// publishes, and a declined hook publishes nothing.
|
||||
ensureGatewayAgent.mockImplementation(async (connectionId, _profile, options) => {
|
||||
if (options?.beforeActivate && !options.beforeActivate()) {
|
||||
return
|
||||
}
|
||||
|
||||
$connection.set({
|
||||
connectionId: connectionId ?? undefined,
|
||||
mode: connectionId === 'local' ? 'local' : 'remote',
|
||||
@@ -134,7 +159,7 @@ describe('connection registry cache', () => {
|
||||
await initializeConnectionsRegistry()
|
||||
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledTimes(1)
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('homelab', 'default')
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('homelab', 'default', expect.anything())
|
||||
expect(setLastUsed).toHaveBeenCalledWith('homelab')
|
||||
})
|
||||
|
||||
@@ -154,7 +179,7 @@ describe('connection registry cache', () => {
|
||||
await initializeConnectionsRegistry()
|
||||
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledTimes(1)
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('homelab', 'default')
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('homelab', 'default', expect.anything())
|
||||
expect(setLastUsed).toHaveBeenCalledWith('homelab')
|
||||
})
|
||||
|
||||
@@ -179,7 +204,7 @@ describe('selectConnection', () => {
|
||||
await selectConnection('homelab')
|
||||
|
||||
expect(openGatewayAgent).toHaveBeenCalledWith('homelab', 'default')
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('homelab', 'default')
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('homelab', 'default', expect.anything())
|
||||
expect(beforeConnectionSwitch).toHaveBeenCalledTimes(1)
|
||||
expect(requestFreshSession).toHaveBeenCalledTimes(1)
|
||||
expect(wipeSessionListsForGatewaySwitch).toHaveBeenCalledTimes(1)
|
||||
@@ -204,7 +229,7 @@ describe('selectConnection', () => {
|
||||
|
||||
await selectConnection('local')
|
||||
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('local', 'default')
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('local', 'default', expect.anything())
|
||||
})
|
||||
|
||||
it('lets a later source choice win while an earlier dial is still pending', async () => {
|
||||
@@ -233,7 +258,7 @@ describe('selectConnection', () => {
|
||||
])
|
||||
// The superseded dial never activates: the user doesn't flip through
|
||||
// homelab on the way back to local, and only the winner commits.
|
||||
expect(ensureGatewayAgent.mock.calls).toEqual([['local', 'default']])
|
||||
expect(ensureGatewayAgent.mock.calls.map(call => [call[0], call[1]])).toEqual([['local', 'default']])
|
||||
expect(beginGatewaySwitch).toHaveBeenCalledTimes(1)
|
||||
expect(wipeSessionListsForGatewaySwitch).toHaveBeenCalledTimes(1)
|
||||
// Only the latest intent repaints the profile list.
|
||||
@@ -251,7 +276,7 @@ describe('selectConnection', () => {
|
||||
|
||||
await selectConnection('local')
|
||||
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('local', 'research')
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('local', 'research', expect.anything())
|
||||
})
|
||||
|
||||
it('does not remember a migrated v1 routing alias as a backend profile', async () => {
|
||||
@@ -262,7 +287,7 @@ describe('selectConnection', () => {
|
||||
|
||||
await selectConnection('homelab')
|
||||
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('homelab', 'default')
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('homelab', 'default', expect.anything())
|
||||
})
|
||||
|
||||
it('does not remember a stale startup profile under the resolved source', async () => {
|
||||
@@ -274,7 +299,7 @@ describe('selectConnection', () => {
|
||||
|
||||
await selectConnection('local')
|
||||
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('local', 'default')
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('local', 'default', expect.anything())
|
||||
})
|
||||
|
||||
it('keeps the current source usable when a dial fails: nothing is severed before the target is reachable', async () => {
|
||||
@@ -304,8 +329,11 @@ describe('selectConnection', () => {
|
||||
setConnectionsRegistry(registry)
|
||||
$connection.set({ connectionId: 'local', mode: 'local' })
|
||||
// The dial opened the socket but the activation was declined (source
|
||||
// edited/removed mid-switch): $connection never moves to homelab.
|
||||
ensureGatewayAgent.mockImplementationOnce(async () => undefined)
|
||||
// edited/removed mid-switch): the commit hook ran (wipe done), yet
|
||||
// $connection never moves to homelab.
|
||||
ensureGatewayAgent.mockImplementationOnce(async (_connectionId, _profile, options) => {
|
||||
options?.beforeActivate?.()
|
||||
})
|
||||
|
||||
await expect(selectConnection('homelab')).rejects.toThrow('did not become active')
|
||||
|
||||
@@ -353,16 +381,171 @@ describe('selectConnection', () => {
|
||||
// The old runtime id was already gone when homelab became visible, and
|
||||
// the barrier was up — nothing could pair 'a93bb39d' with the new backend.
|
||||
expect(published).toEqual([{ activeSessionId: null, connectionId: 'homelab', switching: true }])
|
||||
// dial → commit (barrier + reset + wipe) → activate, in that order.
|
||||
// dial → commit (barrier + reset + wipe, inside the activation's commit
|
||||
// hook) → publish, in that order.
|
||||
expect(openGatewayAgent).toHaveBeenCalledWith('homelab', 'default')
|
||||
expect(openGatewayAgent.mock.invocationCallOrder[0]).toBeLessThan(beginGatewaySwitch.mock.invocationCallOrder[0])
|
||||
expect(beginGatewaySwitch.mock.invocationCallOrder[0]).toBeLessThan(ensureGatewayAgent.mock.invocationCallOrder[0])
|
||||
expect(openGatewayAgent.mock.invocationCallOrder[0]).toBeLessThan(ensureGatewayAgent.mock.invocationCallOrder[0])
|
||||
expect(beginGatewaySwitch).toHaveBeenCalledTimes(1)
|
||||
expect(beforeConnectionSwitch).toHaveBeenCalledTimes(1)
|
||||
expect(endGatewaySwitch).toHaveBeenCalledTimes(1)
|
||||
expect($gatewaySwitching.get()).toBe(false)
|
||||
expect($activeSessionId.get()).toBeNull()
|
||||
})
|
||||
|
||||
it('a click that supersedes a QUEUED commit wins: the superseded switch neither wipes nor activates', async () => {
|
||||
// Both activations sit behind an in-flight profile/agent switch (the
|
||||
// profile store's mutex); their commit hooks only run once it settles.
|
||||
const mutex = deferred()
|
||||
|
||||
setConnectionsRegistry(registry)
|
||||
$connection.set({ connectionId: 'local', mode: 'local' })
|
||||
$activeSessionId.set('a93bb39d')
|
||||
ensureGatewayAgent.mockImplementation(async (connectionId, _profile, options) => {
|
||||
await mutex.promise
|
||||
|
||||
if (options?.beforeActivate && !options.beforeActivate()) {
|
||||
return
|
||||
}
|
||||
|
||||
$connection.set({
|
||||
connectionId: connectionId ?? undefined,
|
||||
mode: 'remote',
|
||||
profile: 'default',
|
||||
registryScoped: true
|
||||
})
|
||||
})
|
||||
|
||||
const published: string[] = []
|
||||
const off = $connection.listen(next => published.push(`${next?.connectionId}:active=${$activeSessionId.get()}`))
|
||||
|
||||
const first = selectConnection('homelab')
|
||||
await vi.waitFor(() => expect(ensureGatewayAgent).toHaveBeenCalledTimes(1))
|
||||
const second = selectConnection('work-vps')
|
||||
await vi.waitFor(() => expect(ensureGatewayAgent).toHaveBeenCalledTimes(2))
|
||||
|
||||
// Nothing is severed while both are still queued.
|
||||
expect(beginGatewaySwitch).not.toHaveBeenCalled()
|
||||
expect($activeSessionId.get()).toBe('a93bb39d')
|
||||
|
||||
mutex.resolve()
|
||||
await Promise.all([first, second])
|
||||
off()
|
||||
|
||||
// homelab stepped aside at its turn: no wipe, no publication, no error
|
||||
// UI; work-vps wiped once and is the only source ever published.
|
||||
expect(ensureGatewayAgent.mock.calls.map(call => call[0])).toEqual(['homelab', 'work-vps'])
|
||||
expect(beginGatewaySwitch).toHaveBeenCalledTimes(1)
|
||||
expect(published).toEqual(['work-vps:active=null'])
|
||||
expect($connection.get()?.connectionId).toBe('work-vps')
|
||||
expect(recoverActiveSourceAfterFailedGatewaySwitch).not.toHaveBeenCalled()
|
||||
expect(requestFreshSession).toHaveBeenCalledTimes(1)
|
||||
expect(setLastUsed).toHaveBeenCalledTimes(1)
|
||||
expect(setLastUsed).toHaveBeenCalledWith('work-vps')
|
||||
expect($gatewaySwitching.get()).toBe(false)
|
||||
expect($pendingConnectionId.get()).toBeNull()
|
||||
})
|
||||
|
||||
it('a dial that never answers times out: nothing severed, the click fails visibly, and the source can be retried', async () => {
|
||||
vi.useFakeTimers()
|
||||
|
||||
try {
|
||||
setConnectionsRegistry(registry)
|
||||
$connection.set({ connectionId: 'local', mode: 'local' })
|
||||
$activeSessionId.set('a93bb39d')
|
||||
openGatewayAgent.mockImplementationOnce(() => new Promise<void>(() => undefined))
|
||||
|
||||
const outcome = selectConnection('homelab').then(
|
||||
() => 'resolved',
|
||||
(error: Error) => error.message
|
||||
)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(20_000)
|
||||
|
||||
expect(await outcome).toMatch(/Timed out connecting to "Homelab"/)
|
||||
expect(ensureGatewayAgent).not.toHaveBeenCalled()
|
||||
expect(beginGatewaySwitch).not.toHaveBeenCalled()
|
||||
expect($activeSessionId.get()).toBe('a93bb39d')
|
||||
expect($gatewaySwitching.get()).toBe(false)
|
||||
expect($pendingConnectionId.get()).toBeNull()
|
||||
|
||||
// The stalled click does not poison the source: a retry is a real switch,
|
||||
// not a duplicate of the pending one.
|
||||
await selectConnection('homelab')
|
||||
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('homelab', 'default', expect.anything())
|
||||
expect($connection.get()?.connectionId).toBe('homelab')
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('an activation that published the new source but never settled counts as committed', async () => {
|
||||
vi.useFakeTimers()
|
||||
|
||||
try {
|
||||
setConnectionsRegistry(registry)
|
||||
$connection.set({ connectionId: 'local', mode: 'local' })
|
||||
// The socket activates and publishes synchronously; only the trailing
|
||||
// descriptor resync (an IPC) stalls.
|
||||
ensureGatewayAgent.mockImplementationOnce((connectionId, _profile, options) => {
|
||||
options?.beforeActivate?.()
|
||||
$connection.set({
|
||||
connectionId: connectionId ?? undefined,
|
||||
mode: 'remote',
|
||||
profile: 'default',
|
||||
registryScoped: true
|
||||
})
|
||||
|
||||
return new Promise<void>(() => undefined)
|
||||
})
|
||||
|
||||
const attempt = selectConnection('homelab')
|
||||
await vi.advanceTimersByTimeAsync(20_000)
|
||||
await attempt
|
||||
|
||||
expect($connection.get()?.connectionId).toBe('homelab')
|
||||
expect($gatewaySwitching.get()).toBe(false)
|
||||
expect(setLastUsed).toHaveBeenCalledWith('homelab')
|
||||
expect(requestFreshSession).toHaveBeenCalledTimes(1)
|
||||
expect(recoverActiveSourceAfterFailedGatewaySwitch).not.toHaveBeenCalled()
|
||||
expect($pendingConnectionId.get()).toBeNull()
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('an activation that stalls AFTER the wipe times out: barrier down, still-active source repainted', async () => {
|
||||
vi.useFakeTimers()
|
||||
|
||||
try {
|
||||
setConnectionsRegistry(registry)
|
||||
$connection.set({ connectionId: 'local', mode: 'local' })
|
||||
ensureGatewayAgent.mockImplementationOnce((_connectionId, _profile, options) => {
|
||||
options?.beforeActivate?.()
|
||||
|
||||
return new Promise<void>(() => undefined)
|
||||
})
|
||||
|
||||
const outcome = selectConnection('homelab').then(
|
||||
() => 'resolved',
|
||||
(error: Error) => error.message
|
||||
)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(20_000)
|
||||
|
||||
expect(await outcome).toMatch(/Timed out activating "Homelab"/)
|
||||
expect(beginGatewaySwitch).toHaveBeenCalledTimes(1)
|
||||
expect($gatewaySwitching.get()).toBe(false)
|
||||
expect(recoverActiveSourceAfterFailedGatewaySwitch).toHaveBeenCalledTimes(1)
|
||||
expect(requestFreshSession).toHaveBeenCalledTimes(1)
|
||||
expect(setLastUsed).not.toHaveBeenCalled()
|
||||
expect($connection.get()?.connectionId).toBe('local')
|
||||
expect($pendingConnectionId.get()).toBeNull()
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('boot-time restore leaves "All profiles" browse mode on (#93197)', async () => {
|
||||
// Fresh boot: nothing active yet, registry restores last-used. The
|
||||
// persisted showAllProfiles=true must survive the silent restore.
|
||||
@@ -371,7 +554,7 @@ describe('selectConnection', () => {
|
||||
|
||||
await initializeConnectionsRegistry()
|
||||
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('homelab', 'default')
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('homelab', 'default', expect.anything())
|
||||
expect($showAllProfiles.get()).toBe(true)
|
||||
})
|
||||
|
||||
@@ -382,7 +565,7 @@ describe('selectConnection', () => {
|
||||
|
||||
await selectConnection('homelab')
|
||||
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('homelab', 'default')
|
||||
expect(ensureGatewayAgent).toHaveBeenCalledWith('homelab', 'default', expect.anything())
|
||||
expect($showAllProfiles.get()).toBe(false)
|
||||
})
|
||||
|
||||
|
||||
@@ -2,9 +2,11 @@ import { atom, computed } from 'nanostores'
|
||||
|
||||
import type { DesktopConnectionsRegistry } from '@/global'
|
||||
import { persistStringRecord, storedStringRecord } from '@/lib/storage'
|
||||
import { isTimeoutError, withTimeout } from '@/lib/with-timeout'
|
||||
import {
|
||||
beginGatewaySwitch,
|
||||
endGatewaySwitch,
|
||||
type GatewaySwitchToken,
|
||||
recoverActiveSourceAfterFailedGatewaySwitch
|
||||
} from '@/store/gateway-switch'
|
||||
import {
|
||||
@@ -21,6 +23,14 @@ import { $connection } from '@/store/session'
|
||||
|
||||
const LAST_PROFILE_STORAGE_KEY = 'hermes.desktop.lastProfileByConnection'
|
||||
|
||||
// Every await of a source switch is bounded. A wedged spawn, ticket mint,
|
||||
// handshake or IPC (the #93454 class) must surface as a failed click — not a
|
||||
// spinner that also swallows every later click on the same source, and never
|
||||
// a barrier left up or a wipe left unpainted.
|
||||
const SWITCH_DIAL_TIMEOUT_MS = 20_000
|
||||
const SWITCH_COMMIT_TIMEOUT_MS = 20_000
|
||||
const SWITCH_REMEMBER_TIMEOUT_MS = 5_000
|
||||
|
||||
export const $connectionsRegistry = atom<DesktopConnectionsRegistry | null>(null)
|
||||
|
||||
// Use only the resolved descriptor identity Electron publishes. `primary`
|
||||
@@ -109,11 +119,17 @@ async function rememberConnection(connectionId: string): Promise<void> {
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await setLastUsed(connectionId)
|
||||
const result = await withTimeout(
|
||||
setLastUsed(connectionId),
|
||||
SWITCH_REMEMBER_TIMEOUT_MS,
|
||||
'Timed out remembering the last-used connection'
|
||||
)
|
||||
|
||||
setConnectionsRegistry(result.registry)
|
||||
} catch {
|
||||
// The source is already usable. A read-only/full userData directory must
|
||||
// not turn a successful backend switch into a false connection failure.
|
||||
// The source is already usable. A read-only/full userData directory (or
|
||||
// a stalled IPC) must not turn a successful backend switch into a false
|
||||
// connection failure.
|
||||
}
|
||||
}
|
||||
|
||||
@@ -169,13 +185,21 @@ export async function initializeConnectionsRegistry(): Promise<DesktopConnection
|
||||
* Two phases, same commit contract as a Settings → Gateway apply (softSwitch):
|
||||
* 1. Dial the target WITHOUT activating it. The previous source stays fully
|
||||
* bound and painted, so a dead target fails with nothing lost.
|
||||
* 2. beginGatewaySwitch() — barrier up, machine-context reset, session
|
||||
* bindings wiped — then activate the already-open socket. Nothing awaits
|
||||
* between the wipe and the publication, so no route/session effect can
|
||||
* observe the new source while $activeSessionId still names the previous
|
||||
* backend's runtime. Activating first and wiping after (across an IPC
|
||||
* round-trip) is exactly how that id leaked to the new backend and came
|
||||
* back as "session not found" (#93937).
|
||||
* 2. Commit: beginGatewaySwitch() — barrier up, machine-context reset,
|
||||
* session bindings wiped — then activate the already-open socket. The
|
||||
* wipe runs inside the activation's serialized section, synchronously
|
||||
* before the publication, so no route/session effect can observe the new
|
||||
* source while $activeSessionId still names the previous backend's
|
||||
* runtime. Activating first and wiping after (across an IPC round-trip)
|
||||
* is exactly how that id leaked to the new backend and came back as
|
||||
* "session not found" (#93937).
|
||||
*
|
||||
* Overlap and stalls: clicks can supersede a switch at any point. A switch
|
||||
* superseded while queued behind another activation declines its commit —
|
||||
* no wipe, no activation — so the winner's wipe is always the one that
|
||||
* precedes the final publication, and the barrier is owned by the latest
|
||||
* switch. Every await is bounded; a commit that stalls after the wipe lowers
|
||||
* the barrier and repaints the source that is still active.
|
||||
*/
|
||||
export async function selectConnection(connectionId: string): Promise<void> {
|
||||
const registry = $connectionsRegistry.get()
|
||||
@@ -224,12 +248,20 @@ export async function selectConnection(connectionId: string): Promise<void> {
|
||||
const revision = ++switchRevision
|
||||
pendingTarget = targetKey
|
||||
$pendingConnectionId.set(connectionId)
|
||||
// Set by the commit hook once THIS switch has wiped — i.e. it owns the
|
||||
// barrier and, if the commit then fails, owes the still-active source a
|
||||
// repaint. Null while queued, or if it stepped aside before its turn.
|
||||
let token = null as GatewaySwitchToken | null
|
||||
|
||||
try {
|
||||
// Phase 1 — open the target's socket; the active route is untouched.
|
||||
// Always use the explicit registry route. `local` must mean This device,
|
||||
// and a registry primary can differ from a legacy per-profile override.
|
||||
await openGatewayAgent(connectionId, targetProfile)
|
||||
await withTimeout(
|
||||
openGatewayAgent(connectionId, targetProfile),
|
||||
SWITCH_DIAL_TIMEOUT_MS,
|
||||
`Timed out connecting to "${targetConnection.label}".`
|
||||
)
|
||||
|
||||
// A newer click owns the switch from here on. The superseded dial never
|
||||
// activates, so the user doesn't flip through it on the way to the source
|
||||
@@ -238,27 +270,51 @@ export async function selectConnection(connectionId: string): Promise<void> {
|
||||
return
|
||||
}
|
||||
|
||||
// Phase 2 — commit: sever the previous backend's bindings FIRST, then
|
||||
// activate. Synchronous from the wipe to the publication (the socket is
|
||||
// already open), and behind the barrier until the descriptor lands.
|
||||
beginGatewaySwitch()
|
||||
|
||||
// Phase 2 — commit. The hook runs inside the activation's serialized
|
||||
// section, right before the socket is activated: sever the previous
|
||||
// backend's bindings, then publish, with nothing in between. A click that
|
||||
// superseded this switch while it was queued makes the hook decline —
|
||||
// neither wipe nor activation — so the user never flips through it.
|
||||
try {
|
||||
await ensureGatewayAgent(connectionId, targetProfile)
|
||||
try {
|
||||
await withTimeout(
|
||||
ensureGatewayAgent(connectionId, targetProfile, {
|
||||
beforeActivate: () => {
|
||||
if (revision !== switchRevision) {
|
||||
return false
|
||||
}
|
||||
|
||||
token = beginGatewaySwitch()
|
||||
|
||||
return true
|
||||
}
|
||||
}),
|
||||
SWITCH_COMMIT_TIMEOUT_MS,
|
||||
`Timed out activating "${targetConnection.label}".`
|
||||
)
|
||||
} catch (error) {
|
||||
// The socket is activated and its descriptor published synchronously;
|
||||
// only the best-effort descriptor resync trails it. A commit that timed
|
||||
// out AFTER the new source became active has landed — the straggler is
|
||||
// fail-open and cannot undo it.
|
||||
if (!isTimeoutError(error) || $connection.get()?.connectionId !== connectionId) {
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
if (revision !== switchRevision) {
|
||||
return
|
||||
}
|
||||
|
||||
if ($connection.get()?.connectionId !== connectionId) {
|
||||
throw new Error(`Connection "${targetConnection.label}" did not become active.`)
|
||||
}
|
||||
} catch (error) {
|
||||
// The wipe already happened but the previous source is still the active
|
||||
// one, and nothing reactive re-pulls its lists (no scope moved). Repaint
|
||||
// it and land on a fresh draft there, matching what a failed Settings
|
||||
// apply leaves behind.
|
||||
recoverActiveSourceAfterFailedGatewaySwitch()
|
||||
requestFreshSession()
|
||||
throw error
|
||||
} finally {
|
||||
endGatewaySwitch()
|
||||
// Lower the barrier the moment the commit settles — before the
|
||||
// bookkeeping awaits below — but only if this switch still owns it.
|
||||
if (token !== null) {
|
||||
endGatewaySwitch(token)
|
||||
}
|
||||
}
|
||||
|
||||
// A newer click owns the final refresh. Serialized gateway activation
|
||||
@@ -277,6 +333,15 @@ export async function selectConnection(connectionId: string): Promise<void> {
|
||||
}
|
||||
} catch (error) {
|
||||
if (revision === switchRevision) {
|
||||
if (token !== null) {
|
||||
// This switch wiped for a commit that never landed. The previous
|
||||
// source is still the active one, and nothing reactive re-pulls its
|
||||
// lists (no scope moved): repaint it and land on a fresh draft there,
|
||||
// matching what a failed Settings apply leaves behind.
|
||||
recoverActiveSourceAfterFailedGatewaySwitch()
|
||||
requestFreshSession()
|
||||
}
|
||||
|
||||
throw error
|
||||
}
|
||||
} finally {
|
||||
|
||||
@@ -132,6 +132,22 @@ describe('beginGatewaySwitch / endGatewaySwitch — the shared switch commit poi
|
||||
off()
|
||||
})
|
||||
|
||||
it('the barrier belongs to the LATEST switch: an older switch ending mid-commit of a newer one is a no-op', () => {
|
||||
const older = beginGatewaySwitch()
|
||||
const newer = beginGatewaySwitch()
|
||||
|
||||
endGatewaySwitch(older)
|
||||
expect($gatewaySwitching.get()).toBe(true)
|
||||
|
||||
endGatewaySwitch(newer)
|
||||
expect($gatewaySwitching.get()).toBe(false)
|
||||
|
||||
// Host teardown forces it down regardless of ownership.
|
||||
beginGatewaySwitch()
|
||||
endGatewaySwitch()
|
||||
expect($gatewaySwitching.get()).toBe(false)
|
||||
})
|
||||
|
||||
it('still severs the bindings when no lifecycle is registered (windows that never mount the boot hook)', () => {
|
||||
beginGatewaySwitch()
|
||||
|
||||
|
||||
@@ -51,6 +51,11 @@ export interface GatewaySwitchLifecycle {
|
||||
|
||||
let switchLifecycle: GatewaySwitchLifecycle | null = null
|
||||
|
||||
/** Ownership handle returned by beginGatewaySwitch; see endGatewaySwitch. */
|
||||
export type GatewaySwitchToken = number
|
||||
|
||||
let latestSwitchToken = 0
|
||||
|
||||
export function registerGatewaySwitchLifecycle(lifecycle: GatewaySwitchLifecycle): () => void {
|
||||
switchLifecycle = lifecycle
|
||||
|
||||
@@ -71,14 +76,26 @@ export function registerGatewaySwitchLifecycle(lifecycle: GatewaySwitchLifecycle
|
||||
* $activeSessionId still named the previous backend's runtime and sent that id
|
||||
* to a backend that had never minted it — "session not found" (#93937).
|
||||
*/
|
||||
export function beginGatewaySwitch(): void {
|
||||
export function beginGatewaySwitch(): GatewaySwitchToken {
|
||||
const token = ++latestSwitchToken
|
||||
$gatewaySwitching.set(true)
|
||||
switchLifecycle?.beforeConnectionSwitch()
|
||||
wipeSessionListsForGatewaySwitch()
|
||||
|
||||
return token
|
||||
}
|
||||
|
||||
/** Lower the barrier once the switch has committed (or failed). */
|
||||
export function endGatewaySwitch(): void {
|
||||
/**
|
||||
* Lower the barrier once the switch that owns it has committed (or failed).
|
||||
* Switches overlap — a click can supersede one that is mid-commit — and the
|
||||
* barrier belongs to the LATEST one: an older switch ending is a no-op while a
|
||||
* newer one is still in flight. No token = force down (host teardown).
|
||||
*/
|
||||
export function endGatewaySwitch(token?: GatewaySwitchToken): void {
|
||||
if (token !== undefined && token !== latestSwitchToken) {
|
||||
return
|
||||
}
|
||||
|
||||
$gatewaySwitching.set(false)
|
||||
}
|
||||
|
||||
|
||||
@@ -193,3 +193,85 @@ describe('ensureGatewayAgent shares the gatewaySwitch mutex with profile switche
|
||||
expect($activeGatewayProfile.get()).toBe('worker')
|
||||
})
|
||||
})
|
||||
|
||||
describe('ensureGatewayAgent commit hook (beforeActivate) — the Sessions switcher door (#93937)', () => {
|
||||
it('runs the hook inside the serialized section, right before the activation; a declined hook activates nothing', async () => {
|
||||
const profileGate = deferred()
|
||||
const order: string[] = []
|
||||
|
||||
ensureGatewayForProfile.mockImplementation(async (profile: string) => {
|
||||
order.push(`profile:${profile}`)
|
||||
await profileGate.promise
|
||||
})
|
||||
ensureGatewayForAgent.mockImplementation(async (_connectionId, profile) => {
|
||||
order.push(`agent:${profile}`)
|
||||
|
||||
return true
|
||||
})
|
||||
getConnection.mockResolvedValue(localConn({ profile: 'worker' }))
|
||||
getConnectionFor.mockResolvedValue(agentConn())
|
||||
|
||||
const profileSwitch = ensureGatewayProfile('worker')
|
||||
await Promise.resolve()
|
||||
|
||||
const declined = ensureGatewayAgent('homelab', 'research', {
|
||||
beforeActivate: () => {
|
||||
order.push('hook:declined')
|
||||
|
||||
return false
|
||||
}
|
||||
})
|
||||
|
||||
await Promise.resolve()
|
||||
// The hook waits for its turn — it must see the world AFTER the earlier
|
||||
// switch published, never before.
|
||||
expect(order).toEqual(['profile:worker'])
|
||||
|
||||
profileGate.resolve()
|
||||
await profileSwitch
|
||||
await declined
|
||||
|
||||
expect(order).toEqual(['profile:worker', 'hook:declined'])
|
||||
expect(ensureGatewayForAgent).not.toHaveBeenCalled()
|
||||
expect(getConnectionFor).not.toHaveBeenCalled()
|
||||
expect($activeGatewayProfile.get()).toBe('worker')
|
||||
|
||||
// An accepted hook runs synchronously before the activation starts.
|
||||
await ensureGatewayAgent('homelab', 'research', {
|
||||
beforeActivate: () => {
|
||||
order.push('hook:accepted')
|
||||
|
||||
return true
|
||||
}
|
||||
})
|
||||
|
||||
expect(order).toEqual(['profile:worker', 'hook:declined', 'hook:accepted', 'agent:research'])
|
||||
expect($activeGatewayProfile.get()).toBe('research')
|
||||
expect($connection.get()?.profile).toBe('research')
|
||||
})
|
||||
|
||||
it('a descriptor lookup that never answers is bounded, fails open, and releases the mutex', async () => {
|
||||
vi.useFakeTimers()
|
||||
|
||||
try {
|
||||
getConnectionFor.mockImplementationOnce(() => new Promise<HermesConnection>(() => undefined))
|
||||
|
||||
const activation = ensureGatewayAgent('homelab', 'research')
|
||||
await vi.advanceTimersByTimeAsync(20_000)
|
||||
await activation
|
||||
|
||||
// Activated; the previous descriptor is kept (fail open), not nulled.
|
||||
expect($activeGatewayProfile.get()).toBe('research')
|
||||
expect($connection.get()?.mode).toBe('local')
|
||||
|
||||
// The next switch is not stuck behind the stalled lookup.
|
||||
getConnectionFor.mockResolvedValue(agentConn({ profile: 'writer' }))
|
||||
await ensureGatewayAgent('homelab', 'writer')
|
||||
|
||||
expect($activeGatewayProfile.get()).toBe('writer')
|
||||
expect($connection.get()?.profile).toBe('writer')
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
@@ -12,6 +12,7 @@ import {
|
||||
storedStringArray,
|
||||
storedStringRecord
|
||||
} from '@/lib/storage'
|
||||
import { withTimeout } from '@/lib/with-timeout'
|
||||
import { invalidateCronModelImpactScopeState } from '@/store/cron-model-impact-scope'
|
||||
import {
|
||||
$gateway,
|
||||
@@ -288,6 +289,13 @@ export function prewarmProfileBackend(name: string): void {
|
||||
|
||||
let gatewaySwitch: Promise<void> | null = null
|
||||
|
||||
// Descriptor lookups are IPC round-trips into Electron main. A wedged main
|
||||
// (the #93454 class: a ticket mint that never answers) must not latch the
|
||||
// gatewaySwitch mutex — and, through it, every later profile/source switch
|
||||
// and the switch barrier — so they are bounded and fail open like any other
|
||||
// lookup failure.
|
||||
const DESCRIPTOR_LOOKUP_TIMEOUT_MS = 20_000
|
||||
|
||||
// The target profile's connection descriptor (mode / baseUrl / …), resolved
|
||||
// CONCURRENTLY with the socket work so the switch can publish the profile
|
||||
// pointer and $connection in one frame. Without this, $connection seeds from
|
||||
@@ -311,7 +319,11 @@ async function resolveConnectionForProfile(profile: string): Promise<HermesConne
|
||||
}
|
||||
|
||||
try {
|
||||
return await getConnection(profile)
|
||||
return await withTimeout(
|
||||
getConnection(profile),
|
||||
DESCRIPTOR_LOOKUP_TIMEOUT_MS,
|
||||
`Timed out resolving the connection descriptor for profile "${profile}"`
|
||||
)
|
||||
} catch (err) {
|
||||
console.warn(`[profile] descriptor lookup for "${profile}" failed; keeping the previous connection`, err)
|
||||
|
||||
@@ -398,7 +410,11 @@ async function resolveConnectionForAgent(connectionId: string, profile: string):
|
||||
}
|
||||
|
||||
try {
|
||||
return await getConnectionFor({ connectionId, profile })
|
||||
return await withTimeout(
|
||||
getConnectionFor({ connectionId, profile }),
|
||||
DESCRIPTOR_LOOKUP_TIMEOUT_MS,
|
||||
`Timed out resolving the connection descriptor for agent "${connectionId}:${profile}"`
|
||||
)
|
||||
} catch (err) {
|
||||
console.warn(
|
||||
`[profile] descriptor lookup for agent "${connectionId}:${profile}" failed; keeping the previous connection`,
|
||||
@@ -442,7 +458,25 @@ export async function openGatewayAgent(connectionId: string, profile: string): P
|
||||
// order and leave the EARLIER setActive() as the last write.
|
||||
// Only a null connectionId falls through to the legacy profile path. Explicit
|
||||
// `local` is a registry identity and must use the genuinely-local route.
|
||||
export async function ensureGatewayAgent(connectionId: null | string, profile: string): Promise<void> {
|
||||
//
|
||||
// `beforeActivate` is the commit hook of the two-phase source switch
|
||||
// (store/connections selectConnection). It runs INSIDE the serialized
|
||||
// section, synchronously right before the socket is activated — i.e. after
|
||||
// every earlier switch has published and before this one does — so the caller
|
||||
// can sever the previous backend's session bindings at exactly that point
|
||||
// (#93937). Returning false declines: nothing is activated or published, the
|
||||
// mutex is released. That is how a switch superseded while queued behind
|
||||
// another one steps aside without a destructive wipe. Not consulted on the
|
||||
// null-connectionId profile fallthrough.
|
||||
export interface EnsureGatewayAgentOptions {
|
||||
beforeActivate?: () => boolean
|
||||
}
|
||||
|
||||
export async function ensureGatewayAgent(
|
||||
connectionId: null | string,
|
||||
profile: string,
|
||||
{ beforeActivate }: EnsureGatewayAgentOptions = {}
|
||||
): Promise<void> {
|
||||
const target = normalizeProfileKey(profile)
|
||||
const connection = (connectionId ?? '').trim() || null
|
||||
|
||||
@@ -450,13 +484,20 @@ export async function ensureGatewayAgent(connectionId: null | string, profile: s
|
||||
return ensureGatewayProfile(target)
|
||||
}
|
||||
|
||||
// Serialize against any in-flight profile/agent switch (shared mutex).
|
||||
if (gatewaySwitch) {
|
||||
// Serialize against any in-flight profile/agent switch (shared mutex). A
|
||||
// loop, not a single await: several waiters wake from the same settled
|
||||
// switch, and the first to re-acquire starts a new one the rest must also
|
||||
// wait out — otherwise two overlapping activations run interleaved.
|
||||
while (gatewaySwitch) {
|
||||
await gatewaySwitch.catch(() => undefined)
|
||||
}
|
||||
|
||||
$gatewaySwapTarget.set(target)
|
||||
gatewaySwitch = (async () => {
|
||||
if (beforeActivate && !beforeActivate()) {
|
||||
return
|
||||
}
|
||||
|
||||
// Descriptor resolves concurrently with the dial, same as the profile
|
||||
// path, so no await sits between the activation and the publication.
|
||||
const [descriptor, activated] = await Promise.all([
|
||||
|
||||
Reference in New Issue
Block a user