diff --git a/src/main/daemon/daemon-fresh-session-routing.ts b/src/main/daemon/daemon-fresh-session-routing.ts index 057a9469145..f7888411378 100644 --- a/src/main/daemon/daemon-fresh-session-routing.ts +++ b/src/main/daemon/daemon-fresh-session-routing.ts @@ -3,6 +3,7 @@ import { DaemonSessionOwnerUnknownError, type DaemonSessionRoute, DaemonSessionUnavailableError, + isCurrentOwnedDaemonSessionRoute, sameDaemonIncarnation } from './daemon-session-route' @@ -44,3 +45,34 @@ export function assertFreshDaemonSessionAvailable( } throw new DaemonSessionOwnerUnknownError(sessionId) } + +export function recordFreshOwnedDaemonSession( + route: DaemonSessionRoute | undefined, + owner: DaemonPtyAdapter, + transfer: () => void, + setRoute: (route: DaemonSessionRoute) => void, + recordOwned: () => void +): void { + if ( + !route || + route.state === 'unavailable' || + (route.state === 'owned' && route.owner === owner && !isCurrentOwnedDaemonSessionRoute(route)) + ) { + transfer() + return + } + if ( + route.state === 'ambiguous' && + route.candidates.has(owner) && + route.candidates.get(owner) === null + ) { + const incarnation = owner.getLastAuthenticatedDaemonIdentity() + if (incarnation) { + const candidates = new Map(route.candidates) + candidates.set(owner, incarnation) + setRoute({ state: 'ambiguous', candidates }) + return + } + } + recordOwned() +} diff --git a/src/main/daemon/daemon-health.test.ts b/src/main/daemon/daemon-health.test.ts index 6eb8d1c8cb8..c4302427b64 100644 --- a/src/main/daemon/daemon-health.test.ts +++ b/src/main/daemon/daemon-health.test.ts @@ -443,6 +443,150 @@ describe('killStaleDaemon pid identity guards', () => { } }) + it('preserves the owned pidfile when SIGTERM is rejected', async () => { + if (process.platform === 'win32') { + return + } + + const child = spawn( + process.execPath, + ['-e', 'setInterval(()=>{},1000)', 'daemon-entry', socketPath, tokenPath], + { stdio: 'ignore' } + ) + const pidPath = getDaemonPidPath(dir) + const pidfile = serializeDaemonPidFile({ + pid: child.pid!, + startedAtMs: null, + launchNonce: 'signal-rejected' + }) + writeFileSync(pidPath, pidfile, { mode: 0o600 }) + const realKill = process.kill.bind(process) + const killSpy = vi.spyOn(process, 'kill').mockImplementation((pid, signal) => { + if (pid === child.pid && signal === 'SIGTERM') { + throw Object.assign(new Error('operation not permitted'), { code: 'EPERM' }) + } + return realKill(pid, signal) + }) + + try { + await expect(killStaleDaemon(dir, socketPath, tokenPath)).resolves.toBe(false) + expect(readFileSync(pidPath, 'utf8')).toBe(pidfile) + expect(child.exitCode).toBeNull() + } finally { + killSpy.mockRestore() + const exited = + child.exitCode === null + ? new Promise((resolve) => child.once('exit', () => resolve())) + : Promise.resolve() + if (child.exitCode === null) { + child.kill('SIGKILL') + } + await exited + } + }) + + it('preserves the owned pidfile when SIGKILL is rejected', async () => { + if (process.platform === 'win32') { + return + } + + const child = spawn( + process.execPath, + ['-e', 'setInterval(()=>{},1000)', 'daemon-entry', socketPath, tokenPath], + { stdio: 'ignore' } + ) + const pidPath = getDaemonPidPath(dir) + const pidfile = serializeDaemonPidFile({ + pid: child.pid!, + startedAtMs: null, + launchNonce: 'kill-rejected' + }) + writeFileSync(pidPath, pidfile, { mode: 0o600 }) + const realKill = process.kill.bind(process) + let observeSigterm!: () => void + const sigtermObserved = new Promise((resolve) => { + observeSigterm = resolve + }) + const killSpy = vi.spyOn(process, 'kill').mockImplementation((pid, signal) => { + if (pid === child.pid && signal === 'SIGTERM') { + observeSigterm() + return true + } + if (pid === child.pid && signal === 'SIGKILL') { + throw Object.assign(new Error('operation not permitted'), { code: 'EPERM' }) + } + return realKill(pid, signal) + }) + + vi.useFakeTimers() + try { + const killing = killStaleDaemon(dir, socketPath, tokenPath) + await sigtermObserved + await vi.advanceTimersByTimeAsync(3_100) + await expect(killing).resolves.toBe(false) + expect(readFileSync(pidPath, 'utf8')).toBe(pidfile) + expect(child.exitCode).toBeNull() + } finally { + vi.useRealTimers() + killSpy.mockRestore() + const exited = + child.exitCode === null + ? new Promise((resolve) => child.once('exit', () => resolve())) + : Promise.resolve() + if (child.exitCode === null) { + child.kill('SIGKILL') + } + await exited + } + }) + + it('preserves the owned pidfile when the post-SIGTERM liveness probe is rejected', async () => { + if (process.platform === 'win32') { + return + } + + const child = spawn( + process.execPath, + ['-e', 'setInterval(()=>{},1000)', 'daemon-entry', socketPath, tokenPath], + { stdio: 'ignore' } + ) + const pidPath = getDaemonPidPath(dir) + const pidfile = serializeDaemonPidFile({ + pid: child.pid!, + startedAtMs: null, + launchNonce: 'probe-rejected' + }) + writeFileSync(pidPath, pidfile, { mode: 0o600 }) + const realKill = process.kill.bind(process) + let termAccepted = false + const killSpy = vi.spyOn(process, 'kill').mockImplementation((pid, signal) => { + if (pid === child.pid && signal === 'SIGTERM') { + termAccepted = true + return true + } + if (pid === child.pid && signal === 0 && termAccepted) { + throw Object.assign(new Error('operation not permitted'), { code: 'EPERM' }) + } + return realKill(pid, signal) + }) + + try { + await expect(killStaleDaemon(dir, socketPath, tokenPath)).resolves.toBe(false) + expect(readFileSync(pidPath, 'utf8')).toBe(pidfile) + expect(child.exitCode).toBeNull() + } finally { + killSpy.mockRestore() + const exited = + child.exitCode === null + ? new Promise((resolve) => child.once('exit', () => resolve())) + : Promise.resolve() + if (child.exitCode === null) { + child.kill('SIGKILL') + } + await exited + } + }) + it('preserves replacement identity files when the pid changes ownership during escalation', async () => { if (process.platform === 'win32') { return diff --git a/src/main/daemon/daemon-health.ts b/src/main/daemon/daemon-health.ts index 5b7336f4007..4656ea827a5 100644 --- a/src/main/daemon/daemon-health.ts +++ b/src/main/daemon/daemon-health.ts @@ -35,6 +35,10 @@ export const E2E_FORCE_DAEMON_HEALTH_UNREACHABLE_ENV = 'ORCA_E2E_FORCE_DAEMON_HE // tolerance keeps the guard effective without false mismatches. const WIN32_START_TIME_TOLERANCE_MS = 10_000 +function isNoSuchProcessError(error: unknown): boolean { + return typeof error === 'object' && error !== null && 'code' in error && error.code === 'ESRCH' +} + // 'rejected' means the daemon answered and refused the handshake (bad token, // foreign protocol) — it can never be adopted, unlike 'unreachable', which // also covers a live-but-wedged daemon that simply missed the RPC budget. @@ -678,13 +682,23 @@ export async function killStaleDaemon( identifiedPid = parsedPid.pid identifiedLaunchNonce = parsedPid.launchNonce const { pid, startedAtMs } = parsedPid - process.kill(pid, 'SIGTERM') - const deadline = Date.now() + KILL_WAIT_MS let exited = false - while (Date.now() < deadline) { + try { + process.kill(pid, 'SIGTERM') + } catch (error) { + if (!isNoSuchProcessError(error)) { + throw error + } + exited = true + } + const deadline = Date.now() + KILL_WAIT_MS + while (!exited && Date.now() < deadline) { try { process.kill(pid, 0) - } catch { + } catch (error) { + if (!isNoSuchProcessError(error)) { + throw error + } exited = true break } @@ -704,8 +718,8 @@ export async function killStaleDaemon( try { process.kill(pid, 'SIGKILL') exited = true - } catch { - // Already dead + } catch (error) { + exited = isNoSuchProcessError(error) } } } @@ -715,7 +729,12 @@ export async function killStaleDaemon( // PID file missing or process already dead } - if (identifiedDaemon && identifiedPid !== null && identifiedLaunchNonce !== null) { + if ( + killedDaemon && + identifiedDaemon && + identifiedPid !== null && + identifiedLaunchNonce !== null + ) { unlinkOwnedDaemonPidFile(pidPath, identifiedPid, identifiedLaunchNonce) } diff --git a/src/main/daemon/daemon-history-handoff-routing.ts b/src/main/daemon/daemon-history-handoff-routing.ts new file mode 100644 index 00000000000..e943fa391ce --- /dev/null +++ b/src/main/daemon/daemon-history-handoff-routing.ts @@ -0,0 +1,20 @@ +import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import type { DaemonSessionRoute } from './daemon-session-route' + +export function createDaemonHistoryHandoffRoute( + owner: DaemonPtyAdapter, + target: DaemonPtyAdapter +): DaemonSessionRoute { + return { + state: 'unavailable', + owner, + incarnation: owner.getLastAuthenticatedDaemonIdentity(), + historyHandoffTarget: target + } +} + +export function daemonHistoryHandoffTarget( + route: DaemonSessionRoute | undefined +): DaemonPtyAdapter | undefined { + return route?.state === 'unavailable' ? route.historyHandoffTarget : undefined +} diff --git a/src/main/daemon/daemon-pty-router-history-handoff.test.ts b/src/main/daemon/daemon-pty-router-history-handoff.test.ts index a53a4b29b63..92ba3c03480 100644 --- a/src/main/daemon/daemon-pty-router-history-handoff.test.ts +++ b/src/main/daemon/daemon-pty-router-history-handoff.test.ts @@ -111,6 +111,7 @@ describe('DaemonPtyRouter history handoff', () => { const request = vi.spyOn(client, 'request') await router.shutdown(sessionId, { immediate: true, keepHistory: true }) + expect(router.getSessionRouteState(sessionId)).toBe('unavailable') const restored = await router.spawn({ sessionId, cols: 400, rows: 24 }) expect(restored.coldRestore?.scrollback).toContain(marker) diff --git a/src/main/daemon/daemon-pty-router-routing-safety-fixture.ts b/src/main/daemon/daemon-pty-router-routing-safety-fixture.ts new file mode 100644 index 00000000000..382f88b7026 --- /dev/null +++ b/src/main/daemon/daemon-pty-router-routing-safety-fixture.ts @@ -0,0 +1,103 @@ +import { vi } from 'vitest' +import type { PtyProcessInfo, PtySpawnOptions } from '../providers/types' +import type { DaemonEndpointIdentity } from './daemon-hello-protocol' +import type { DaemonPtyAdapter } from './daemon-pty-adapter' + +export type AdapterHarness = { + adapter: DaemonPtyAdapter + emitData: (sessionId: string, data: string) => void + emitIdentityChange: (previous: DaemonEndpointIdentity, current: DaemonEndpointIdentity) => void + emitExit: (sessionId: string, code: number) => void + setIdentity: (identity: DaemonEndpointIdentity) => void +} + +export function identity(label: string, pid: number): DaemonEndpointIdentity { + return { + pid, + startedAtMs: pid * 1_000, + launchNonce: `${label}-${pid}` + } +} + +export function createAdapter( + label: string, + sessions: string[] = [], + reconcileResult: { alive: string[]; killed: string[] } = { alive: [], killed: [] } +): AdapterHarness { + let daemonIdentity = identity(label, label === 'current' ? 20 : 10) + const identityListeners: ((event: { + previous: DaemonEndpointIdentity + current: DaemonEndpointIdentity + }) => void)[] = [] + const dataListeners: ((payload: { id: string; data: string }) => void)[] = [] + const exitListeners: ((payload: { id: string; code: number }) => void)[] = [] + const adapter = { + protocolVersion: label === 'current' ? 30 : 29, + getLastAuthenticatedDaemonIdentity: vi.fn(() => ({ ...daemonIdentity })), + matchesLastAuthenticatedDaemonIdentity: vi.fn( + (candidate: DaemonEndpointIdentity | null) => + candidate !== null && + candidate.pid === daemonIdentity.pid && + candidate.startedAtMs === daemonIdentity.startedAtMs && + candidate.launchNonce === daemonIdentity.launchNonce + ), + onDaemonIdentityChanged: vi.fn( + ( + listener: (event: { + previous: DaemonEndpointIdentity + current: DaemonEndpointIdentity + }) => void + ) => { + identityListeners.push(listener) + return () => {} + } + ), + listProcesses: vi.fn( + async (): Promise => sessions.map((id) => ({ id, cwd: '', title: label })) + ), + spawn: vi.fn(async (opts: PtySpawnOptions) => ({ + id: opts.sessionId ?? `${label}-fresh` + })), + probePtyLiveness: vi.fn(async (id: string) => sessions.includes(id)), + hasPty: vi.fn((id: string) => sessions.includes(id)), + shutdown: vi.fn(async () => {}), + sendSignal: vi.fn(async () => {}), + getBufferSnapshot: vi.fn(async () => null), + ackColdRestore: vi.fn(), + reconcileOnStartup: vi.fn(async () => reconcileResult), + onData: vi.fn((listener: (payload: { id: string; data: string }) => void) => { + dataListeners.push(listener) + return () => {} + }), + onExit: vi.fn((listener: (payload: { id: string; code: number }) => void) => { + exitListeners.push(listener) + return () => {} + }), + onBackgroundStreamEvent: vi.fn(() => () => {}), + onWriteUnavailable: vi.fn(() => () => {}), + dispose: vi.fn(), + disconnectOnly: vi.fn(async () => {}) + } as unknown as DaemonPtyAdapter + + return { + adapter, + emitData: (sessionId, data) => { + for (const listener of dataListeners) { + listener({ id: sessionId, data }) + } + }, + setIdentity: (next) => { + daemonIdentity = next + }, + emitIdentityChange: (previous, current) => { + for (const listener of identityListeners) { + listener({ previous, current }) + } + }, + emitExit: (sessionId, code) => { + for (const listener of exitListeners) { + listener({ id: sessionId, code }) + } + } + } +} diff --git a/src/main/daemon/daemon-pty-router-routing-safety.test.ts b/src/main/daemon/daemon-pty-router-routing-safety.test.ts index f79792e8801..5b67aa5442c 100644 --- a/src/main/daemon/daemon-pty-router-routing-safety.test.ts +++ b/src/main/daemon/daemon-pty-router-routing-safety.test.ts @@ -1,107 +1,6 @@ import { afterEach, describe, expect, it, vi } from 'vitest' -import type { DaemonEndpointIdentity } from './daemon-hello-protocol' -import type { DaemonPtyAdapter } from './daemon-pty-adapter' import { DaemonPtyRouter } from './daemon-pty-router' -import type { PtyProcessInfo, PtySpawnOptions } from '../providers/types' - -type AdapterHarness = { - adapter: DaemonPtyAdapter - emitData: (sessionId: string, data: string) => void - emitIdentityChange: (previous: DaemonEndpointIdentity, current: DaemonEndpointIdentity) => void - emitExit: (sessionId: string, code: number) => void - setIdentity: (identity: DaemonEndpointIdentity) => void -} - -function identity(label: string, pid: number): DaemonEndpointIdentity { - return { - pid, - startedAtMs: pid * 1_000, - launchNonce: `${label}-${pid}` - } -} - -function createAdapter( - label: string, - sessions: string[] = [], - reconcileResult: { alive: string[]; killed: string[] } = { alive: [], killed: [] } -): AdapterHarness { - let daemonIdentity = identity(label, label === 'current' ? 20 : 10) - const identityListeners: ((event: { - previous: DaemonEndpointIdentity - current: DaemonEndpointIdentity - }) => void)[] = [] - const dataListeners: ((payload: { id: string; data: string }) => void)[] = [] - const exitListeners: ((payload: { id: string; code: number }) => void)[] = [] - const adapter = { - protocolVersion: label === 'current' ? 30 : 29, - getLastAuthenticatedDaemonIdentity: vi.fn(() => ({ ...daemonIdentity })), - matchesLastAuthenticatedDaemonIdentity: vi.fn( - (candidate: DaemonEndpointIdentity | null) => - candidate !== null && - candidate.pid === daemonIdentity.pid && - candidate.startedAtMs === daemonIdentity.startedAtMs && - candidate.launchNonce === daemonIdentity.launchNonce - ), - onDaemonIdentityChanged: vi.fn( - ( - listener: (event: { - previous: DaemonEndpointIdentity - current: DaemonEndpointIdentity - }) => void - ) => { - identityListeners.push(listener) - return () => {} - } - ), - listProcesses: vi.fn( - async (): Promise => sessions.map((id) => ({ id, cwd: '', title: label })) - ), - spawn: vi.fn(async (opts: PtySpawnOptions) => ({ - id: opts.sessionId ?? `${label}-fresh` - })), - probePtyLiveness: vi.fn(async (id: string) => sessions.includes(id)), - hasPty: vi.fn((id: string) => sessions.includes(id)), - shutdown: vi.fn(async () => {}), - sendSignal: vi.fn(async () => {}), - getBufferSnapshot: vi.fn(async () => null), - ackColdRestore: vi.fn(), - reconcileOnStartup: vi.fn(async () => reconcileResult), - onData: vi.fn((listener: (payload: { id: string; data: string }) => void) => { - dataListeners.push(listener) - return () => {} - }), - onExit: vi.fn((listener: (payload: { id: string; code: number }) => void) => { - exitListeners.push(listener) - return () => {} - }), - onBackgroundStreamEvent: vi.fn(() => () => {}), - onWriteUnavailable: vi.fn(() => () => {}), - dispose: vi.fn(), - disconnectOnly: vi.fn(async () => {}) - } as unknown as DaemonPtyAdapter - - return { - adapter, - emitData: (sessionId, data) => { - for (const listener of dataListeners) { - listener({ id: sessionId, data }) - } - }, - setIdentity: (next) => { - daemonIdentity = next - }, - emitIdentityChange: (previous, current) => { - for (const listener of identityListeners) { - listener({ previous, current }) - } - }, - emitExit: (sessionId, code) => { - for (const listener of exitListeners) { - listener({ id: sessionId, code }) - } - } - } -} +import { createAdapter, identity } from './daemon-pty-router-routing-safety-fixture' describe('DaemonPtyRouter routing safety', () => { afterEach(() => { @@ -171,6 +70,67 @@ describe('DaemonPtyRouter routing safety', () => { }) }) + it('keeps a history handoff fenced until the current owner produces output', async () => { + const sessionId = 'history-handoff' + const current = createAdapter('current') + const legacy = createAdapter('legacy', [sessionId]) + let resolveSpawn: ((result: { id: string }) => void) | undefined + vi.mocked(current.adapter.spawn).mockReturnValue( + new Promise((resolve) => { + resolveSpawn = resolve + }) + ) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + const data = vi.fn() + router.onData(data) + + await router.discoverLegacySessions() + await router.shutdown(sessionId, { immediate: true, keepHistory: true }) + expect(router.getSessionRouteState(sessionId)).toBe('unavailable') + + const spawning = router.spawn({ sessionId, cols: 80, rows: 24 }) + current.emitData(sessionId, 'restored frame') + expect(data).toHaveBeenCalledExactlyOnceWith({ id: sessionId, data: 'restored frame' }) + + resolveSpawn?.({ id: sessionId }) + await spawning + expect(router.getSessionRouteState(sessionId)).toBe('owned') + }) + + it('preserves collision evidence during a history handoff', async () => { + const sessionId = 'colliding-history-handoff' + const currentSessions: string[] = [] + const current = createAdapter('current', currentSessions) + const legacy = createAdapter('legacy', [sessionId]) + let resolveSpawn: ((result: { id: string }) => void) | undefined + vi.mocked(current.adapter.spawn).mockReturnValue( + new Promise((resolve) => { + resolveSpawn = resolve + }) + ) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + + await router.discoverLegacySessions() + await router.shutdown(sessionId, { immediate: true, keepHistory: true }) + const spawning = router.spawn({ sessionId, cols: 80, rows: 24 }) + legacy.emitData(sessionId, 'foreign frame') + current.emitData(sessionId, 'restored frame') + currentSessions.push(sessionId) + resolveSpawn?.({ id: sessionId }) + await spawning + + expect(router.getSessionRouteState(sessionId)).toBe('ambiguous') + await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + }) + it('probes adapters outside an ambiguous route while discovery is incomplete', async () => { const sessionId = 'hidden-collision' const firstSessions = [sessionId] @@ -469,6 +429,31 @@ describe('DaemonPtyRouter routing safety', () => { expect(legacy.adapter.sendSignal).not.toHaveBeenCalled() }) + it('rejects a liveness result when ownership changes in flight', async () => { + const sessionId = 'stale-owned-liveness' + const current = createAdapter('current') + const legacy = createAdapter('legacy', [sessionId]) + let finishProbe: ((result: boolean) => void) | undefined + vi.mocked(legacy.adapter.probePtyLiveness).mockReturnValue( + new Promise((resolve) => { + finishProbe = resolve + }) + ) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + + await router.discoverLegacySessions() + const probing = router.probePtyLiveness(sessionId) + await vi.waitFor(() => expect(finishProbe).toBeDefined()) + current.emitData(sessionId, 'colliding frame') + finishProbe?.(false) + + await expect(probing).resolves.toBeNull() + expect(router.getSessionRouteState(sessionId)).toBe('ambiguous') + }) + it('does not apply reconciliation after an earlier adapter is replaced', async () => { const sessionId = 'stale-reconciliation' const current = createAdapter('current', [], { diff --git a/src/main/daemon/daemon-pty-router-session-routing.ts b/src/main/daemon/daemon-pty-router-session-routing.ts index c585dcf71fc..93acee652b0 100644 --- a/src/main/daemon/daemon-pty-router-session-routing.ts +++ b/src/main/daemon/daemon-pty-router-session-routing.ts @@ -7,7 +7,7 @@ import { DaemonSnapshotAcknowledgementRoutes } from './daemon-snapshot-acknowled export class DaemonPtyRouterSessionRouting { private readonly routes: DaemonSessionRouteTable private readonly snapshotAcks = new DaemonSnapshotAcknowledgementRoutes() - private readonly freshSpawnsInFlight = new Map< + private readonly authoritativeSpawnsInFlight = new Map< string, { owner: DaemonPtyAdapter; count: number } >() @@ -48,41 +48,54 @@ export class DaemonPtyRouterSessionRouting { if (!opts.sessionId || opts.isNewSession) { return this.routes.ownerForFreshSpawn(opts.sessionId, this.current) } + const historyHandoffTarget = this.routes.historyHandoffTarget(opts.sessionId) + if (historyHandoffTarget) { + return historyHandoffTarget + } return this.routes.resolveOwner(opts.sessionId) } - recordSpawn(result: PtySpawnResult, target: DaemonPtyAdapter, opts: PtySpawnOptions): void { + recordSpawn( + result: PtySpawnResult, + target: DaemonPtyAdapter, + opts: PtySpawnOptions, + authoritativeIntent: boolean + ): void { if (result.exitedBeforeSpawnReply) { return } - if (opts.isNewSession && opts.sessionId === result.id) { + if (authoritativeIntent && opts.sessionId === result.id) { this.routes.recordFreshOwned(result.id, target) return } this.routes.recordOwned(result.id, target) } - beginSpawn(opts: PtySpawnOptions, target: DaemonPtyAdapter): void { - if (!opts.isNewSession || !opts.sessionId) { - return + beginSpawn(opts: PtySpawnOptions, target: DaemonPtyAdapter): boolean { + if ( + !opts.sessionId || + (!opts.isNewSession && this.routes.historyHandoffTarget(opts.sessionId) !== target) + ) { + return false } - const inFlight = this.freshSpawnsInFlight.get(opts.sessionId) - this.freshSpawnsInFlight.set(opts.sessionId, { + const inFlight = this.authoritativeSpawnsInFlight.get(opts.sessionId) + this.authoritativeSpawnsInFlight.set(opts.sessionId, { owner: target, count: (inFlight?.count ?? 0) + 1 }) + return true } - endSpawn(opts: PtySpawnOptions): void { - if (!opts.isNewSession || !opts.sessionId) { + endSpawn(opts: PtySpawnOptions, authoritativeIntent: boolean): void { + if (!authoritativeIntent || !opts.sessionId) { return } - const inFlight = this.freshSpawnsInFlight.get(opts.sessionId) + const inFlight = this.authoritativeSpawnsInFlight.get(opts.sessionId) if (!inFlight || inFlight.count <= 1) { - this.freshSpawnsInFlight.delete(opts.sessionId) + this.authoritativeSpawnsInFlight.delete(opts.sessionId) return } - this.freshSpawnsInFlight.set(opts.sessionId, { + this.authoritativeSpawnsInFlight.set(opts.sessionId, { owner: inFlight.owner, count: inFlight.count - 1 }) @@ -143,7 +156,7 @@ export class DaemonPtyRouterSessionRouting { migrateHistory: boolean ): void { if (migrateHistory) { - this.routes.transfer(sessionId, this.current) + this.routes.recordHistoryHandoff(sessionId, owner, this.current) return } if (!keepHistory) { @@ -186,9 +199,13 @@ export class DaemonPtyRouterSessionRouting { } shouldForwardStreamEvent(sessionId: string, owner: DaemonPtyAdapter): boolean { - const freshSpawn = this.freshSpawnsInFlight.get(sessionId) - if (freshSpawn && freshSpawn.owner !== owner) { - this.routes.recordFreshOwned(sessionId, freshSpawn.owner) + const authoritativeSpawn = this.authoritativeSpawnsInFlight.get(sessionId) + if (authoritativeSpawn?.owner === owner) { + this.routes.recordFreshOwned(sessionId, owner) + return this.routes.shouldForwardEvent(sessionId, owner) + } + if (authoritativeSpawn) { + this.routes.recordFreshOwned(sessionId, authoritativeSpawn.owner) this.routes.recordOwned(sessionId, owner) return false } @@ -196,8 +213,8 @@ export class DaemonPtyRouterSessionRouting { } recordExit(sessionId: string, owner: DaemonPtyAdapter): boolean { - const freshSpawn = this.freshSpawnsInFlight.get(sessionId) - if (freshSpawn && freshSpawn.owner !== owner) { + const authoritativeSpawn = this.authoritativeSpawnsInFlight.get(sessionId) + if (authoritativeSpawn && authoritativeSpawn.owner !== owner) { if (this.routes.get(sessionId)) { this.routes.markUnavailable(sessionId, owner) } @@ -231,7 +248,7 @@ export class DaemonPtyRouterSessionRouting { for (const unsubscribe of this.identityUnsubscribes.splice(0)) { unsubscribe() } - this.freshSpawnsInFlight.clear() + this.authoritativeSpawnsInFlight.clear() this.snapshotAcks.clear() } } diff --git a/src/main/daemon/daemon-pty-router.ts b/src/main/daemon/daemon-pty-router.ts index 8e616e5e8ae..e8cc451c151 100644 --- a/src/main/daemon/daemon-pty-router.ts +++ b/src/main/daemon/daemon-pty-router.ts @@ -63,13 +63,13 @@ export class DaemonPtyRouter implements IPtyProvider { async spawn(opts: PtySpawnOptions): Promise { const targetResult = this.routing.spawnTarget(opts) const target = targetResult instanceof Promise ? await targetResult : targetResult - this.routing.beginSpawn(opts, target) + const authoritativeIntent = this.routing.beginSpawn(opts, target) try { const result = await target.spawn(opts) - this.routing.recordSpawn(result, target, opts) + this.routing.recordSpawn(result, target, opts, authoritativeIntent) return result } finally { - this.routing.endSpawn(opts) + this.routing.endSpawn(opts, authoritativeIntent) } } diff --git a/src/main/daemon/daemon-session-route-table.ts b/src/main/daemon/daemon-session-route-table.ts index 935426fb1fb..6e5e2ed8ed3 100644 --- a/src/main/daemon/daemon-session-route-table.ts +++ b/src/main/daemon/daemon-session-route-table.ts @@ -9,10 +9,7 @@ import { } from './daemon-session-route' import { probeDaemonSessionOwner } from './daemon-session-owner-probes' import { DaemonSessionRouteStore } from './daemon-session-route-store' -import { - assertFreshDaemonSessionAvailable, - ownerForFreshDaemonSession -} from './daemon-fresh-session-routing' +import * as freshSessionRouting from './daemon-fresh-session-routing' import { resolveDaemonSessionOwnerSync } from './daemon-session-owner-sync-resolution' import { resolveDaemonSessionOwner } from './daemon-session-owner-resolution' import { @@ -20,6 +17,7 @@ import { markDaemonSessionUnavailable } from './daemon-session-route-availability' import { DaemonSessionOwnerResolutionCache } from './daemon-session-owner-resolution-cache' +import * as historyHandoffRouting from './daemon-history-handoff-routing' export class DaemonSessionRouteTable { private readonly routes = new DaemonSessionRouteStore() @@ -126,16 +124,21 @@ export class DaemonSessionRouteTable { } recordFreshOwned(sessionId: string, owner: DaemonPtyAdapter): void { - const route = this.routes.get(sessionId) - if ( - !route || - route.state === 'unavailable' || - (route.state === 'owned' && route.owner === owner && !isCurrentOwnedDaemonSessionRoute(route)) - ) { - this.transfer(sessionId, owner) - return - } - this.recordOwned(sessionId, owner) + freshSessionRouting.recordFreshOwnedDaemonSession( + this.routes.get(sessionId), + owner, + () => this.transfer(sessionId, owner), + (route) => this.routes.set(sessionId, route), + () => this.recordOwned(sessionId, owner) + ) + } + + recordHistoryHandoff(sessionId: string, owner: DaemonPtyAdapter, target: DaemonPtyAdapter): void { + this.routes.set(sessionId, historyHandoffRouting.createDaemonHistoryHandoffRoute(owner, target)) + } + + historyHandoffTarget(sessionId: string): DaemonPtyAdapter | undefined { + return historyHandoffRouting.daemonHistoryHandoffTarget(this.routes.get(sessionId)) } recordDiscoveryFailure(owner: DaemonPtyAdapter): void { @@ -169,7 +172,7 @@ export class DaemonSessionRouteTable { } ownerForFreshSpawn(sessionId: string | undefined, current: DaemonPtyAdapter): DaemonPtyAdapter { - return ownerForFreshDaemonSession( + return freshSessionRouting.ownerForFreshDaemonSession( sessionId, sessionId ? this.routes.get(sessionId) : undefined, current, @@ -178,7 +181,7 @@ export class DaemonSessionRouteTable { } assertFreshSpawnAvailable(sessionId: string | undefined): void { - assertFreshDaemonSessionAvailable( + freshSessionRouting.assertFreshDaemonSessionAvailable( sessionId, sessionId ? this.routes.get(sessionId) : undefined, () => sessionId && this.routes.delete(sessionId) @@ -281,7 +284,8 @@ export class DaemonSessionRouteTable { return null } const result = await probeDaemonSessionOwner(route.owner, sessionId) - return sameDaemonIncarnation(route.incarnation, result.incarnation) && + return this.routes.get(sessionId) === route && + sameDaemonIncarnation(route.incarnation, result.incarnation) && sameDaemonIncarnation(result.incarnation, result.owner.getLastAuthenticatedDaemonIdentity()) ? result.result : null diff --git a/src/main/daemon/daemon-session-route.ts b/src/main/daemon/daemon-session-route.ts index 249e07f684e..06061f3cbe7 100644 --- a/src/main/daemon/daemon-session-route.ts +++ b/src/main/daemon/daemon-session-route.ts @@ -12,6 +12,7 @@ export type DaemonSessionRoute = state: 'unavailable' owner: DaemonPtyAdapter incarnation: DaemonEndpointIdentity | null + historyHandoffTarget?: DaemonPtyAdapter } | { state: 'ambiguous' diff --git a/src/main/daemon/degraded-daemon-pty-provider.test.ts b/src/main/daemon/degraded-daemon-pty-provider.test.ts index e393eb104c0..17d41ceadd9 100644 --- a/src/main/daemon/degraded-daemon-pty-provider.test.ts +++ b/src/main/daemon/degraded-daemon-pty-provider.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it, vi } from 'vitest' import { DegradedDaemonPtyProvider } from './degraded-daemon-pty-provider' -import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import type { DaemonIdentityChangeEvent, DaemonPtyAdapter } from './daemon-pty-adapter' import type { IPtyProvider, PtySpawnOptions, PtySpawnResult } from '../providers/types' import type { PtyProcessInspection } from '../providers/pty-process-inspection' @@ -346,6 +346,45 @@ describe('DegradedDaemonPtyProvider', () => { await expect(provider.probePtyLiveness('unknown-session')).resolves.toBe(true) }) + it('keeps fallback liveness unknown when daemon ownership appears during the probe', async () => { + const sessionId = 'fallback-probe-collision' + const current = createDaemonAdapter('current') + const fallback = createProvider('fallback') + let settleProbe!: (result: boolean | null) => void + fallback.probePtyLiveness = vi.fn( + () => new Promise((resolve) => (settleProbe = resolve)) + ) + const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback }) + await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 }) + + const liveness = provider.probePtyLiveness(sessionId) + current.emitData(sessionId, 'colliding daemon output') + settleProbe(false) + + await expect(liveness).resolves.toBeNull() + }) + + it('discards fallback liveness after same-provider reuse of the session id', async () => { + const sessionId = 'reused-fallback-probe' + const current = createDaemonAdapter('current') + const fallback = createProvider('fallback') + let settleProbe!: (result: boolean | null) => void + fallback.probePtyLiveness = vi.fn( + () => new Promise((resolve) => (settleProbe = resolve)) + ) + const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback }) + await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 }) + + const staleLiveness = provider.probePtyLiveness(sessionId) + fallback.emitExit(sessionId, 0) + await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 }) + settleProbe(false) + + await expect(staleLiveness).resolves.toBeNull() + provider.write(sessionId, 'replacement survives') + expect(fallback.write).toHaveBeenCalledExactlyOnceWith(sessionId, 'replacement survives') + }) + it('routes authoritative recovery snapshots to the owning daemon', async () => { const current = createDaemonAdapter('daemon', ['daemon-session']) const fallback = createProvider('fallback') @@ -487,6 +526,40 @@ describe('DegradedDaemonPtyProvider', () => { expect(exit).toHaveBeenCalledExactlyOnceWith({ id: sessionId, code: 0 }) }) + it('releases a fallback survivor when the colliding daemon changes incarnation', async () => { + const sessionId = 'daemon-incarnation-collision' + const previous = { pid: 1, startedAtMs: 1, launchNonce: 'previous' } + const currentIdentity = { pid: 2, startedAtMs: 2, launchNonce: 'current' } + const current = createDaemonAdapter('daemon') + const fallback = createProvider('fallback') + let identity = previous + let notifyIdentityChange!: (event: DaemonIdentityChangeEvent) => void + vi.mocked(current.getLastAuthenticatedDaemonIdentity).mockImplementation(() => identity) + vi.mocked(current.onDaemonIdentityChanged).mockImplementation((listener) => { + notifyIdentityChange = listener + return () => {} + }) + const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback }) + const data = vi.fn() + provider.onData(data) + await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 }) + + current.emitData(sessionId, 'withheld collision') + expect(() => provider.write(sessionId, 'withheld write')).toThrow( + 'daemon_session_routing_unavailable' + ) + identity = currentIdentity + notifyIdentityChange({ previous, current: currentIdentity }) + fallback.emitData(sessionId, 'fallback survivor') + provider.write(sessionId, 'surviving write') + + expect(data).toHaveBeenCalledExactlyOnceWith({ + id: sessionId, + data: 'fallback survivor' + }) + expect(fallback.write).toHaveBeenCalledExactlyOnceWith(sessionId, 'surviving write') + }) + it('keeps a fallback collision fenced while another daemon owner survives', async () => { const sessionId = 'two-daemon-fallback-collision' const current = createDaemonAdapter('current') @@ -516,26 +589,37 @@ describe('DegradedDaemonPtyProvider', () => { expect(fallback.write).toHaveBeenCalledExactlyOnceWith(sessionId, 'allowed') }) - it('records fallback ownership from output emitted before spawn resolves', async () => { + it('fences foreign events before a fresh fallback reply and keeps the fallback routable', async () => { const sessionId = 'in-flight-fallback-spawn' const current = createDaemonAdapter('daemon') const fallback = createProvider('fallback') const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback }) const data = vi.fn() + const exit = vi.fn() provider.onData(data) - vi.mocked(fallback.spawn).mockImplementationOnce(async () => { - fallback.emitData(sessionId, 'fallback first') - return { id: sessionId } - }) + provider.onExit(exit) + let resolveSpawn!: (result: PtySpawnResult) => void + vi.mocked(fallback.spawn).mockReturnValueOnce( + new Promise((resolve) => { + resolveSpawn = resolve + }) + ) - await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 }) - current.emitData(sessionId, 'daemon collision') - fallback.emitData(sessionId, 'withheld fallback collision') + const spawning = provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 }) + current.emitData(sessionId, 'foreign daemon output') + current.emitExit(sessionId, 0) + fallback.emitData(sessionId, 'fallback first') + resolveSpawn({ id: sessionId }) + await spawning + provider.write(sessionId, 'fallback input') expect(data).toHaveBeenCalledExactlyOnceWith({ id: sessionId, data: 'fallback first' }) + expect(exit).not.toHaveBeenCalled() + expect(fallback.write).toHaveBeenCalledExactlyOnceWith(sessionId, 'fallback input') + expect(current.write).not.toHaveBeenCalled() }) it('rejects unknown fallback output after unavailable tombstones are evicted', async () => { diff --git a/src/main/daemon/degraded-daemon-session-routing.ts b/src/main/daemon/degraded-daemon-session-routing.ts index 09c718eafdc..c28a6d5404c 100644 --- a/src/main/daemon/degraded-daemon-session-routing.ts +++ b/src/main/daemon/degraded-daemon-session-routing.ts @@ -4,6 +4,7 @@ import { shutdownDegradedFallbackSessions } from './degraded-daemon-fallback-shu import { DaemonSessionRouteTable } from './daemon-session-route-table' import { DaemonSessionOwnerUnknownError, sameDaemonIncarnation } from './daemon-session-route' import type { DaemonSnapshotAcknowledgementRoutes } from './daemon-snapshot-acknowledgement-routes' +import { DegradedFallbackSessionRoutes } from './degraded-fallback-session-routes' import { DegradedFallbackSpawnRoutes } from './degraded-fallback-spawn-routes' export class DegradedDaemonSessionRouting { @@ -11,7 +12,7 @@ export class DegradedDaemonSessionRouting { private readonly adapterSet: ReadonlySet private readonly routes: DaemonSessionRouteTable private readonly fallbackSessions = new Map() - private readonly fallbackCollisions = new Set() + private readonly fallbackRoutes: DegradedFallbackSessionRoutes private readonly fallbackSpawns = new DegradedFallbackSpawnRoutes() private readonly identityUnsubscribes: (() => void)[] = [] @@ -24,11 +25,12 @@ export class DegradedDaemonSessionRouting { this.adapters = [current, ...legacy] this.adapterSet = new Set(this.adapters) this.routes = new DaemonSessionRouteTable(this.adapters) + this.fallbackRoutes = new DegradedFallbackSessionRoutes(this.fallbackSessions, this.routes) for (const adapter of this.adapters) { this.identityUnsubscribes.push( adapter.onDaemonIdentityChanged(({ previous }) => { this.snapshotAcks.dropAdapterIncarnation(adapter, previous) - this.routes.markOwnerUnavailable(adapter, previous) + this.fallbackRoutes.markDaemonOwnerUnavailable(adapter, previous) }) ) } @@ -50,7 +52,7 @@ export class DegradedDaemonSessionRouting { ) for (const session of sessions) { if (this.fallbackSessions.has(session.id)) { - this.fallbackCollisions.add(session.id) + this.fallbackRoutes.recordCollision(session.id) } } } catch (error) { @@ -61,7 +63,8 @@ export class DegradedDaemonSessionRouting { } async spawn(opts: PtySpawnOptions): Promise { - const target = await this.spawnTarget(opts) + const targetOrPromise = this.spawnTarget(opts) + const target = targetOrPromise instanceof Promise ? await targetOrPromise : targetOrPromise const inFlightFallbackId = target === this.fallback ? opts.sessionId : undefined if (inFlightFallbackId) { this.fallbackSpawns.begin(inFlightFallbackId) @@ -69,11 +72,13 @@ export class DegradedDaemonSessionRouting { try { const result = await target.spawn(opts) if (target === this.fallback) { - if (!this.fallbackSpawns.hasExited(result.id)) { - this.fallbackSessions.set(result.id, target) + const fallbackSurvived = !this.fallbackSpawns.hasExited(result.id) + if (fallbackSurvived) { + this.fallbackRoutes.recordSession(result.id, target) } - if (this.routes.get(result.id)) { - this.fallbackCollisions.add(result.id) + const daemonRoute = this.routes.get(result.id) + if (fallbackSurvived && daemonRoute && daemonRoute.state !== 'unavailable') { + this.fallbackRoutes.recordCollision(result.id) } } else { this.routes.recordOwned(result.id, target as DaemonPtyAdapter) @@ -87,7 +92,7 @@ export class DegradedDaemonSessionRouting { } providerFor(sessionId: string): IPtyProvider { - if (this.fallbackCollisions.has(sessionId)) { + if (this.fallbackRoutes.hasCollision(sessionId)) { throw new DaemonSessionOwnerUnknownError(sessionId) } return this.fallbackSessions.get(sessionId) ?? this.routes.resolveOwnerSync(sessionId) @@ -102,7 +107,7 @@ export class DegradedDaemonSessionRouting { } hasPty(sessionId: string): boolean { - if (this.fallbackCollisions.has(sessionId)) { + if (this.fallbackRoutes.hasCollision(sessionId)) { return true } const fallback = this.fallbackSessions.get(sessionId) @@ -110,24 +115,25 @@ export class DegradedDaemonSessionRouting { } async probePtyLiveness(sessionId: string): Promise { - if (this.fallbackCollisions.has(sessionId)) { + if (this.fallbackRoutes.hasCollision(sessionId)) { return null } - const fallback = this.fallbackSessions.get(sessionId) - if (!fallback) { + const probe = this.fallbackRoutes.captureLivenessProbe(sessionId) + if (!probe) { return await this.routes.probeLiveness(sessionId) } - return fallback.probePtyLiveness - ? await fallback.probePtyLiveness(sessionId) - : (fallback.hasPty?.(sessionId) ?? null) + const result = probe.provider.probePtyLiveness + ? await probe.provider.probePtyLiveness(sessionId) + : (probe.provider.hasPty?.(sessionId) ?? null) + return this.fallbackRoutes.acceptsLivenessResult(sessionId, probe) ? result : null } shouldForwardStreamEvent(provider: IPtyProvider, sessionId: string): boolean { const adapter = this.daemonAdapter(provider) if (adapter) { const shouldForward = this.routes.recordDataOwner(sessionId, adapter) - if (this.fallbackSessions.has(sessionId)) { - this.fallbackCollisions.add(sessionId) + if (this.fallbackSessions.has(sessionId) || this.fallbackSpawns.hasLiveCandidate(sessionId)) { + this.fallbackRoutes.recordCollision(sessionId) return false } return shouldForward @@ -141,15 +147,16 @@ export class DegradedDaemonSessionRouting { ) } if ( - this.fallbackSessions.get(sessionId) === provider && + (this.fallbackSessions.get(sessionId) === provider || + this.fallbackSpawns.hasLiveCandidate(sessionId)) && route.state === 'unavailable' && - !this.fallbackCollisions.has(sessionId) + !this.fallbackRoutes.hasCollision(sessionId) ) { - this.fallbackCollisions.delete(sessionId) + this.fallbackRoutes.clearCollision(sessionId) return true } if (route) { - this.fallbackCollisions.add(sessionId) + this.fallbackRoutes.recordCollision(sessionId) return false } return true @@ -160,21 +167,20 @@ export class DegradedDaemonSessionRouting { if (!adapter) { const isOwned = this.fallbackSessions.get(sessionId) === provider || - this.fallbackSpawns.isInFlight(sessionId) + this.fallbackSpawns.hasLiveCandidate(sessionId) if (!isOwned) { return false } this.fallbackSpawns.recordExit(sessionId) - this.fallbackSessions.delete(sessionId) - this.fallbackCollisions.delete(sessionId) + this.fallbackRoutes.deleteSession(sessionId) const route = this.routes.get(sessionId) return !route || route.state === 'unavailable' } this.snapshotAcks.dropForProducer(sessionId, adapter) - if (this.fallbackSessions.has(sessionId)) { + if (this.fallbackSessions.has(sessionId) || this.fallbackSpawns.hasLiveCandidate(sessionId)) { this.routes.markUnavailable(sessionId, adapter) if (this.routes.get(sessionId)?.state === 'unavailable') { - this.fallbackCollisions.delete(sessionId) + this.fallbackRoutes.clearCollision(sessionId) } return false } @@ -188,8 +194,7 @@ export class DegradedDaemonSessionRouting { return } if (provider === this.fallback) { - this.fallbackSessions.delete(sessionId) - this.fallbackCollisions.delete(sessionId) + this.fallbackRoutes.deleteSession(sessionId) } else { const adapter = provider as DaemonPtyAdapter this.routes.markUnavailable(sessionId, adapter) @@ -232,7 +237,7 @@ export class DegradedDaemonSessionRouting { for (const id of sessionIds) { this.routes.recordOwnedIncarnation(id, adapter, incarnation) if (this.fallbackSessions.has(id)) { - this.fallbackCollisions.add(id) + this.fallbackRoutes.recordCollision(id) } } } @@ -241,7 +246,7 @@ export class DegradedDaemonSessionRouting { this.routes.markUnavailable(id, adapter, incarnation) this.snapshotAcks.dropForProducer(id, adapter, incarnation) if (this.routes.get(id)?.state === 'unavailable') { - this.fallbackCollisions.delete(id) + this.fallbackRoutes.clearCollision(id) } } } @@ -265,14 +270,16 @@ export class DegradedDaemonSessionRouting { this.routes.markUnavailable(id, this.current) this.snapshotAcks.dropForProducer(id, this.current) if (this.routes.get(id)?.state === 'unavailable') { - this.fallbackCollisions.delete(id) + this.fallbackRoutes.clearCollision(id) } } return forwarded } async shutdownFallbackSessions(): Promise { - return await shutdownDegradedFallbackSessions(this.fallbackSessions, this.fallback) + const count = await shutdownDegradedFallbackSessions(this.fallbackSessions, this.fallback) + this.fallbackRoutes.pruneSessionEpochs() + return count } allProviders(): IPtyProvider[] { @@ -287,23 +294,20 @@ export class DegradedDaemonSessionRouting { for (const unsubscribe of this.identityUnsubscribes.splice(0)) { unsubscribe() } - this.fallbackSessions.clear() - this.fallbackCollisions.clear() + this.fallbackRoutes.clear() this.fallbackSpawns.clear() this.snapshotAcks.clear() } - private async spawnTarget(opts: PtySpawnOptions): Promise { - if (opts.sessionId && this.fallbackCollisions.has(opts.sessionId)) { + private spawnTarget(opts: PtySpawnOptions): IPtyProvider | Promise { + if (opts.sessionId && this.fallbackRoutes.hasCollision(opts.sessionId)) { throw new DaemonSessionOwnerUnknownError(opts.sessionId) } if (!opts.sessionId || opts.isNewSession) { this.routes.assertFreshSpawnAvailable(opts.sessionId) return this.fallback } - return ( - this.fallbackSessions.get(opts.sessionId) ?? (await this.routes.resolveOwner(opts.sessionId)) - ) + return this.fallbackSessions.get(opts.sessionId) ?? this.routes.resolveOwner(opts.sessionId) } private daemonAdapter(provider: IPtyProvider): DaemonPtyAdapter | null { diff --git a/src/main/daemon/degraded-fallback-session-routes.ts b/src/main/daemon/degraded-fallback-session-routes.ts new file mode 100644 index 00000000000..aac2642bb69 --- /dev/null +++ b/src/main/daemon/degraded-fallback-session-routes.ts @@ -0,0 +1,110 @@ +import type { IPtyProvider } from '../providers/types' +import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import { sameDaemonIncarnation } from './daemon-session-route' +import type { DaemonSessionRouteTable } from './daemon-session-route-table' + +type FallbackLivenessProbe = { + provider: IPtyProvider + epoch: symbol + daemonRoute: ReturnType +} + +export class DegradedFallbackSessionRoutes { + private readonly collisionIds = new Set() + private readonly epochs = new Map() + + constructor( + private readonly sessions: Map, + private readonly daemonRoutes: DaemonSessionRouteTable + ) {} + + hasCollision(sessionId: string): boolean { + return this.collisionIds.has(sessionId) + } + + recordCollision(sessionId: string): void { + this.collisionIds.add(sessionId) + } + + clearCollision(sessionId: string): void { + this.collisionIds.delete(sessionId) + } + + recordSession(sessionId: string, provider: IPtyProvider): void { + this.sessions.set(sessionId, provider) + this.epochs.set(sessionId, Symbol()) + } + + deleteSession(sessionId: string): void { + this.sessions.delete(sessionId) + this.epochs.delete(sessionId) + this.collisionIds.delete(sessionId) + } + + captureLivenessProbe(sessionId: string): FallbackLivenessProbe | null { + const provider = this.sessions.get(sessionId) + const epoch = this.epochs.get(sessionId) + if (!provider || !epoch || this.collisionIds.has(sessionId)) { + return null + } + return { provider, epoch, daemonRoute: this.daemonRoutes.get(sessionId) } + } + + acceptsLivenessResult(sessionId: string, probe: FallbackLivenessProbe): boolean { + return ( + this.sessions.get(sessionId) === probe.provider && + this.epochs.get(sessionId) === probe.epoch && + this.daemonRoutes.get(sessionId) === probe.daemonRoute && + !this.collisionIds.has(sessionId) + ) + } + + pruneSessionEpochs(): void { + for (const sessionId of this.epochs.keys()) { + if (!this.sessions.has(sessionId)) { + this.epochs.delete(sessionId) + this.collisionIds.delete(sessionId) + } + } + } + + clear(): void { + this.sessions.clear() + this.epochs.clear() + this.collisionIds.clear() + } + + markDaemonOwnerUnavailable( + owner: DaemonPtyAdapter, + incarnation: ReturnType + ): void { + const invalidated = [...this.collisionIds].filter((sessionId) => + this.isCollisionOwnedBy(sessionId, owner, incarnation) + ) + this.daemonRoutes.markOwnerUnavailable(owner, incarnation) + for (const sessionId of invalidated) { + if (this.daemonRoutes.get(sessionId)?.state === 'unavailable') { + this.collisionIds.delete(sessionId) + } + } + } + + private isCollisionOwnedBy( + sessionId: string, + owner: DaemonPtyAdapter, + incarnation: ReturnType + ): boolean { + if (!this.sessions.has(sessionId)) { + return false + } + const route = this.daemonRoutes.get(sessionId) + if (route?.state === 'owned') { + return route.owner === owner && sameDaemonIncarnation(route.incarnation, incarnation) + } + return ( + route?.state === 'ambiguous' && + route.candidates.has(owner) && + sameDaemonIncarnation(route.candidates.get(owner) ?? null, incarnation) + ) + } +} diff --git a/src/main/daemon/degraded-fallback-spawn-routes.ts b/src/main/daemon/degraded-fallback-spawn-routes.ts index 6977a8f0dcb..289f7d36ee5 100644 --- a/src/main/daemon/degraded-fallback-spawn-routes.ts +++ b/src/main/daemon/degraded-fallback-spawn-routes.ts @@ -28,6 +28,10 @@ export class DegradedFallbackSpawnRoutes { return this.exited.has(sessionId) } + hasLiveCandidate(sessionId: string): boolean { + return this.isInFlight(sessionId) && !this.hasExited(sessionId) + } + recordExit(sessionId: string): void { if (this.isInFlight(sessionId)) { this.exited.add(sessionId) diff --git a/src/main/ipc/pty-management.test.ts b/src/main/ipc/pty-management.test.ts index 4a63a08b211..4fcd469f9a1 100644 --- a/src/main/ipc/pty-management.test.ts +++ b/src/main/ipc/pty-management.test.ts @@ -328,24 +328,35 @@ describe('pty:management IPC handlers', () => { expect(current.shutdown).toHaveBeenCalledTimes(1) }) - it('keeps sessions remaining when their owner becomes unavailable while polling', async () => { + it('stops polling an unavailable owner while healthy owners converge', async () => { const current = makeAdapter(5, [makeSession('uncertain-session')]) + const legacy = makeAdapter(3, []) vi.mocked(current.listSessions) .mockResolvedValueOnce([ makeSession('uncertain-session') as Omit ]) .mockRejectedValue(new Error('listing unavailable')) + vi.mocked(legacy.listSessions) + .mockResolvedValueOnce([ + makeSession('eventual-session') as Omit + ]) + .mockResolvedValueOnce([ + makeSession('eventual-session') as Omit + ]) + .mockResolvedValue([]) const { registerDaemonManagementHandlers } = await importFresh() - getDaemonProviderMock.mockReturnValue(await makeRouter(current)) + getDaemonProviderMock.mockReturnValue(await makeRouter(current, [legacy])) registerDaemonManagementHandlers() const result = await runKillAllWithPolls(buildHandlerMap()['pty:management:killAll']) expect(result).toEqual({ - killedCount: 0, + killedCount: 1, remainingCount: 1, - killedSessionIds: [] + killedSessionIds: ['eventual-session'] }) + expect(current.listSessions).toHaveBeenCalledTimes(2) + expect(legacy.listSessions).toHaveBeenCalledTimes(3) }) it('does not count respawned sessions with fresh IDs against remainingCount', async () => { diff --git a/src/main/ipc/pty-management.ts b/src/main/ipc/pty-management.ts index 8f784fad459..0b9d4b35ba3 100644 --- a/src/main/ipc/pty-management.ts +++ b/src/main/ipc/pty-management.ts @@ -149,18 +149,29 @@ export function registerDaemonManagementHandlers(): void { // Why: count only the initial-snapshot intersection so renderer respawns mid-kill aren't counted as remaining. let remainingOriginal = initial + let pollingAdapters = adapters + const unavailableOwners = new Set() for (let attempt = 0; attempt < MAX_POLL_ATTEMPTS; attempt += 1) { await sleep(POLL_INTERVAL_MS) - const current = await collectSessions(adapters) + const current = await collectSessions(pollingAdapters) + for (const owner of current.unavailableOwners) { + unavailableOwners.add(owner) + } const currentIndex = indexSessionIncarnations(current.sessions) remainingOriginal = initial.filter( (original) => - current.unavailableOwners.has(original.owner) || - hasSessionIncarnation(currentIndex, original) + unavailableOwners.has(original.owner) || hasSessionIncarnation(currentIndex, original) ) if (remainingOriginal.length === 0) { break } + const remainingOwners = new Set(remainingOriginal.map(({ owner }) => owner)) + pollingAdapters = pollingAdapters.filter( + (adapter) => remainingOwners.has(adapter) && !unavailableOwners.has(adapter) + ) + if (pollingAdapters.length === 0) { + break + } } const remainingSet = new Set(remainingOriginal)