From 86d3331aa0a1405e7e9c58282561bf64c08b54cd Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 30 Jul 2026 22:37:03 -0700 Subject: [PATCH] fix(daemon): harden generation route fences --- .../daemon/daemon-fresh-session-routing.ts | 46 ++ src/main/daemon/daemon-health.test.ts | 61 +++ src/main/daemon/daemon-health.ts | 14 +- src/main/daemon/daemon-pty-adapter.test.ts | 5 + src/main/daemon/daemon-pty-adapter.ts | 7 + src/main/daemon/daemon-pty-liveness-probe.ts | 20 - .../daemon-pty-router-routing-safety.test.ts | 499 +++++++++++++++++- .../daemon-pty-router-session-routing.ts | 110 +++- src/main/daemon/daemon-pty-router.test.ts | 35 ++ src/main/daemon/daemon-pty-router.ts | 50 +- .../daemon/daemon-session-owner-probes.ts | 49 ++ .../daemon-session-owner-resolution-cache.ts | 24 + .../daemon/daemon-session-owner-resolution.ts | 51 ++ .../daemon-session-owner-sync-resolution.ts | 56 ++ .../daemon-session-route-availability.ts | 83 +++ src/main/daemon/daemon-session-route-store.ts | 57 ++ src/main/daemon/daemon-session-route-table.ts | 364 +++++++------ src/main/daemon/daemon-session-route.ts | 16 +- .../daemon-snapshot-acknowledgement-routes.ts | 102 +++- .../degraded-daemon-pty-provider.test.ts | 286 +++++++++- .../daemon/degraded-daemon-pty-provider.ts | 161 +++--- .../daemon/degraded-daemon-session-routing.ts | 326 +++++++++++- .../daemon/degraded-fallback-spawn-routes.ts | 41 ++ src/main/ipc/pty-management.test.ts | 28 + src/main/ipc/pty-management.ts | 53 +- 25 files changed, 2162 insertions(+), 382 deletions(-) create mode 100644 src/main/daemon/daemon-fresh-session-routing.ts delete mode 100644 src/main/daemon/daemon-pty-liveness-probe.ts create mode 100644 src/main/daemon/daemon-session-owner-probes.ts create mode 100644 src/main/daemon/daemon-session-owner-resolution-cache.ts create mode 100644 src/main/daemon/daemon-session-owner-resolution.ts create mode 100644 src/main/daemon/daemon-session-owner-sync-resolution.ts create mode 100644 src/main/daemon/daemon-session-route-availability.ts create mode 100644 src/main/daemon/daemon-session-route-store.ts create mode 100644 src/main/daemon/degraded-fallback-spawn-routes.ts diff --git a/src/main/daemon/daemon-fresh-session-routing.ts b/src/main/daemon/daemon-fresh-session-routing.ts new file mode 100644 index 00000000000..057a9469145 --- /dev/null +++ b/src/main/daemon/daemon-fresh-session-routing.ts @@ -0,0 +1,46 @@ +import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import { + DaemonSessionOwnerUnknownError, + type DaemonSessionRoute, + DaemonSessionUnavailableError, + sameDaemonIncarnation +} from './daemon-session-route' + +export function ownerForFreshDaemonSession( + sessionId: string | undefined, + route: DaemonSessionRoute | undefined, + current: DaemonPtyAdapter, + clearRoute: () => void +): DaemonPtyAdapter { + if (!sessionId || !route) { + return current + } + if (route.state === 'owned' && route.owner === current) { + return current + } + assertFreshDaemonSessionAvailable(sessionId, route, clearRoute) + return current +} + +export function assertFreshDaemonSessionAvailable( + sessionId: string | undefined, + route: DaemonSessionRoute | undefined, + clearRoute: () => void +): void { + if (!sessionId || !route) { + return + } + if (route.state === 'unavailable') { + const ownerIncarnation = route.owner.getLastAuthenticatedDaemonIdentity() + if ( + route.incarnation && + ownerIncarnation && + !sameDaemonIncarnation(route.incarnation, ownerIncarnation) + ) { + clearRoute() + return + } + throw new DaemonSessionUnavailableError(sessionId) + } + throw new DaemonSessionOwnerUnknownError(sessionId) +} diff --git a/src/main/daemon/daemon-health.test.ts b/src/main/daemon/daemon-health.test.ts index f1128e7e370..6eb8d1c8cb8 100644 --- a/src/main/daemon/daemon-health.test.ts +++ b/src/main/daemon/daemon-health.test.ts @@ -510,4 +510,65 @@ describe('killStaleDaemon pid identity guards', () => { await closeServer(replacementSocket) } }) + + it('preserves a replacement pidfile installed as the targeted process exits', async () => { + if (process.platform === 'win32') { + return + } + + const replacementSocket = createServer((socket) => socket.end()) + await new Promise((resolve, reject) => { + replacementSocket.once('error', reject) + replacementSocket.listen(socketPath, () => { + replacementSocket.off('error', reject) + resolve() + }) + }) + const pidPath = getDaemonPidPath(dir) + const replacementPidfile = serializeDaemonPidFile({ + pid: process.pid, + startedAtMs: null, + launchNonce: 'replacement' + }) + const child = spawn( + process.execPath, + [ + '-e', + "const fs=require('node:fs');const [pidPath,pidfile]=process.argv.slice(1);process.on('SIGTERM',()=>{fs.writeFileSync(pidPath,pidfile);process.exit(0)});process.send?.('ready');setInterval(()=>{},1000)", + pidPath, + replacementPidfile, + 'daemon-entry', + socketPath, + tokenPath + ], + { stdio: ['ignore', 'ignore', 'ignore', 'ipc'] } + ) + await new Promise((resolve, reject) => { + child.once('error', reject) + child.on('message', (message) => { + if (message === 'ready') { + resolve() + } + }) + }) + writeFileSync( + pidPath, + serializeDaemonPidFile({ + pid: child.pid!, + startedAtMs: null, + launchNonce: 'target' + }), + { mode: 0o600 } + ) + + try { + await expect(killStaleDaemon(dir, socketPath, tokenPath)).resolves.toBe(true) + expect(readFileSync(pidPath, 'utf8')).toBe(replacementPidfile) + } finally { + if (child.exitCode === null) { + child.kill('SIGKILL') + } + await closeServer(replacementSocket) + } + }) }) diff --git a/src/main/daemon/daemon-health.ts b/src/main/daemon/daemon-health.ts index 384c7e4ecb4..5b7336f4007 100644 --- a/src/main/daemon/daemon-health.ts +++ b/src/main/daemon/daemon-health.ts @@ -10,7 +10,7 @@ import { } from '../../shared/process-output-field-scanner' import { isStartupDiagnosticsEnabled, logStartupDiagnostic } from '../startup/startup-diagnostics' import { encodeNdjson } from './ndjson' -import { getDaemonPidPath } from './daemon-spawner' +import { getDaemonPidPath, unlinkOwnedDaemonPidFile } from './daemon-spawner' import { PROTOCOL_VERSION, type HelloMessage, @@ -666,6 +666,8 @@ export async function killStaleDaemon( const pidPath = getDaemonPidPath(runtimeDir, protocolVersion) let killedDaemon = false let identifiedDaemon = false + let identifiedPid: number | null = null + let identifiedLaunchNonce: string | null = null try { const parsedPid = parseDaemonPidFile(readFileSync(pidPath, 'utf8')) if ( @@ -673,6 +675,8 @@ export async function killStaleDaemon( (await isDaemonProcess(parsedPid.pid, socketPath, tokenPath, parsedPid.startedAtMs)) ) { identifiedDaemon = true + identifiedPid = parsedPid.pid + identifiedLaunchNonce = parsedPid.launchNonce const { pid, startedAtMs } = parsedPid process.kill(pid, 'SIGTERM') const deadline = Date.now() + KILL_WAIT_MS @@ -711,12 +715,8 @@ export async function killStaleDaemon( // PID file missing or process already dead } - if (identifiedDaemon) { - try { - unlinkSync(pidPath) - } catch { - // Best-effort - } + if (identifiedDaemon && identifiedPid !== null && identifiedLaunchNonce !== null) { + unlinkOwnedDaemonPidFile(pidPath, identifiedPid, identifiedLaunchNonce) } const socketIsLive = await canConnectSocket(socketPath) diff --git a/src/main/daemon/daemon-pty-adapter.test.ts b/src/main/daemon/daemon-pty-adapter.test.ts index 5a3d766b85d..885ad152e8b 100644 --- a/src/main/daemon/daemon-pty-adapter.test.ts +++ b/src/main/daemon/daemon-pty-adapter.test.ts @@ -1527,6 +1527,7 @@ describe('DaemonPtyAdapter (IPtyProvider)', () => { await adapter.listProcesses() const firstIdentity = adapter.getLastAuthenticatedDaemonIdentity() expect(firstIdentity).not.toBeNull() + expect(adapter.matchesLastAuthenticatedDaemonIdentity(firstIdentity)).toBe(true) const identityChanges: { previous: NonNullable current: NonNullable @@ -1568,6 +1569,10 @@ describe('DaemonPtyAdapter (IPtyProvider)', () => { } ]) expect(adapter.getLastAuthenticatedDaemonIdentity()).toEqual(identityChanges[0]?.current) + expect(adapter.matchesLastAuthenticatedDaemonIdentity(firstIdentity)).toBe(false) + expect( + adapter.matchesLastAuthenticatedDaemonIdentity(identityChanges[0]?.current ?? null) + ).toBe(true) }) it('isolates audit observation listeners from inventory and later listeners', async () => { diff --git a/src/main/daemon/daemon-pty-adapter.ts b/src/main/daemon/daemon-pty-adapter.ts index 9fcf5a8633d..a039926086e 100644 --- a/src/main/daemon/daemon-pty-adapter.ts +++ b/src/main/daemon/daemon-pty-adapter.ts @@ -299,6 +299,13 @@ export class DaemonPtyAdapter implements IPtyProvider { return this.lastAuthenticatedIdentity ? { ...this.lastAuthenticatedIdentity } : null } + matchesLastAuthenticatedDaemonIdentity(identity: DaemonEndpointIdentity | null): boolean { + if (!identity || !this.lastAuthenticatedIdentity) { + return identity === this.lastAuthenticatedIdentity + } + return sameEndpointIdentity(identity, this.lastAuthenticatedIdentity) + } + getLastAuditObservation(): DaemonAuditObservation | null { return this.lastAuditObservation } diff --git a/src/main/daemon/daemon-pty-liveness-probe.ts b/src/main/daemon/daemon-pty-liveness-probe.ts deleted file mode 100644 index ba2c448bbd2..00000000000 --- a/src/main/daemon/daemon-pty-liveness-probe.ts +++ /dev/null @@ -1,20 +0,0 @@ -import type { IPtyProvider } from '../providers/types' -import type { DaemonPtyAdapter } from './daemon-pty-adapter' - -export async function probePtyOwners( - id: string, - routed: IPtyProvider | undefined, - possibleOwners: readonly DaemonPtyAdapter[] -): Promise { - if (routed) { - return routed.probePtyLiveness - ? await routed.probePtyLiveness(id) - : (routed.hasPty?.(id) ?? null) - } - const results = await Promise.all(possibleOwners.map((provider) => provider.probePtyLiveness(id))) - return results.some((result) => result === true) - ? true - : results.every((result) => result === false) - ? false - : null -} 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 c38f5367468..f79792e8801 100644 --- a/src/main/daemon/daemon-pty-router-routing-safety.test.ts +++ b/src/main/daemon/daemon-pty-router-routing-safety.test.ts @@ -6,7 +6,9 @@ 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 } @@ -28,9 +30,18 @@ function createAdapter( 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: { @@ -55,8 +66,14 @@ function createAdapter( getBufferSnapshot: vi.fn(async () => null), ackColdRestore: vi.fn(), reconcileOnStartup: vi.fn(async () => reconcileResult), - onData: vi.fn(() => () => {}), - onExit: vi.fn(() => () => {}), + 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(), @@ -65,6 +82,11 @@ function createAdapter( return { adapter, + emitData: (sessionId, data) => { + for (const listener of dataListeners) { + listener({ id: sessionId, data }) + } + }, setIdentity: (next) => { daemonIdentity = next }, @@ -72,6 +94,11 @@ function createAdapter( for (const listener of identityListeners) { listener({ previous, current }) } + }, + emitExit: (sessionId, code) => { + for (const listener of exitListeners) { + listener({ id: sessionId, code }) + } } } } @@ -81,6 +108,25 @@ describe('DaemonPtyRouter routing safety', () => { vi.restoreAllMocks() }) + it('does not clone daemon identity for established stream chunks', () => { + const current = createAdapter('current') + const data = vi.fn() + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [] + }) + router.onData(data) + + current.emitData('stream-session', 'first') + vi.mocked(current.adapter.getLastAuthenticatedDaemonIdentity).mockClear() + for (let index = 0; index < 100; index += 1) { + current.emitData('stream-session', `chunk-${index}`) + } + + expect(data).toHaveBeenCalledTimes(101) + expect(current.adapter.getLastAuthenticatedDaemonIdentity).not.toHaveBeenCalled() + }) + it('uses conclusive probes while fencing absence after a failed discovery', async () => { const current = createAdapter('current') const legacy = createAdapter('legacy', ['legacy-session']) @@ -151,6 +197,36 @@ describe('DaemonPtyRouter routing safety', () => { expect(undiscovered.adapter.sendSignal).not.toHaveBeenCalled() }) + it('probes every adapter before collapsing an ambiguous route', async () => { + const sessionId = 'late-collision' + const firstSessions = [sessionId] + const lateSessions: string[] = [] + const current = createAdapter('current', [sessionId], { + alive: [sessionId], + killed: [] + }) + const first = createAdapter('legacy-first', firstSessions, { + alive: [sessionId], + killed: [] + }) + const late = createAdapter('legacy-late', lateSessions) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [first.adapter, late.adapter] + }) + + await router.reconcileOnStartup(new Set()) + firstSessions.splice(0) + lateSessions.push(sessionId) + + await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + expect(current.adapter.sendSignal).not.toHaveBeenCalled() + expect(first.adapter.sendSignal).not.toHaveBeenCalled() + expect(late.adapter.sendSignal).not.toHaveBeenCalled() + }) + it('fails existing operations unknown when an owner probe is inconclusive', async () => { const current = createAdapter('current') const legacy = createAdapter('legacy') @@ -175,7 +251,41 @@ describe('DaemonPtyRouter routing safety', () => { expect(current.adapter.sendSignal).not.toHaveBeenCalled() }) - it('probes every adapter once and memoizes the unique owner', async () => { + it('keeps inconclusive synchronous liveness non-authoritative', async () => { + const sessionId = 'uncertain-liveness' + const current = createAdapter('current') + const legacy = createAdapter('legacy') + vi.mocked(legacy.adapter.listProcesses).mockRejectedValueOnce(new Error('listing unavailable')) + vi.spyOn(console, 'warn').mockImplementation(() => {}) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + + await router.discoverLegacySessions() + + expect(() => router.hasPty(sessionId)).toThrow('daemon_session_routing_unavailable') + + legacy.emitExit(sessionId, 0) + expect(() => router.hasPty(sessionId)).toThrow('daemon_session_routing_unavailable') + }) + + it('keeps replacement-owned synchronous liveness non-authoritative', async () => { + const sessionId = 'replacement-liveness' + const legacy = createAdapter('legacy', [sessionId]) + const router = new DaemonPtyRouter({ + current: createAdapter('current').adapter, + legacy: [legacy.adapter] + }) + + await router.discoverLegacySessions() + legacy.setIdentity(identity('legacy', 11)) + + expect(() => router.hasPty(sessionId)).toThrow('daemon_session_routing_unavailable') + expect(legacy.adapter.hasPty).not.toHaveBeenCalled() + }) + + it('deduplicates concurrent owner probes and memoizes the unique owner', async () => { const current = createAdapter('current') const legacy = createAdapter('legacy', ['surviving-session']) const router = new DaemonPtyRouter({ @@ -183,8 +293,10 @@ describe('DaemonPtyRouter routing safety', () => { legacy: [legacy.adapter] }) - await router.sendSignal('surviving-session', 'SIGINT') - await router.shutdown('surviving-session', { immediate: true }) + await Promise.all([ + router.sendSignal('surviving-session', 'SIGINT'), + router.shutdown('surviving-session', { immediate: true }) + ]) expect(current.adapter.probePtyLiveness).toHaveBeenCalledExactlyOnceWith('surviving-session') expect(legacy.adapter.probePtyLiveness).toHaveBeenCalledExactlyOnceWith('surviving-session') @@ -249,6 +361,193 @@ describe('DaemonPtyRouter routing safety', () => { expect(legacy.adapter.sendSignal).not.toHaveBeenCalled() }) + it('does not collapse an ambiguous route onto a replacement incarnation', async () => { + const sessionId = 'replacement-collision' + const currentSessions = [sessionId] + const current = createAdapter('current', currentSessions, { + alive: [sessionId], + killed: [] + }) + const legacy = createAdapter('legacy', [sessionId], { + alive: [sessionId], + killed: [] + }) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + + await router.reconcileOnStartup(new Set()) + currentSessions.splice(0) + legacy.setIdentity(identity('legacy', 11)) + + await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + expect(current.adapter.sendSignal).not.toHaveBeenCalled() + expect(legacy.adapter.sendSignal).not.toHaveBeenCalled() + }) + + it('does not admit a replacement ambiguous candidate after the other owner exits', async () => { + const sessionId = 'stale-identity-collision' + const current = createAdapter('current', [sessionId], { + alive: [sessionId], + killed: [] + }) + const legacy = createAdapter('legacy', [sessionId], { + alive: [sessionId], + killed: [] + }) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + const previous = identity('legacy', 10) + const replacement = identity('legacy', 11) + + await router.reconcileOnStartup(new Set()) + legacy.setIdentity(replacement) + await router.discoverLegacySessions() + legacy.emitIdentityChange(previous, replacement) + current.emitExit(sessionId, 0) + + expect(router.getSessionRouteState(sessionId)).toBe('unavailable') + await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + expect(current.adapter.sendSignal).not.toHaveBeenCalled() + expect(legacy.adapter.sendSignal).not.toHaveBeenCalled() + }) + + it('does not transfer an owned route from replacement output', async () => { + const sessionId = 'replacement-output' + const current = createAdapter('current') + const legacy = createAdapter('legacy', [sessionId]) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + const data = vi.fn() + router.onData(data) + + await router.discoverLegacySessions() + legacy.setIdentity(identity('legacy', 11)) + legacy.emitData(sessionId, 'replacement frame') + + expect(data).not.toHaveBeenCalled() + expect(router.getSessionRouteState(sessionId)).toBe('unavailable') + await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + }) + + it('rejects an owner probe whose daemon incarnation changes in flight', async () => { + const sessionId = 'probe-replacement-race' + 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] + }) + const previous = identity('legacy', 10) + const replacement = identity('legacy', 11) + + const signaling = router.sendSignal(sessionId, 'SIGTERM') + await vi.waitFor(() => expect(finishProbe).toBeDefined()) + legacy.setIdentity(replacement) + legacy.emitIdentityChange(previous, replacement) + finishProbe?.(true) + + await expect(signaling).rejects.toThrow('daemon_session_routing_unavailable') + expect(current.adapter.sendSignal).not.toHaveBeenCalled() + expect(legacy.adapter.sendSignal).not.toHaveBeenCalled() + }) + + it('does not apply reconciliation after an earlier adapter is replaced', async () => { + const sessionId = 'stale-reconciliation' + const current = createAdapter('current', [], { + alive: [sessionId], + killed: [] + }) + const legacy = createAdapter('legacy') + let finishLegacyReconciliation: + | ((result: { alive: string[]; killed: string[] }) => void) + | undefined + vi.mocked(legacy.adapter.reconcileOnStartup).mockReturnValue( + new Promise((resolve) => { + finishLegacyReconciliation = resolve + }) + ) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + const previous = identity('current', 20) + const replacement = identity('current', 21) + + const reconciling = router.reconcileOnStartup(new Set()) + await vi.waitFor(() => expect(finishLegacyReconciliation).toBeDefined()) + current.setIdentity(replacement) + current.emitIdentityChange(previous, replacement) + await router.spawn({ + sessionId, + isNewSession: true, + cols: 80, + rows: 24 + }) + finishLegacyReconciliation?.({ alive: [], killed: [] }) + + await expect(reconciling).rejects.toThrow( + 'daemon incarnation changed during startup reconciliation' + ) + expect(router.getSessionRouteState(sessionId)).toBe('owned') + await router.sendSignal(sessionId, 'SIGTERM') + expect(current.adapter.sendSignal).toHaveBeenCalledExactlyOnceWith(sessionId, 'SIGTERM') + }) + + it('rejects a conclusive-absence result when discovery becomes incomplete in flight', async () => { + const sessionId = 'stale-absence-probe' + const current = createAdapter('current') + const legacy = createAdapter('legacy') + let finishCurrentProbe: ((result: boolean) => void) | undefined + let finishLegacyProbe: ((result: boolean) => void) | undefined + vi.mocked(current.adapter.probePtyLiveness).mockReturnValue( + new Promise((resolve) => { + finishCurrentProbe = resolve + }) + ) + vi.mocked(legacy.adapter.probePtyLiveness).mockReturnValue( + new Promise((resolve) => { + finishLegacyProbe = resolve + }) + ) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + + const signaling = router.sendSignal(sessionId, 'SIGTERM') + await vi.waitFor(() => { + expect(finishCurrentProbe).toBeDefined() + expect(finishLegacyProbe).toBeDefined() + }) + vi.mocked(legacy.adapter.listProcesses).mockRejectedValueOnce(new Error('listing unavailable')) + vi.spyOn(console, 'warn').mockImplementation(() => {}) + await router.discoverLegacySessions() + finishCurrentProbe?.(false) + finishLegacyProbe?.(false) + + await expect(signaling).rejects.toThrow('daemon_session_routing_unavailable') + expect(current.adapter.sendSignal).not.toHaveBeenCalled() + expect(legacy.adapter.sendSignal).not.toHaveBeenCalled() + }) + it('keeps a canonical spawn collision ambiguous', async () => { const current = createAdapter('current') const legacy = createAdapter('legacy', ['canonical-session']) @@ -279,7 +578,7 @@ describe('DaemonPtyRouter routing safety', () => { expect(legacy.adapter.sendSignal).not.toHaveBeenCalled() }) - it('scopes unavailable tombstones to the authenticated daemon incarnation', async () => { + it('keeps an existing id tombstoned when a replacement reports the same id', async () => { const sessionId = 'legacy-session' const current = createAdapter('current') const legacy = createAdapter('legacy', [sessionId]) @@ -301,9 +600,31 @@ describe('DaemonPtyRouter routing safety', () => { expect(current.adapter.sendSignal).not.toHaveBeenCalled() await router.discoverLegacySessions() - expect(router.getSessionRouteState(sessionId)).toBe('owned') - await router.sendSignal(sessionId, 'SIGTERM') - expect(legacy.adapter.sendSignal).toHaveBeenCalledWith(sessionId, 'SIGTERM') + expect(router.getSessionRouteState(sessionId)).toBe('unavailable') + await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + expect(legacy.adapter.sendSignal).not.toHaveBeenCalled() + }) + + it('does not transfer an owned route from replacement discovery', async () => { + const sessionId = 'replacement-discovery' + const current = createAdapter('current') + const legacy = createAdapter('legacy', [sessionId]) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + + await router.discoverLegacySessions() + legacy.setIdentity(identity('legacy', 11)) + await router.discoverLegacySessions() + + expect(router.getSessionRouteState(sessionId)).toBe('unavailable') + await expect(router.sendSignal(sessionId, 'SIGTERM')).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + expect(legacy.adapter.sendSignal).not.toHaveBeenCalled() }) it('allows explicit fresh reuse after the tombstone owner is replaced', async () => { @@ -389,6 +710,166 @@ describe('DaemonPtyRouter routing safety', () => { expect(legacy.adapter.ackColdRestore).not.toHaveBeenCalled() }) + it('keeps the newest snapshot producer when captures finish out of order', async () => { + const sessionId = 'concurrent-snapshot' + const current = createAdapter('current') + const legacy = createAdapter('legacy', [sessionId]) + let finishFirstCapture: ((snapshot: null) => void) | undefined + vi.mocked(legacy.adapter.getBufferSnapshot) + .mockImplementationOnce( + () => + new Promise((resolve) => { + finishFirstCapture = resolve + }) + ) + .mockResolvedValueOnce({ + data: 'newest frame', + cols: 80, + rows: 24, + seq: 2, + source: 'headless' + }) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + + await router.discoverLegacySessions() + const firstCapture = router.getBufferSnapshot(sessionId) + await vi.waitFor(() => expect(finishFirstCapture).toBeDefined()) + await router.getBufferSnapshot(sessionId) + finishFirstCapture?.(null) + await firstCapture + router.ackColdRestore(sessionId) + + expect(legacy.adapter.ackColdRestore).toHaveBeenCalledExactlyOnceWith(sessionId) + }) + + it('does not acknowledge an old snapshot against a replacement daemon', async () => { + const sessionId = 'replacement-snapshot' + const current = createAdapter('current') + const legacy = createAdapter('legacy', [sessionId]) + vi.mocked(legacy.adapter.getBufferSnapshot).mockResolvedValue({ + data: 'legacy frame', + cols: 80, + rows: 24, + seq: 1, + source: 'headless' + }) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + const previous = identity('legacy', 10) + const replacement = identity('legacy', 11) + + await router.discoverLegacySessions() + await router.getBufferSnapshot(sessionId) + legacy.setIdentity(replacement) + legacy.emitIdentityChange(previous, replacement) + router.ackColdRestore(sessionId) + + expect(legacy.adapter.ackColdRestore).not.toHaveBeenCalled() + }) + + it('does not acknowledge a snapshot whose producer changes incarnation in flight', async () => { + const sessionId = 'in-flight-replacement-snapshot' + const current = createAdapter('current') + const legacy = createAdapter('legacy', [sessionId]) + let finishSnapshot: + | ((snapshot: { + data: string + cols: number + rows: number + seq: number + source: 'headless' + }) => void) + | undefined + vi.mocked(legacy.adapter.getBufferSnapshot).mockImplementation( + () => + new Promise((resolve) => { + finishSnapshot = resolve + }) + ) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + const previous = identity('legacy', 10) + const replacement = identity('legacy', 11) + + await router.discoverLegacySessions() + const capturing = router.getBufferSnapshot(sessionId) + await vi.waitFor(() => expect(finishSnapshot).toBeDefined()) + legacy.setIdentity(replacement) + legacy.emitIdentityChange(previous, replacement) + finishSnapshot?.({ + data: 'legacy frame', + cols: 80, + rows: 24, + seq: 1, + source: 'headless' + }) + await capturing + router.ackColdRestore(sessionId) + + expect(legacy.adapter.ackColdRestore).not.toHaveBeenCalled() + }) + + it('drops pending snapshot acknowledgements only for the exiting producer', async () => { + const sessionId = 'exiting-snapshot' + const current = createAdapter('current') + const legacy = createAdapter('legacy', [sessionId]) + vi.mocked(legacy.adapter.getBufferSnapshot).mockResolvedValue({ + data: 'legacy frame', + cols: 80, + rows: 24, + seq: 1, + source: 'headless' + }) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [legacy.adapter] + }) + + await router.discoverLegacySessions() + await router.getBufferSnapshot(sessionId) + current.emitExit(sessionId, 0) + router.ackColdRestore(sessionId) + expect(legacy.adapter.ackColdRestore).toHaveBeenCalledExactlyOnceWith(sessionId) + + await router.getBufferSnapshot(sessionId) + legacy.emitExit(sessionId, 0) + router.ackColdRestore(sessionId) + expect(legacy.adapter.ackColdRestore).toHaveBeenCalledOnce() + }) + + it('bounds unavailable route retention', async () => { + const killed = Array.from({ length: 1001 }, (_, index) => `closed-${index}`) + const currentSessions: string[] = [] + const current = createAdapter('current', currentSessions, { alive: [], killed }) + const router = new DaemonPtyRouter({ + current: current.adapter, + legacy: [] + }) + const exit = vi.fn() + router.onExit(exit) + + await router.reconcileOnStartup(new Set()) + + expect(router.getSessionRouteState('closed-0')).toBeNull() + expect(router.getSessionRouteState('closed-1')).toBe('unavailable') + expect(router.getSessionRouteState('closed-1000')).toBe('unavailable') + + currentSessions.push('closed-0') + await expect(router.sendSignal('closed-0', 'SIGTERM')).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + expect(() => current.emitExit('closed-0', 0)).not.toThrow() + expect(exit).not.toHaveBeenCalled() + expect(current.adapter.sendSignal).not.toHaveBeenCalled() + }) + it('passes folder and git-worktree keys through the same reconciliation path', async () => { const current = createAdapter('current') const legacy = createAdapter('legacy') diff --git a/src/main/daemon/daemon-pty-router-session-routing.ts b/src/main/daemon/daemon-pty-router-session-routing.ts index a8043213adb..c585dcf71fc 100644 --- a/src/main/daemon/daemon-pty-router-session-routing.ts +++ b/src/main/daemon/daemon-pty-router-session-routing.ts @@ -1,11 +1,16 @@ import type { PtyProviderBufferSnapshot, PtySpawnOptions, PtySpawnResult } from '../providers/types' import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import type { DaemonEndpointIdentity } from './daemon-hello-protocol' import { DaemonSessionRouteTable } from './daemon-session-route-table' import { DaemonSnapshotAcknowledgementRoutes } from './daemon-snapshot-acknowledgement-routes' export class DaemonPtyRouterSessionRouting { private readonly routes: DaemonSessionRouteTable private readonly snapshotAcks = new DaemonSnapshotAcknowledgementRoutes() + private readonly freshSpawnsInFlight = new Map< + string, + { owner: DaemonPtyAdapter; count: number } + >() private readonly identityUnsubscribes: (() => void)[] = [] constructor( @@ -16,16 +21,22 @@ export class DaemonPtyRouterSessionRouting { for (const adapter of adapters) { this.identityUnsubscribes.push( adapter.onDaemonIdentityChanged(({ previous }) => { + this.snapshotAcks.dropAdapterIncarnation(adapter, previous) this.routes.markOwnerUnavailable(adapter, previous) }) ) } } - recordDiscovery(adapter: DaemonPtyAdapter, sessions: readonly { id: string }[]): void { + recordDiscovery( + adapter: DaemonPtyAdapter, + sessions: readonly { id: string }[], + incarnation?: DaemonEndpointIdentity | null + ): void { this.routes.recordCompleteDiscovery( adapter, - sessions.map((session) => session.id) + sessions.map((session) => session.id), + incarnation ) } @@ -51,6 +62,32 @@ export class DaemonPtyRouterSessionRouting { this.routes.recordOwned(result.id, target) } + beginSpawn(opts: PtySpawnOptions, target: DaemonPtyAdapter): void { + if (!opts.isNewSession || !opts.sessionId) { + return + } + const inFlight = this.freshSpawnsInFlight.get(opts.sessionId) + this.freshSpawnsInFlight.set(opts.sessionId, { + owner: target, + count: (inFlight?.count ?? 0) + 1 + }) + } + + endSpawn(opts: PtySpawnOptions): void { + if (!opts.isNewSession || !opts.sessionId) { + return + } + const inFlight = this.freshSpawnsInFlight.get(opts.sessionId) + if (!inFlight || inFlight.count <= 1) { + this.freshSpawnsInFlight.delete(opts.sessionId) + return + } + this.freshSpawnsInFlight.set(opts.sessionId, { + owner: inFlight.owner, + count: inFlight.count - 1 + }) + } + async owner(sessionId: string): Promise { return await this.routes.resolveOwner(sessionId) } @@ -72,14 +109,7 @@ export class DaemonPtyRouterSessionRouting { } hasPty(sessionId: string): boolean { - const route = this.routes.get(sessionId) - if (route?.state === 'unavailable') { - return false - } - if (route?.state === 'owned') { - return route.owner.hasPty(sessionId) - } - return this.adapters.some((adapter) => adapter.hasPty(sessionId)) + return this.routes.hasPty(sessionId) } async probePtyLiveness(sessionId: string): Promise { @@ -99,9 +129,7 @@ export class DaemonPtyRouterSessionRouting { opts?: { scrollbackRows?: number } ): Promise { const owner = await this.owner(sessionId) - const snapshot = await owner.getBufferSnapshot(sessionId, opts) - this.snapshotAcks.record(sessionId, snapshot, owner) - return snapshot + return await this.snapshotAcks.capture(sessionId, opts, owner, this.adapters) } acknowledgeBufferSnapshot(sessionId: string): void { @@ -120,17 +148,37 @@ export class DaemonPtyRouterSessionRouting { } if (!keepHistory) { this.routes.markUnavailable(sessionId, owner) - this.snapshotAcks.drop(sessionId) + this.snapshotAcks.dropForProducer(sessionId, owner) } } - recordReconciledAlive(sessionId: string, owner: DaemonPtyAdapter): void { - this.routes.recordOwned(sessionId, owner) + recordReconciledAlive( + sessionId: string, + owner: DaemonPtyAdapter, + incarnation?: DaemonEndpointIdentity | null + ): void { + this.routes.recordOwnedIncarnation( + sessionId, + owner, + incarnation ?? owner.getLastAuthenticatedDaemonIdentity() + ) } - recordReconciledKilled(sessionId: string, owner: DaemonPtyAdapter): void { - this.routes.markUnavailable(sessionId, owner) - this.snapshotAcks.drop(sessionId) + recordReconciledKilled( + sessionId: string, + owner: DaemonPtyAdapter, + incarnation?: DaemonEndpointIdentity | null + ): void { + this.routes.markUnavailable( + sessionId, + owner, + incarnation ?? owner.getLastAuthenticatedDaemonIdentity() + ) + this.snapshotAcks.dropForProducer( + sessionId, + owner, + incarnation ?? owner.getLastAuthenticatedDaemonIdentity() + ) } shouldForwardEvent(sessionId: string, owner: DaemonPtyAdapter): boolean { @@ -138,18 +186,36 @@ 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) + this.routes.recordOwned(sessionId, owner) + return false + } return this.routes.recordDataOwner(sessionId, owner) } recordExit(sessionId: string, owner: DaemonPtyAdapter): boolean { + const freshSpawn = this.freshSpawnsInFlight.get(sessionId) + if (freshSpawn && freshSpawn.owner !== owner) { + if (this.routes.get(sessionId)) { + this.routes.markUnavailable(sessionId, owner) + } + this.snapshotAcks.dropForProducer(sessionId, owner) + return false + } const shouldForward = this.shouldForwardEvent(sessionId, owner) this.routes.markUnavailable(sessionId, owner) + this.snapshotAcks.dropForProducer(sessionId, owner) return shouldForward } - markAdapterUnavailable(adapter: DaemonPtyAdapter): void { + markAdapterUnavailable( + adapter: DaemonPtyAdapter, + incarnation = adapter.getLastAuthenticatedDaemonIdentity() + ): void { if (this.adapters.includes(adapter)) { - this.routes.markOwnerUnavailable(adapter) + this.routes.markOwnerUnavailable(adapter, incarnation) } } @@ -165,5 +231,7 @@ export class DaemonPtyRouterSessionRouting { for (const unsubscribe of this.identityUnsubscribes.splice(0)) { unsubscribe() } + this.freshSpawnsInFlight.clear() + this.snapshotAcks.clear() } } diff --git a/src/main/daemon/daemon-pty-router.test.ts b/src/main/daemon/daemon-pty-router.test.ts index ab49db4cb21..0771d691b69 100644 --- a/src/main/daemon/daemon-pty-router.test.ts +++ b/src/main/daemon/daemon-pty-router.test.ts @@ -533,6 +533,41 @@ describe('DaemonPtyRouter', () => { ) }) + it('fences foreign stream events received before a fresh spawn reply', async () => { + const sessionId = 'fresh-stale-events' + const current = createAdapter('current') + const legacy = createAdapter('legacy') + let resolveSpawn: ((result: PtySpawnResult) => void) | undefined + vi.mocked(current.spawn).mockReturnValue( + new Promise((resolve) => { + resolveSpawn = resolve + }) + ) + const router = new DaemonPtyRouter({ current, legacy: [legacy] }) + const data = vi.fn() + const exit = vi.fn() + router.onData(data) + router.onExit(exit) + + const spawning = router.spawn({ + sessionId, + isNewSession: true, + cols: 80, + rows: 24 + }) + legacy.emitData(sessionId, 'stale legacy frame') + legacy.emitExit(sessionId, 0) + resolveSpawn?.({ id: sessionId }) + await spawning + + expect(data).not.toHaveBeenCalled() + expect(exit).not.toHaveBeenCalled() + expect(router.getSessionRouteState(sessionId)).toBe('owned') + await router.sendSignal(sessionId, 'SIGTERM') + expect(current.sendSignal).toHaveBeenCalledExactlyOnceWith(sessionId, 'SIGTERM') + expect(legacy.sendSignal).not.toHaveBeenCalled() + }) + it('forwards gap events and explicit sequence accounting from every adapter', () => { const current = createAdapter('current') const legacy = createAdapter('legacy') diff --git a/src/main/daemon/daemon-pty-router.ts b/src/main/daemon/daemon-pty-router.ts index ca31abcaa5a..8e616e5e8ae 100644 --- a/src/main/daemon/daemon-pty-router.ts +++ b/src/main/daemon/daemon-pty-router.ts @@ -12,6 +12,7 @@ import type { PtyProcessInspection } from '../providers/pty-process-inspection' import { shouldHandoffDaemonHistory } from './daemon-history-handoff' import type { DaemonPtyRouterDataEvent, DaemonPtyRouterExitEvent } from './daemon-pty-router-events' import { DaemonPtyRouterSessionRouting } from './daemon-pty-router-session-routing' +import { sameDaemonIncarnation } from './daemon-session-route' export class DaemonPtyRouter implements IPtyProvider { private current: DaemonPtyAdapter @@ -45,8 +46,13 @@ export class DaemonPtyRouter implements IPtyProvider { async discoverLegacySessions(): Promise { for (const adapter of this.legacy) { try { + const before = adapter.getLastAuthenticatedDaemonIdentity() const sessions = await adapter.listProcesses() - this.routing.recordDiscovery(adapter, sessions) + const incarnation = adapter.getLastAuthenticatedDaemonIdentity() + if (before && !sameDaemonIncarnation(before, incarnation)) { + throw new Error('daemon incarnation changed during session discovery') + } + this.routing.recordDiscovery(adapter, sessions, incarnation) } catch (error) { this.routing.recordDiscoveryFailure(adapter) console.warn('[daemon] Failed to discover legacy daemon sessions', error) @@ -57,9 +63,14 @@ export class DaemonPtyRouter implements IPtyProvider { async spawn(opts: PtySpawnOptions): Promise { const targetResult = this.routing.spawnTarget(opts) const target = targetResult instanceof Promise ? await targetResult : targetResult - const result = await target.spawn(opts) - this.routing.recordSpawn(result, target, opts) - return result + this.routing.beginSpawn(opts, target) + try { + const result = await target.spawn(opts) + this.routing.recordSpawn(result, target, opts) + return result + } finally { + this.routing.endSpawn(opts) + } } supportsGitCredentialGuardHost(sessionId?: string): boolean { @@ -241,8 +252,21 @@ export class DaemonPtyRouter implements IPtyProvider { }> { const alive: string[] = [] const killed: string[] = [] + const reconciled: { + adapter: DaemonPtyAdapter + incarnation: ReturnType + alive: string[] + killed: string[] + }[] = [] for (const adapter of this.allAdapters()) { + const before = adapter.getLastAuthenticatedDaemonIdentity() const result = await adapter.reconcileOnStartup(validWorktreeIds) + const incarnation = adapter.getLastAuthenticatedDaemonIdentity() + if (before && !sameDaemonIncarnation(before, incarnation)) { + this.routing.markAdapterUnavailable(adapter, before) + throw new Error('daemon incarnation changed during startup reconciliation') + } + reconciled.push({ adapter, incarnation, ...result }) // Why: daemon startup can reconcile many restored sessions; spreading // those arrays into push can exceed JavaScript's argument limit. for (const id of result.alive) { @@ -251,11 +275,21 @@ export class DaemonPtyRouter implements IPtyProvider { for (const id of result.killed) { killed.push(id) } - for (const id of result.alive) { - this.routing.recordReconciledAlive(id, adapter) + } + for (const { adapter, incarnation } of reconciled) { + if (!sameDaemonIncarnation(incarnation, adapter.getLastAuthenticatedDaemonIdentity())) { + this.routing.markAdapterUnavailable(adapter, incarnation) + throw new Error('daemon incarnation changed during startup reconciliation') } - for (const id of result.killed) { - this.routing.recordReconciledKilled(id, adapter) + } + for (const { adapter, incarnation, alive: sessionIds } of reconciled) { + for (const id of sessionIds) { + this.routing.recordReconciledAlive(id, adapter, incarnation) + } + } + for (const { adapter, incarnation, killed: sessionIds } of reconciled) { + for (const id of sessionIds) { + this.routing.recordReconciledKilled(id, adapter, incarnation) } } return { alive, killed } diff --git a/src/main/daemon/daemon-session-owner-probes.ts b/src/main/daemon/daemon-session-owner-probes.ts new file mode 100644 index 00000000000..3287ba34f6b --- /dev/null +++ b/src/main/daemon/daemon-session-owner-probes.ts @@ -0,0 +1,49 @@ +import type { DaemonEndpointIdentity } from './daemon-hello-protocol' +import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import { sameDaemonIncarnation } from './daemon-session-route' + +export type DaemonSessionOwnerProbe = { + owner: DaemonPtyAdapter + result: boolean | null + incarnation: DaemonEndpointIdentity | null +} + +export async function probeDaemonSessionOwner( + owner: DaemonPtyAdapter, + sessionId: string +): Promise { + const before = owner.getLastAuthenticatedDaemonIdentity() + try { + const result = await owner.probePtyLiveness(sessionId) + const incarnation = owner.getLastAuthenticatedDaemonIdentity() + return { + owner, + result: before && !sameDaemonIncarnation(before, incarnation) ? null : result, + incarnation + } + } catch { + return { + owner, + result: null, + incarnation: owner.getLastAuthenticatedDaemonIdentity() + } + } +} + +export function probeDaemonSessionOwnerSync( + owner: DaemonPtyAdapter, + sessionId: string +): DaemonSessionOwnerProbe { + const before = owner.getLastAuthenticatedDaemonIdentity() + try { + const result = owner.hasPty(sessionId) + const incarnation = owner.getLastAuthenticatedDaemonIdentity() + return { + owner, + result: before && !sameDaemonIncarnation(before, incarnation) ? null : result, + incarnation + } + } catch { + return { owner, result: null, incarnation: owner.getLastAuthenticatedDaemonIdentity() } + } +} diff --git a/src/main/daemon/daemon-session-owner-resolution-cache.ts b/src/main/daemon/daemon-session-owner-resolution-cache.ts new file mode 100644 index 00000000000..34e22cd07bb --- /dev/null +++ b/src/main/daemon/daemon-session-owner-resolution-cache.ts @@ -0,0 +1,24 @@ +import type { DaemonPtyAdapter } from './daemon-pty-adapter' + +export class DaemonSessionOwnerResolutionCache { + private readonly pending = new Map>() + + async resolve( + sessionId: string, + start: () => Promise + ): Promise { + const existing = this.pending.get(sessionId) + if (existing) { + return await existing + } + const resolution = start() + this.pending.set(sessionId, resolution) + try { + return await resolution + } finally { + if (this.pending.get(sessionId) === resolution) { + this.pending.delete(sessionId) + } + } + } +} diff --git a/src/main/daemon/daemon-session-owner-resolution.ts b/src/main/daemon/daemon-session-owner-resolution.ts new file mode 100644 index 00000000000..007e4ac9b0e --- /dev/null +++ b/src/main/daemon/daemon-session-owner-resolution.ts @@ -0,0 +1,51 @@ +import type { DaemonEndpointIdentity } from './daemon-hello-protocol' +import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import { probeDaemonSessionOwner } from './daemon-session-owner-probes' +import { + DaemonSessionGoneError, + DaemonSessionOwnerUnknownError, + sameDaemonIncarnation +} from './daemon-session-route' + +type ResolveDaemonSessionOwnerOptions = { + sessionId: string + candidates: readonly DaemonPtyAdapter[] + expectedIncarnations?: ReadonlyMap + discoveryComplete: boolean + transfer: (owner: DaemonPtyAdapter, incarnation: DaemonEndpointIdentity | null) => void + recordAmbiguous: ( + candidates: ReadonlyMap + ) => void +} + +export async function resolveDaemonSessionOwner({ + sessionId, + candidates, + expectedIncarnations, + discoveryComplete, + transfer, + recordAmbiguous +}: ResolveDaemonSessionOwnerOptions): Promise { + const probed = await Promise.all( + candidates.map((owner) => probeDaemonSessionOwner(owner, sessionId)) + ) + const results = probed.map((result) => + sameDaemonIncarnation(result.incarnation, result.owner.getLastAuthenticatedDaemonIdentity()) && + (!expectedIncarnations?.has(result.owner) || + sameDaemonIncarnation(expectedIncarnations.get(result.owner) ?? null, result.incarnation)) + ? result + : { ...result, result: null } + ) + const owners = results.filter(({ result }) => result === true) + if (owners.length === 1 && results.every(({ result }) => result !== null)) { + transfer(owners[0].owner, owners[0].incarnation) + return owners[0].owner + } + if (owners.length > 1) { + recordAmbiguous(new Map(owners.map(({ owner, incarnation }) => [owner, incarnation]))) + } + if (owners.length === 0 && results.every(({ result }) => result === false) && discoveryComplete) { + throw new DaemonSessionGoneError(sessionId) + } + throw new DaemonSessionOwnerUnknownError(sessionId) +} diff --git a/src/main/daemon/daemon-session-owner-sync-resolution.ts b/src/main/daemon/daemon-session-owner-sync-resolution.ts new file mode 100644 index 00000000000..4b7a04950d5 --- /dev/null +++ b/src/main/daemon/daemon-session-owner-sync-resolution.ts @@ -0,0 +1,56 @@ +import type { DaemonEndpointIdentity } from './daemon-hello-protocol' +import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import { probeDaemonSessionOwnerSync } from './daemon-session-owner-probes' +import { + DaemonSessionGoneError, + DaemonSessionOwnerUnknownError, + type DaemonSessionRoute, + DaemonSessionUnavailableError +} from './daemon-session-route' + +type ResolveDaemonSessionOwnerSyncOptions = { + sessionId: string + route: DaemonSessionRoute | undefined + discoveryIncomplete: boolean + adapters: readonly DaemonPtyAdapter[] + transfer: (owner: DaemonPtyAdapter, incarnation: DaemonEndpointIdentity | null) => void + recordAmbiguous: ( + candidates: ReadonlyMap + ) => void +} + +export function resolveDaemonSessionOwnerSync({ + sessionId, + route, + discoveryIncomplete, + adapters, + transfer, + recordAmbiguous +}: ResolveDaemonSessionOwnerSyncOptions): DaemonPtyAdapter { + if (route?.state === 'owned') { + return route.owner + } + if (route?.state === 'unavailable') { + throw new DaemonSessionUnavailableError(sessionId) + } + if (route?.state === 'ambiguous') { + throw new DaemonSessionOwnerUnknownError(sessionId) + } + const results = adapters.map((owner) => probeDaemonSessionOwnerSync(owner, sessionId)) + const owners = results.filter(({ result }) => result === true) + if (results.some(({ result }) => result === null)) { + throw new DaemonSessionOwnerUnknownError(sessionId) + } + if (owners.length === 1) { + transfer(owners[0].owner, owners[0].incarnation) + return owners[0].owner + } + if (owners.length > 1) { + recordAmbiguous(new Map(owners.map(({ owner, incarnation }) => [owner, incarnation]))) + throw new DaemonSessionOwnerUnknownError(sessionId) + } + if (discoveryIncomplete) { + throw new DaemonSessionOwnerUnknownError(sessionId) + } + throw new DaemonSessionGoneError(sessionId) +} diff --git a/src/main/daemon/daemon-session-route-availability.ts b/src/main/daemon/daemon-session-route-availability.ts new file mode 100644 index 00000000000..89860f3d7e9 --- /dev/null +++ b/src/main/daemon/daemon-session-route-availability.ts @@ -0,0 +1,83 @@ +import type { DaemonEndpointIdentity } from './daemon-hello-protocol' +import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import { createOwnedDaemonSessionRoute, sameDaemonIncarnation } from './daemon-session-route' +import type { DaemonSessionRouteStore } from './daemon-session-route-store' + +export function markDaemonSessionUnavailable( + routes: DaemonSessionRouteStore, + sessionId: string, + owner: DaemonPtyAdapter, + incarnation: DaemonEndpointIdentity | null +): void { + const route = routes.get(sessionId) + if (!route) { + routes.set(sessionId, { + state: 'unavailable', + owner, + incarnation + }) + return + } + if ( + route.state === 'owned' && + route.owner === owner && + sameDaemonIncarnation(route.incarnation, incarnation) + ) { + routes.set(sessionId, { + state: 'unavailable', + owner, + incarnation: route.incarnation + }) + return + } + if ( + route.state !== 'ambiguous' || + !route.candidates.has(owner) || + !sameDaemonIncarnation(route.candidates.get(owner) ?? null, incarnation) + ) { + return + } + const candidates = new Map(route.candidates) + candidates.delete(owner) + if (candidates.size === 1) { + for (const [candidate, candidateIncarnation] of candidates) { + routes.set(sessionId, createOwnedDaemonSessionRoute(candidate, candidateIncarnation)) + } + } else if (candidates.size > 1) { + routes.set(sessionId, { state: 'ambiguous', candidates }) + } else { + routes.set(sessionId, { + state: 'unavailable', + owner, + incarnation + }) + } +} + +export function markDaemonSessionOwnerUnavailable( + routes: DaemonSessionRouteStore, + owner: DaemonPtyAdapter, + incarnation: DaemonEndpointIdentity | null +): void { + for (const [sessionId, route] of routes) { + if ( + route.state === 'ambiguous' && + route.candidates.has(owner) && + sameDaemonIncarnation(route.candidates.get(owner) ?? null, incarnation) + ) { + markDaemonSessionUnavailable(routes, sessionId, owner, incarnation) + continue + } + if ( + route.state === 'owned' && + route.owner === owner && + sameDaemonIncarnation(route.incarnation, incarnation) + ) { + routes.set(sessionId, { + state: 'unavailable', + owner, + incarnation: route.incarnation + }) + } + } +} diff --git a/src/main/daemon/daemon-session-route-store.ts b/src/main/daemon/daemon-session-route-store.ts new file mode 100644 index 00000000000..24db8e89b24 --- /dev/null +++ b/src/main/daemon/daemon-session-route-store.ts @@ -0,0 +1,57 @@ +import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import type { DaemonSessionRoute } from './daemon-session-route' + +const MAX_UNAVAILABLE_ROUTES = 1000 + +export class DaemonSessionRouteStore implements Iterable<[string, DaemonSessionRoute]> { + private readonly routes = new Map() + private readonly unavailableRouteIds = new Set() + private unavailableRouteEvicted = false + + get(sessionId: string): DaemonSessionRoute | undefined { + return this.routes.get(sessionId) + } + + set(sessionId: string, route: DaemonSessionRoute): void { + this.routes.set(sessionId, route) + this.unavailableRouteIds.delete(sessionId) + if (route.state !== 'unavailable') { + return + } + this.unavailableRouteIds.add(sessionId) + while (this.unavailableRouteIds.size > MAX_UNAVAILABLE_ROUTES) { + const oldest = this.unavailableRouteIds.values().next().value + if (!oldest) { + return + } + this.unavailableRouteIds.delete(oldest) + if (this.routes.get(oldest)?.state === 'unavailable') { + this.routes.delete(oldest) + this.unavailableRouteEvicted = true + } + } + } + + delete(sessionId: string): void { + this.routes.delete(sessionId) + this.unavailableRouteIds.delete(sessionId) + } + + ownedSessionIds(owner: DaemonPtyAdapter): string[] { + const sessionIds: string[] = [] + for (const [sessionId, route] of this.routes) { + if (route.state === 'owned' && route.owner === owner) { + sessionIds.push(sessionId) + } + } + return sessionIds + } + + hasUnavailableRouteEviction(): boolean { + return this.unavailableRouteEvicted + } + + [Symbol.iterator](): MapIterator<[string, DaemonSessionRoute]> { + return this.routes[Symbol.iterator]() + } +} diff --git a/src/main/daemon/daemon-session-route-table.ts b/src/main/daemon/daemon-session-route-table.ts index a8462e246a4..935426fb1fb 100644 --- a/src/main/daemon/daemon-session-route-table.ts +++ b/src/main/daemon/daemon-session-route-table.ts @@ -3,34 +3,56 @@ import { createOwnedDaemonSessionRoute, DaemonSessionGoneError, DaemonSessionOwnerUnknownError, - type DaemonSessionRoute, DaemonSessionUnavailableError, + isCurrentOwnedDaemonSessionRoute, sameDaemonIncarnation } 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 { resolveDaemonSessionOwnerSync } from './daemon-session-owner-sync-resolution' +import { resolveDaemonSessionOwner } from './daemon-session-owner-resolution' +import { + markDaemonSessionOwnerUnavailable, + markDaemonSessionUnavailable +} from './daemon-session-route-availability' +import { DaemonSessionOwnerResolutionCache } from './daemon-session-owner-resolution-cache' export class DaemonSessionRouteTable { - private readonly routes = new Map() + private readonly routes = new DaemonSessionRouteStore() private readonly incompleteDiscoveries = new Set() + private readonly ownerResolutions = new DaemonSessionOwnerResolutionCache() constructor(private readonly adapters: readonly DaemonPtyAdapter[]) {} - get(sessionId: string): DaemonSessionRoute | undefined { - return this.routes.get(sessionId) - } + readonly get = this.routes.get.bind(this.routes) getOwned(sessionId: string): DaemonPtyAdapter | undefined { const route = this.routes.get(sessionId) - return route?.state === 'owned' ? route.owner : undefined + return route?.state === 'owned' && isCurrentOwnedDaemonSessionRoute(route) + ? route.owner + : undefined } shouldForwardEvent(sessionId: string, owner: DaemonPtyAdapter): boolean { const route = this.routes.get(sessionId) - return !route || (route.state === 'owned' && route.owner === owner) + if (!route) { + return !this.routes.hasUnavailableRouteEviction() + } + return ( + route.state === 'owned' && route.owner === owner && isCurrentOwnedDaemonSessionRoute(route) + ) } recordDataOwner(sessionId: string, owner: DaemonPtyAdapter): boolean { const route = this.routes.get(sessionId) if (!route) { + if (this.routes.hasUnavailableRouteEviction()) { + return false + } this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner)) return true } @@ -38,43 +60,78 @@ export class DaemonSessionRouteTable { this.recordOwned(sessionId, owner) return this.shouldForwardEvent(sessionId, owner) } - if (route.state === 'ambiguous' || route.owner === owner) { - return route.state === 'owned' + if (route.state === 'ambiguous') { + return false + } + if (route.owner === owner) { + if (isCurrentOwnedDaemonSessionRoute(route)) { + return true + } + this.markUnavailable(sessionId, owner, route.incarnation) + return false } this.recordOwned(sessionId, owner) return false } recordOwned(sessionId: string, owner: DaemonPtyAdapter): void { + this.recordOwnedIncarnation(sessionId, owner, owner.getLastAuthenticatedDaemonIdentity()) + } + + recordOwnedIncarnation( + sessionId: string, + owner: DaemonPtyAdapter, + incarnation: ReturnType + ): void { const route = this.routes.get(sessionId) if (!route) { - this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner)) + if (this.routes.hasUnavailableRouteEviction()) { + return + } + this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner, incarnation)) return } if (route.state === 'owned' && route.owner === owner) { - this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner)) + if (!sameDaemonIncarnation(route.incarnation, incarnation)) { + this.markUnavailable(sessionId, owner, route.incarnation) + return + } + this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner, incarnation)) return } if (route.state === 'unavailable') { - const incarnation = owner.getLastAuthenticatedDaemonIdentity() - if (route.owner !== owner || !sameDaemonIncarnation(route.incarnation, incarnation)) { - this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner)) - } return } + if (route.state === 'ambiguous' && route.candidates.has(owner)) { + const candidateIncarnation = route.candidates.get(owner) ?? null + if (!sameDaemonIncarnation(candidateIncarnation, incarnation)) { + this.markUnavailable(sessionId, owner, candidateIncarnation) + return + } + } const candidates = - route.state === 'ambiguous' ? new Set(route.candidates) : new Set([route.owner]) - candidates.add(owner) + route.state === 'ambiguous' + ? new Map(route.candidates) + : new Map([[route.owner, route.incarnation]]) + candidates.set(owner, incarnation) this.routes.set(sessionId, { state: 'ambiguous', candidates }) } - transfer(sessionId: string, owner: DaemonPtyAdapter): void { - this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner)) + transfer( + sessionId: string, + owner: DaemonPtyAdapter, + incarnation = owner.getLastAuthenticatedDaemonIdentity() + ): void { + this.routes.set(sessionId, createOwnedDaemonSessionRoute(owner, incarnation)) } recordFreshOwned(sessionId: string, owner: DaemonPtyAdapter): void { const route = this.routes.get(sessionId) - if (route?.state === 'unavailable' && route.owner === owner) { + if ( + !route || + route.state === 'unavailable' || + (route.state === 'owned' && route.owner === owner && !isCurrentOwnedDaemonSessionRoute(route)) + ) { this.transfer(sessionId, owner) return } @@ -85,142 +142,149 @@ export class DaemonSessionRouteTable { this.incompleteDiscoveries.add(owner) } - recordCompleteDiscovery(owner: DaemonPtyAdapter, sessionIds: readonly string[]): void { + recordCompleteDiscovery( + owner: DaemonPtyAdapter, + sessionIds: readonly string[], + incarnation = owner.getLastAuthenticatedDaemonIdentity() + ): void { for (const sessionId of sessionIds) { - this.recordOwned(sessionId, owner) + this.recordOwnedIncarnation(sessionId, owner, incarnation) } this.incompleteDiscoveries.delete(owner) } - markUnavailable(sessionId: string, owner: DaemonPtyAdapter): void { - const route = this.routes.get(sessionId) - if (!route) { - this.routes.set(sessionId, { - state: 'unavailable', - owner, - incarnation: owner.getLastAuthenticatedDaemonIdentity() - }) - return - } - if (route.state === 'owned' && route.owner === owner) { - this.routes.set(sessionId, { - state: 'unavailable', - owner, - incarnation: route.incarnation - }) - return - } - if (route.state === 'ambiguous' && route.candidates.has(owner)) { - const candidates = new Set(route.candidates) - candidates.delete(owner) - if (candidates.size === 1) { - for (const candidate of candidates) { - this.routes.set(sessionId, createOwnedDaemonSessionRoute(candidate)) - } - } else if (candidates.size > 1) { - this.routes.set(sessionId, { state: 'ambiguous', candidates }) - } else { - this.routes.set(sessionId, { - state: 'unavailable', - owner, - incarnation: owner.getLastAuthenticatedDaemonIdentity() - }) - } - } + markUnavailable( + sessionId: string, + owner: DaemonPtyAdapter, + incarnation = owner.getLastAuthenticatedDaemonIdentity() + ): void { + markDaemonSessionUnavailable(this.routes, sessionId, owner, incarnation) } markOwnerUnavailable( owner: DaemonPtyAdapter, incarnation = owner.getLastAuthenticatedDaemonIdentity() ): void { - for (const [sessionId, route] of this.routes) { - if (route.state === 'ambiguous' && route.candidates.has(owner)) { - this.markUnavailable(sessionId, owner) - continue - } - if ( - route.state === 'owned' && - route.owner === owner && - sameDaemonIncarnation(route.incarnation, incarnation) - ) { - this.routes.set(sessionId, { - state: 'unavailable', - owner, - incarnation: route.incarnation - }) - } - } + markDaemonSessionOwnerUnavailable(this.routes, owner, incarnation) } ownerForFreshSpawn(sessionId: string | undefined, current: DaemonPtyAdapter): DaemonPtyAdapter { - if (!sessionId) { - return current - } - const route = this.routes.get(sessionId) - if (!route) { - return current - } - if (route.state === 'owned' && route.owner === current) { - return current - } - if (route.state === 'unavailable') { - const ownerIncarnation = route.owner.getLastAuthenticatedDaemonIdentity() - if ( - route.incarnation && - ownerIncarnation && - !sameDaemonIncarnation(route.incarnation, ownerIncarnation) - ) { - this.routes.delete(sessionId) - return current - } - throw new DaemonSessionUnavailableError(sessionId) - } - throw new DaemonSessionOwnerUnknownError(sessionId) + return ownerForFreshDaemonSession( + sessionId, + sessionId ? this.routes.get(sessionId) : undefined, + current, + () => sessionId && this.routes.delete(sessionId) + ) } - async resolveOwner(sessionId: string): Promise { + assertFreshSpawnAvailable(sessionId: string | undefined): void { + assertFreshDaemonSessionAvailable( + sessionId, + sessionId ? this.routes.get(sessionId) : undefined, + () => sessionId && this.routes.delete(sessionId) + ) + } + + hasPty(sessionId: string): boolean { const route = this.routes.get(sessionId) - if (route?.state === 'owned') { - return route.owner + if (!route && this.routes.hasUnavailableRouteEviction()) { + throw new DaemonSessionOwnerUnknownError(sessionId) } if (route?.state === 'unavailable') { throw new DaemonSessionUnavailableError(sessionId) } - const candidates = - route?.state === 'ambiguous' && this.incompleteDiscoveries.size === 0 - ? [...route.candidates] - : this.adapters - const results = await Promise.all( - candidates.map(async (owner) => ({ - owner, - result: await probeOwner(owner, sessionId) - })) - ) - const owners = results.filter(({ result }) => result === true).map(({ owner }) => owner) - if (owners.length === 1 && results.every(({ result }) => result !== null)) { - this.transfer(sessionId, owners[0]) - return owners[0] + if (route?.state === 'ambiguous') { + throw new DaemonSessionOwnerUnknownError(sessionId) } - if (owners.length > 1) { - this.routes.set(sessionId, { state: 'ambiguous', candidates: new Set(owners) }) + if (route?.state === 'owned') { + if (!isCurrentOwnedDaemonSessionRoute(route)) { + this.markUnavailable(sessionId, route.owner, route.incarnation) + throw new DaemonSessionUnavailableError(sessionId) + } + return route.owner.hasPty(sessionId) } - if ( - owners.length === 0 && - results.every(({ result }) => result === false) && - this.incompleteDiscoveries.size === 0 - ) { - throw new DaemonSessionGoneError(sessionId) + try { + this.resolveOwnerSync(sessionId) + return true + } catch (error) { + if (error instanceof DaemonSessionGoneError) { + return false + } + throw error } - throw new DaemonSessionOwnerUnknownError(sessionId) + } + + async resolveOwner(sessionId: string): Promise { + const route = this.routes.get(sessionId) + if (!route && this.routes.hasUnavailableRouteEviction()) { + throw new DaemonSessionOwnerUnknownError(sessionId) + } + if (route?.state === 'owned') { + if (isCurrentOwnedDaemonSessionRoute(route)) { + return route.owner + } + this.markUnavailable(sessionId, route.owner, route.incarnation) + throw new DaemonSessionUnavailableError(sessionId) + } + if (route?.state === 'unavailable') { + throw new DaemonSessionUnavailableError(sessionId) + } + return await this.ownerResolutions.resolve(sessionId, async () => { + const discoveryComplete = this.incompleteDiscoveries.size === 0 + const accept = (): void => { + if (this.routes.get(sessionId) !== route) { + throw new DaemonSessionOwnerUnknownError(sessionId) + } + } + try { + return await resolveDaemonSessionOwner({ + sessionId, + candidates: this.adapters, + expectedIncarnations: route?.state === 'ambiguous' ? route.candidates : undefined, + discoveryComplete, + transfer: (owner, incarnation) => { + accept() + this.transfer(sessionId, owner, incarnation) + }, + recordAmbiguous: (ambiguousCandidates) => { + accept() + this.routes.set(sessionId, { + state: 'ambiguous', + candidates: ambiguousCandidates + }) + } + }) + } catch (error) { + if ( + error instanceof DaemonSessionGoneError && + (this.routes.get(sessionId) !== route || + discoveryComplete !== (this.incompleteDiscoveries.size === 0)) + ) { + throw new DaemonSessionOwnerUnknownError(sessionId) + } + throw error + } + }) } async probeLiveness(sessionId: string): Promise { const route = this.routes.get(sessionId) + if (!route && this.routes.hasUnavailableRouteEviction()) { + return null + } if (route?.state === 'unavailable' || route?.state === 'ambiguous') { return null } if (route?.state === 'owned') { - return await probeOwner(route.owner, sessionId) + if (!isCurrentOwnedDaemonSessionRoute(route)) { + this.markUnavailable(sessionId, route.owner, route.incarnation) + return null + } + const result = await probeDaemonSessionOwner(route.owner, sessionId) + return sameDaemonIncarnation(route.incarnation, result.incarnation) && + sameDaemonIncarnation(result.incarnation, result.owner.getLastAuthenticatedDaemonIdentity()) + ? result.result + : null } try { await this.resolveOwner(sessionId) @@ -232,47 +296,23 @@ export class DaemonSessionRouteTable { resolveOwnerSync(sessionId: string): DaemonPtyAdapter { const route = this.routes.get(sessionId) - if (route?.state === 'owned') { - return route.owner + if (!route && this.routes.hasUnavailableRouteEviction()) { + throw new DaemonSessionOwnerUnknownError(sessionId) } - if (route?.state === 'unavailable') { + if (route?.state === 'owned' && !isCurrentOwnedDaemonSessionRoute(route)) { + this.markUnavailable(sessionId, route.owner, route.incarnation) throw new DaemonSessionUnavailableError(sessionId) } - if (route?.state === 'ambiguous' || this.incompleteDiscoveries.size > 0) { - throw new DaemonSessionOwnerUnknownError(sessionId) - } - const results = this.adapters.map((owner) => ({ - owner, - result: hasPty(owner, sessionId) - })) - const owners = results.filter(({ result }) => result === true).map(({ owner }) => owner) - if (results.some(({ result }) => result === null)) { - throw new DaemonSessionOwnerUnknownError(sessionId) - } - if (owners.length === 1) { - this.transfer(sessionId, owners[0]) - return owners[0] - } - if (owners.length > 1) { - this.routes.set(sessionId, { state: 'ambiguous', candidates: new Set(owners) }) - throw new DaemonSessionOwnerUnknownError(sessionId) - } - throw new DaemonSessionGoneError(sessionId) + return resolveDaemonSessionOwnerSync({ + sessionId, + route, + discoveryIncomplete: this.incompleteDiscoveries.size > 0, + adapters: this.adapters, + transfer: (owner, incarnation) => this.transfer(sessionId, owner, incarnation), + recordAmbiguous: (candidates) => + this.routes.set(sessionId, { state: 'ambiguous', candidates }) + }) } -} -async function probeOwner(owner: DaemonPtyAdapter, sessionId: string): Promise { - try { - return await owner.probePtyLiveness(sessionId) - } catch { - return null - } -} - -function hasPty(owner: DaemonPtyAdapter, sessionId: string): boolean | null { - try { - return owner.hasPty(sessionId) - } catch { - return null - } + readonly getOwnedSessionIds = this.routes.ownedSessionIds.bind(this.routes) } diff --git a/src/main/daemon/daemon-session-route.ts b/src/main/daemon/daemon-session-route.ts index 18882170c25..249e07f684e 100644 --- a/src/main/daemon/daemon-session-route.ts +++ b/src/main/daemon/daemon-session-route.ts @@ -15,7 +15,7 @@ export type DaemonSessionRoute = } | { state: 'ambiguous' - candidates: ReadonlySet + candidates: ReadonlyMap } export class DaemonSessionOwnerUnknownError extends Error { @@ -40,12 +40,13 @@ export class DaemonSessionGoneError extends Error { } export function createOwnedDaemonSessionRoute( - owner: DaemonPtyAdapter + owner: DaemonPtyAdapter, + incarnation = owner.getLastAuthenticatedDaemonIdentity() ): Extract { return { state: 'owned', owner, - incarnation: owner.getLastAuthenticatedDaemonIdentity() + incarnation } } @@ -62,3 +63,12 @@ export function sameDaemonIncarnation( left.launchNonce === right.launchNonce ) } + +export function isCurrentOwnedDaemonSessionRoute( + route: Extract +): boolean { + if (typeof route.owner.matchesLastAuthenticatedDaemonIdentity === 'function') { + return route.owner.matchesLastAuthenticatedDaemonIdentity(route.incarnation) + } + return sameDaemonIncarnation(route.incarnation, route.owner.getLastAuthenticatedDaemonIdentity()) +} diff --git a/src/main/daemon/daemon-snapshot-acknowledgement-routes.ts b/src/main/daemon/daemon-snapshot-acknowledgement-routes.ts index ad0b1f7b812..21e7c504133 100644 --- a/src/main/daemon/daemon-snapshot-acknowledgement-routes.ts +++ b/src/main/daemon/daemon-snapshot-acknowledgement-routes.ts @@ -1,8 +1,21 @@ import type { IPtyProvider, PtyProviderBufferSnapshot } from '../providers/types' import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import type { DaemonEndpointIdentity } from './daemon-hello-protocol' +import { sameDaemonIncarnation } from './daemon-session-route' export class DaemonSnapshotAcknowledgementRoutes { - private readonly producers = new Map() + private readonly producers = new Map< + string, + { adapter: DaemonPtyAdapter; incarnation: DaemonEndpointIdentity | null } + >() + private readonly pendingCaptures = new Map< + string, + { + token: symbol + adapter: DaemonPtyAdapter | undefined + incarnation: DaemonEndpointIdentity | null + } + >() async capture( sessionId: string, @@ -10,44 +23,95 @@ export class DaemonSnapshotAcknowledgementRoutes { provider: IPtyProvider, adapters: readonly DaemonPtyAdapter[] ): Promise { - const snapshot = (await provider.getBufferSnapshot?.(sessionId, opts)) ?? null - this.recordAdapter(sessionId, snapshot, provider, adapters) - return snapshot + const producer = adapters.find((adapter) => adapter === provider) + const before = producer?.getLastAuthenticatedDaemonIdentity() ?? null + const token = Symbol(sessionId) + this.pendingCaptures.set(sessionId, { token, adapter: producer, incarnation: before }) + try { + const snapshot = (await provider.getBufferSnapshot?.(sessionId, opts)) ?? null + if (this.pendingCaptures.get(sessionId)?.token !== token) { + return snapshot + } + const incarnation = producer?.getLastAuthenticatedDaemonIdentity() ?? null + if (before && !sameDaemonIncarnation(before, incarnation)) { + this.producers.delete(sessionId) + return snapshot + } + this.record(sessionId, snapshot, producer, incarnation) + return snapshot + } finally { + if (this.pendingCaptures.get(sessionId)?.token === token) { + this.pendingCaptures.delete(sessionId) + } + } } - record( + private record( sessionId: string, snapshot: PtyProviderBufferSnapshot | null, - producer: DaemonPtyAdapter + producer: DaemonPtyAdapter | undefined, + incarnation: DaemonEndpointIdentity | null ): void { - if (snapshot) { - this.producers.set(sessionId, producer) + if (snapshot && producer) { + this.producers.set(sessionId, { + adapter: producer, + incarnation + }) } else { this.producers.delete(sessionId) } } - recordAdapter( + dropForProducer( sessionId: string, - snapshot: PtyProviderBufferSnapshot | null, - provider: IPtyProvider, - adapters: readonly DaemonPtyAdapter[] + adapter: DaemonPtyAdapter, + incarnation = adapter.getLastAuthenticatedDaemonIdentity() ): void { - const producer = adapters.find((adapter) => adapter === provider) - if (producer) { - this.record(sessionId, snapshot, producer) - } else { + const producer = this.producers.get(sessionId) + if (producer?.adapter === adapter && sameDaemonIncarnation(producer.incarnation, incarnation)) { this.producers.delete(sessionId) } + const pending = this.pendingCaptures.get(sessionId) + if (pending?.adapter === adapter && sameDaemonIncarnation(pending.incarnation, incarnation)) { + this.pendingCaptures.delete(sessionId) + } + } + + dropAdapterIncarnation( + adapter: DaemonPtyAdapter, + incarnation: DaemonEndpointIdentity | null + ): void { + for (const [sessionId, producer] of this.producers) { + if ( + producer.adapter === adapter && + sameDaemonIncarnation(producer.incarnation, incarnation) + ) { + this.producers.delete(sessionId) + } + } + for (const [sessionId, pending] of this.pendingCaptures) { + if (pending.adapter === adapter && sameDaemonIncarnation(pending.incarnation, incarnation)) { + this.pendingCaptures.delete(sessionId) + } + } } acknowledge(sessionId: string): void { const producer = this.producers.get(sessionId) this.producers.delete(sessionId) - producer?.ackColdRestore(sessionId) + if ( + producer && + sameDaemonIncarnation( + producer.incarnation, + producer.adapter.getLastAuthenticatedDaemonIdentity() + ) + ) { + producer.adapter.ackColdRestore(sessionId) + } } - drop(sessionId: string): void { - this.producers.delete(sessionId) + clear(): void { + this.producers.clear() + this.pendingCaptures.clear() } } diff --git a/src/main/daemon/degraded-daemon-pty-provider.test.ts b/src/main/daemon/degraded-daemon-pty-provider.test.ts index 56287fd7299..e393eb104c0 100644 --- a/src/main/daemon/degraded-daemon-pty-provider.test.ts +++ b/src/main/daemon/degraded-daemon-pty-provider.test.ts @@ -36,6 +36,9 @@ function createProvider( providesAgentSessionOwnerListings: vi.fn(() => authoritativeOwnerListings), write: vi.fn(), resize: vi.fn(), + pauseProducer: vi.fn(), + resumeProducer: vi.fn(), + setPtyBackgrounded: vi.fn(), shutdown: vi.fn(async (id: string) => { const idx = sessions.indexOf(id) if (idx !== -1) { @@ -124,6 +127,8 @@ function createDaemonAdapter( return { ...createProvider(label, sessions, true), protocolVersion: 13, + getLastAuthenticatedDaemonIdentity: vi.fn(() => null), + onDaemonIdentityChanged: vi.fn(() => () => {}), listSessions: vi.fn(async () => []), ackColdRestore: vi.fn(), clearTombstone: vi.fn(), @@ -228,20 +233,71 @@ describe('DegradedDaemonPtyProvider', () => { expect(fallback.write).toHaveBeenCalledWith(fresh.id, 'new\n') }) - it('routes a previously daemon-backed id to fallback after daemon exit removes the mapping', async () => { + it('fences existing ids after incomplete daemon discovery while allowing explicit fresh ids', async () => { + const sessionId = 'uncertain-session' + const current = createDaemonAdapter('daemon') + const legacy = createDaemonAdapter('legacy') + const fallback = createProvider('fallback') + vi.mocked(legacy.listProcesses).mockRejectedValueOnce(new Error('listing unavailable')) + vi.mocked(legacy.probePtyLiveness).mockResolvedValue(null) + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + const provider = new DegradedDaemonPtyProvider({ current, legacy: [legacy], fallback }) + + await provider.discoverDaemonSessions() + + await expect(provider.spawn({ sessionId, cols: 80, rows: 24 })).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + await expect( + provider.spawn({ sessionId: 'fresh-session', isNewSession: true, cols: 80, rows: 24 }) + ).resolves.toMatchObject({ id: 'fresh-session' }) + expect(fallback.spawn).toHaveBeenCalledExactlyOnceWith({ + sessionId: 'fresh-session', + isNewSession: true, + cols: 80, + rows: 24 + }) + warn.mockRestore() + }) + + it('keeps cross-generation daemon collisions away from the fallback', async () => { + const sessionId = 'daemon-collision' + const current = createDaemonAdapter('current', [sessionId]) + const legacy = createDaemonAdapter('legacy', [sessionId]) + const fallback = createProvider('fallback') + const provider = new DegradedDaemonPtyProvider({ current, legacy: [legacy], fallback }) + const data = vi.fn() + provider.onData(data) + + await provider.discoverDaemonSessions() + current.emitData(sessionId, 'current output') + legacy.emitData(sessionId, 'legacy output') + + await expect(provider.spawn({ sessionId, cols: 80, rows: 24 })).rejects.toThrow( + 'daemon_session_routing_unavailable' + ) + expect(() => provider.pauseProducer(sessionId)).not.toThrow() + expect(() => provider.resumeProducer(sessionId)).not.toThrow() + expect(() => provider.setPtyBackgrounded(sessionId, true)).not.toThrow() + expect(data).not.toHaveBeenCalled() + expect(fallback.spawn).not.toHaveBeenCalled() + expect(fallback.pauseProducer).not.toHaveBeenCalled() + expect(fallback.resumeProducer).not.toHaveBeenCalled() + expect(fallback.setPtyBackgrounded).not.toHaveBeenCalled() + }) + + it('tombstones a previously daemon-backed id after its daemon exits', async () => { const current = createDaemonAdapter('daemon', ['daemon-session']) const fallback = createProvider('fallback') const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback }) await provider.discoverDaemonSessions() current.emitExit('daemon-session', 0) - await provider.spawn({ sessionId: 'daemon-session', cols: 80, rows: 24 }) + await expect( + provider.spawn({ sessionId: 'daemon-session', cols: 80, rows: 24 }) + ).rejects.toThrow('daemon_session_routing_unavailable') - expect(fallback.spawn).toHaveBeenCalledWith({ - sessionId: 'daemon-session', - cols: 80, - rows: 24 - }) + expect(fallback.spawn).not.toHaveBeenCalled() }) it('caches a provider discovered by hasPty before routing later operations', () => { @@ -254,6 +310,23 @@ describe('DegradedDaemonPtyProvider', () => { expect(current.write).toHaveBeenCalledWith('daemon-session', 'kept-on-daemon\n') expect(fallback.write).not.toHaveBeenCalled() + expect(current.hasPty).toHaveBeenCalledExactlyOnceWith('daemon-session') + }) + + it('probes each daemon once before memoizing a unique owner', () => { + const current = createDaemonAdapter('current') + const legacy = createDaemonAdapter('legacy', ['legacy-session']) + const fallback = createProvider('fallback') + const provider = new DegradedDaemonPtyProvider({ current, legacy: [legacy], fallback }) + + provider.write('legacy-session', 'first\n') + provider.resize('legacy-session', 100, 30) + + expect(current.hasPty).toHaveBeenCalledExactlyOnceWith('legacy-session') + expect(legacy.hasPty).toHaveBeenCalledExactlyOnceWith('legacy-session') + expect(legacy.write).toHaveBeenCalledExactlyOnceWith('legacy-session', 'first\n') + expect(legacy.resize).toHaveBeenCalledExactlyOnceWith('legacy-session', 100, 30) + expect(fallback.hasPty).not.toHaveBeenCalled() }) it('probes daemon owners without borrowing fallback liveness', async () => { @@ -267,6 +340,9 @@ describe('DegradedDaemonPtyProvider', () => { expect(fallback.probePtyLiveness).not.toHaveBeenCalled() vi.mocked(current.probePtyLiveness).mockResolvedValue(true) + await expect(provider.probePtyLiveness('unknown-session')).resolves.toBeNull() + + vi.mocked(legacy.probePtyLiveness).mockResolvedValue(false) await expect(provider.probePtyLiveness('unknown-session')).resolves.toBe(true) }) @@ -294,9 +370,9 @@ describe('DegradedDaemonPtyProvider', () => { }) }) - it('acknowledges a colliding id through the daemon that produced its snapshot', async () => { + it('keeps colliding liveness unknown and acknowledges through the snapshot producer', async () => { const sessionId = 'colliding-session' - const current = createDaemonAdapter('current', [sessionId]) + const current = createDaemonAdapter('current') const legacy = createDaemonAdapter('legacy', [sessionId]) const fallback = createProvider('fallback') legacy.getBufferSnapshot = vi.fn(async () => ({ @@ -314,21 +390,27 @@ describe('DegradedDaemonPtyProvider', () => { await provider.discoverDaemonSessions() await provider.getBufferSnapshot(sessionId) - legacy.emitExit(sessionId, 0) - expect(provider.hasPty(sessionId)).toBe(true) + current.emitData(sessionId, 'colliding current output') + expect(() => provider.hasPty(sessionId)).toThrow('daemon_session_routing_unavailable') provider.ackColdRestore(sessionId) expect(legacy.ackColdRestore).toHaveBeenCalledExactlyOnceWith(sessionId) expect(current.ackColdRestore).not.toHaveBeenCalled() }) - it('forwards replay output from fallback and daemon providers', () => { + it('forwards replay output from owned fallback and daemon providers', async () => { const current = createDaemonAdapter('daemon') const fallback = createProvider('fallback') const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback }) const replaySpy = vi.fn() const unsubscribe = provider.onReplay(replaySpy) + await provider.spawn({ + sessionId: 'fallback-session', + isNewSession: true, + cols: 80, + rows: 24 + }) current.emitReplay('daemon-session', 'daemon replay') fallback.emitReplay('fallback-session', 'fallback replay') unsubscribe() @@ -345,6 +427,154 @@ describe('DegradedDaemonPtyProvider', () => { }) }) + it('withholds replay output when daemon and fallback ids collide', async () => { + const sessionId = 'replay-collision' + const current = createDaemonAdapter('daemon') + const fallback = createProvider('fallback') + const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback }) + const replay = vi.fn() + provider.onReplay(replay) + await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 }) + + current.emitReplay(sessionId, 'daemon replay') + fallback.emitReplay(sessionId, 'fallback replay') + + expect(replay).not.toHaveBeenCalled() + }) + + it('keeps a daemon survivor live when a colliding fallback exits first', async () => { + const sessionId = 'fallback-exits-first' + 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) + provider.onExit(exit) + await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 }) + + current.emitData(sessionId, 'withheld collision') + fallback.emitExit(sessionId, 0) + current.emitData(sessionId, 'daemon survivor') + + expect(exit).not.toHaveBeenCalled() + expect(data).toHaveBeenCalledExactlyOnceWith({ + id: sessionId, + data: 'daemon survivor' + }) + }) + + it('keeps a fallback survivor live when a colliding daemon exits first', async () => { + const sessionId = 'daemon-exits-first' + 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) + provider.onExit(exit) + await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 }) + + current.emitData(sessionId, 'withheld collision') + current.emitExit(sessionId, 0) + fallback.emitData(sessionId, 'fallback survivor') + fallback.emitExit(sessionId, 0) + + expect(data).toHaveBeenCalledExactlyOnceWith({ + id: sessionId, + data: 'fallback survivor' + }) + expect(exit).toHaveBeenCalledExactlyOnceWith({ id: sessionId, code: 0 }) + }) + + it('keeps a fallback collision fenced while another daemon owner survives', async () => { + const sessionId = 'two-daemon-fallback-collision' + const current = createDaemonAdapter('current') + const legacy = createDaemonAdapter('legacy') + const fallback = createProvider('fallback') + const provider = new DegradedDaemonPtyProvider({ current, legacy: [legacy], fallback }) + const data = vi.fn() + provider.onData(data) + await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 }) + + current.emitData(sessionId, 'current collision') + legacy.emitData(sessionId, 'legacy collision') + current.emitExit(sessionId, 0) + fallback.emitData(sessionId, 'still colliding') + + expect(() => provider.write(sessionId, 'blocked')).toThrow('daemon_session_routing_unavailable') + expect(data).not.toHaveBeenCalled() + + legacy.emitExit(sessionId, 0) + fallback.emitData(sessionId, 'fallback survivor') + provider.write(sessionId, 'allowed') + + expect(data).toHaveBeenCalledExactlyOnceWith({ + id: sessionId, + data: 'fallback survivor' + }) + expect(fallback.write).toHaveBeenCalledExactlyOnceWith(sessionId, 'allowed') + }) + + it('records fallback ownership from output emitted before spawn resolves', 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() + provider.onData(data) + vi.mocked(fallback.spawn).mockImplementationOnce(async () => { + fallback.emitData(sessionId, 'fallback first') + return { id: sessionId } + }) + + await provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 }) + current.emitData(sessionId, 'daemon collision') + fallback.emitData(sessionId, 'withheld fallback collision') + + expect(data).toHaveBeenCalledExactlyOnceWith({ + id: sessionId, + data: 'fallback first' + }) + }) + + it('rejects unknown fallback output after unavailable tombstones are evicted', async () => { + const current = createDaemonAdapter('daemon') + const fallback = createProvider('fallback') + const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback }) + + for (let index = 0; index <= 1_000; index += 1) { + const sessionId = `evicted-${index}` + current.emitData(sessionId, 'frame') + current.emitExit(sessionId, 0) + } + + const data = vi.fn() + provider.onData(data) + fallback.emitData('unknown-fallback', 'unsafe frame') + + expect(data).not.toHaveBeenCalled() + expect(() => provider.write('unknown-fallback', 'unsafe input')).toThrow( + 'daemon_session_routing_unavailable' + ) + + vi.mocked(fallback.spawn).mockImplementationOnce(async () => { + fallback.emitData('explicit-fallback', 'safe frame') + return { id: 'explicit-fallback' } + }) + await provider.spawn({ + sessionId: 'explicit-fallback', + isNewSession: true, + cols: 80, + rows: 24 + }) + + expect(data).toHaveBeenCalledExactlyOnceWith({ + id: 'explicit-fallback', + data: 'safe frame' + }) + }) + it('preserves explicit sequence accounting on daemon data events', () => { const current = createDaemonAdapter('daemon') const fallback = createProvider('fallback') @@ -396,8 +626,13 @@ describe('DegradedDaemonPtyProvider', () => { const current = createDaemonAdapter('daemon') const fallback = createProvider('fallback') const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback }) - const stuck = await provider.spawn({ sessionId: 'stuck', cols: 80, rows: 24 }) - await provider.spawn({ sessionId: 'ok', cols: 80, rows: 24 }) + const stuck = await provider.spawn({ + sessionId: 'stuck', + isNewSession: true, + cols: 80, + rows: 24 + }) + await provider.spawn({ sessionId: 'ok', isNewSession: true, cols: 80, rows: 24 }) const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) vi.mocked(fallback.shutdown).mockImplementation(async (id: string) => { if (id === stuck.id) { @@ -434,6 +669,29 @@ describe('DegradedDaemonPtyProvider', () => { expect(provider.hasPty('legacy-session')).toBe(true) }) + it('reconciles live owners before same-id tombstones regardless of adapter order', async () => { + const sessionId = 'reconciled-owner' + const current = createDaemonAdapter('current') + const legacy = createDaemonAdapter('legacy', [sessionId]) + const fallback = createProvider('fallback') + vi.mocked(current.reconcileOnStartup).mockResolvedValue({ + alive: [], + killed: [sessionId] + }) + vi.mocked(legacy.reconcileOnStartup).mockResolvedValue({ + alive: [sessionId], + killed: [] + }) + const provider = new DegradedDaemonPtyProvider({ current, legacy: [legacy], fallback }) + + await provider.reconcileOnStartup(new Set()) + await provider.sendSignal(sessionId, 'SIGTERM') + + expect(legacy.sendSignal).toHaveBeenCalledExactlyOnceWith(sessionId, 'SIGTERM') + expect(current.sendSignal).not.toHaveBeenCalled() + expect(fallback.sendSignal).not.toHaveBeenCalled() + }) + it('keeps an exited legacy daemon poisoning listProcesses after construction', async () => { const current = createDaemonAdapter('daemon', ['current-session']) const legacy = createDaemonAdapter('legacy', ['legacy-session']) diff --git a/src/main/daemon/degraded-daemon-pty-provider.ts b/src/main/daemon/degraded-daemon-pty-provider.ts index 7b1add71043..93a87d98212 100644 --- a/src/main/daemon/degraded-daemon-pty-provider.ts +++ b/src/main/daemon/degraded-daemon-pty-provider.ts @@ -1,6 +1,5 @@ import type { DaemonPtyAdapter } from './daemon-pty-adapter' import { combineUnsubscribes } from './combine-unsubscribes' -import { shutdownDegradedFallbackSessions } from './degraded-daemon-fallback-shutdown' import { inspectPtyProviderProcess } from '../providers/pty-process-inspection' import type { IPtyProvider, @@ -11,9 +10,8 @@ import type { PtySpawnOptions, PtySpawnResult } from '../providers/types' -import { findDaemonAdapter, listProviderSessionIds } from './degraded-daemon-session-routing' -import { probePtyOwners } from './daemon-pty-liveness-probe' import { DaemonSnapshotAcknowledgementRoutes } from './daemon-snapshot-acknowledgement-routes' +import { DegradedDaemonSessionRouting } from './degraded-daemon-session-routing' export class DegradedDaemonPtyProvider implements IPtyProvider { readonly routesFreshSpawnsToLocalProvider = true @@ -23,7 +21,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { private current: DaemonPtyAdapter private legacy: DaemonPtyAdapter[] private fallback: IPtyProvider - private sessionProviders = new Map() + private readonly routing: DegradedDaemonSessionRouting private readonly snapshotAcks = new DaemonSnapshotAcknowledgementRoutes() private unsubscribers: (() => void)[] = [] private dataListeners: ((payload: PtyDataEvent) => void)[] = [] @@ -37,16 +35,27 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { this.current = opts.current this.legacy = opts.legacy this.fallback = opts.fallback + this.routing = new DegradedDaemonSessionRouting( + this.current, + this.legacy, + this.fallback, + this.snapshotAcks + ) for (const provider of this.allProviders()) { this.unsubscribers.push( provider.onData((payload) => { + if (!this.routing.shouldForwardStreamEvent(provider, payload.id)) { + return + } for (const listener of this.dataListeners) { listener(payload) } }), provider.onExit((payload) => { - this.sessionProviders.delete(payload.id) + if (!this.routing.recordExit(provider, payload.id)) { + return + } for (const listener of this.exitListeners) { listener(payload) } @@ -56,124 +65,114 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { } async discoverDaemonSessions(): Promise { - for (const adapter of this.allDaemonAdapters()) { - try { - const sessions = await adapter.listProcesses() - for (const session of sessions) { - this.sessionProviders.set(session.id, adapter) - } - } catch (error) { - console.warn('[daemon] Failed to discover degraded daemon sessions', error) - } - } + await this.routing.discover() } async spawn(opts: PtySpawnOptions): Promise { - const mapped = opts.sessionId ? this.sessionProviders.get(opts.sessionId) : undefined - const target = mapped ?? this.fallback - const result = await target.spawn(opts) - this.sessionProviders.set(result.id, target) - return result + return await this.routing.spawn(opts) } async attach(id: string): Promise { - await this.providerFor(id).attach(id) + await this.routing.providerFor(id).attach(id) } hasPty(id: string): boolean { - const mapped = this.sessionProviders.get(id) - return mapped ? (mapped.hasPty?.(id) ?? true) : this.findProviderForExistingSession(id) !== null + return this.routing.hasPty(id) } async probePtyLiveness(id: string): Promise { - return await probePtyOwners(id, this.sessionProviders.get(id), this.allDaemonAdapters()) + return await this.routing.probePtyLiveness(id) } // Why: an unknown id cannot borrow listing authority from the fresh-spawn provider. providesAgentSessionOwnerListings = (ptyId: string): boolean => - ( - this.sessionProviders.get(ptyId) ?? this.findProviderForExistingSession(ptyId) - )?.providesAgentSessionOwnerListings?.(ptyId) === true + this.routing.ownerForHint(ptyId)?.providesAgentSessionOwnerListings?.(ptyId) === true write(id: string, data: string): void { - this.providerFor(id).write(id, data) + this.routing.providerFor(id).write(id, data) } resize(id: string, cols: number, rows: number): void { - this.providerFor(id).resize(id, cols, rows) + this.routing.providerFor(id).resize(id, cols, rows) } pauseProducer(id: string): void { - this.providerFor(id).pauseProducer?.(id) + this.routing.ownerForHint(id)?.pauseProducer?.(id) } resumeProducer(id: string): void { - this.providerFor(id).resumeProducer?.(id) + this.routing.ownerForHint(id)?.resumeProducer?.(id) } setPtyBackgrounded(id: string, background: boolean): void { - this.providerFor(id).setPtyBackgrounded?.(id, background) + this.routing.ownerForHint(id)?.setPtyBackgrounded?.(id, background) } async shutdown( id: string, opts: { immediate?: boolean; keepHistory?: boolean; deadlineMs?: number } ): Promise { - await this.providerFor(id).shutdown(id, opts) + const provider = this.routing.providerFor(id) + await provider.shutdown(id, opts) if (!opts.keepHistory) { - this.sessionProviders.delete(id) + this.routing.recordShutdown(id, provider) } } async sendSignal(id: string, signal: string): Promise { - await this.providerFor(id).sendSignal(id, signal) + await this.routing.providerFor(id).sendSignal(id, signal) } async getCwd(id: string): Promise { - return this.providerFor(id).getCwd(id) + return this.routing.providerFor(id).getCwd(id) } async getInitialCwd(id: string): Promise { - return this.providerFor(id).getInitialCwd(id) + return this.routing.providerFor(id).getInitialCwd(id) } async getAppliedSize(id: string): Promise<{ cols: number; rows: number } | null> { - return (await this.providerFor(id).getAppliedSize?.(id)) ?? null + return (await this.routing.providerFor(id).getAppliedSize?.(id)) ?? null } async getBufferSnapshot( id: string, opts?: { scrollbackRows?: number } ): Promise { - return await this.snapshotAcks.capture(id, opts, this.providerFor(id), this.allDaemonAdapters()) + return await this.snapshotAcks.capture( + id, + opts, + this.routing.providerFor(id), + this.allDaemonAdapters() + ) } async clearBuffer(id: string): Promise { - await this.providerFor(id).clearBuffer(id) + await this.routing.providerFor(id).clearBuffer(id) } async closeStartupQueryAuthority(id: string): Promise { - return (await this.providerFor(id).closeStartupQueryAuthority?.(id)) ?? 0 + return (await this.routing.providerFor(id).closeStartupQueryAuthority?.(id)) ?? 0 } acknowledgeDataEvent(id: string, charCount: number): void { - this.providerFor(id).acknowledgeDataEvent(id, charCount) + this.routing.providerFor(id).acknowledgeDataEvent(id, charCount) } async hasChildProcesses(id: string): Promise { - return this.providerFor(id).hasChildProcesses(id) + return this.routing.providerFor(id).hasChildProcesses(id) } async getForegroundProcess(id: string): Promise { - return this.providerFor(id).getForegroundProcess(id) + return this.routing.providerFor(id).getForegroundProcess(id) } inspectProcess(id: string) { return this.hasPty(id) - ? inspectPtyProviderProcess(this.providerFor(id), id) + ? inspectPtyProviderProcess(this.routing.providerFor(id), id) : Promise.reject(new Error('terminal_gone')) } async confirmForegroundProcess(id: string): Promise { - return this.providerFor(id).confirmForegroundProcess?.(id) ?? null + return this.routing.providerFor(id).confirmForegroundProcess?.(id) ?? null } async serialize(ids: string[]): Promise { @@ -186,7 +185,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { async listProcesses(opts?: { deadlineMs?: number }): Promise { const results = await Promise.all( - this.allProviders().map((provider) => provider.listProcesses(opts)) + this.routing.allProviders().map((provider) => provider.listProcesses(opts)) ) return results.flat() } @@ -211,7 +210,15 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { onBackgroundStreamEvent(callback: (payload: PtyBackgroundStreamEvent) => void): () => void { return combineUnsubscribes( - this.allProviders().flatMap((provider) => provider.onBackgroundStreamEvent?.(callback) ?? []) + this.allProviders().flatMap((provider) => { + return ( + provider.onBackgroundStreamEvent?.((payload) => { + if (this.routing.shouldForwardStreamEvent(provider, payload.id)) { + callback(payload) + } + }) ?? [] + ) + }) ) } @@ -225,7 +232,13 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { } onReplay(callback: (payload: { id: string; data: string }) => void): () => void { - const unsubscribes = this.allProviders().map((provider) => provider.onReplay(callback)) + const unsubscribes = this.allProviders().map((provider) => + provider.onReplay((payload) => { + if (this.routing.shouldForwardStreamEvent(provider, payload.id)) { + callback(payload) + } + }) + ) let active = true const trackedUnsubscribe = (): void => { if (!active) { @@ -257,29 +270,15 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { } clearTombstone(sessionId: string): void { - findDaemonAdapter(this.sessionProviders, this.allDaemonAdapters(), sessionId)?.clearTombstone( - sessionId - ) + this.routing.clearTombstone(sessionId) } async reconcileOnStartup(validWorktreeIds: Set): Promise<{ alive: string[] killed: string[] }> { - const alive: string[] = [] - const killed: string[] = [] - for (const adapter of this.allDaemonAdapters()) { - const result = await adapter.reconcileOnStartup(validWorktreeIds) - for (const id of result.alive) { - alive.push(id) - this.sessionProviders.set(id, adapter) - } - for (const id of result.killed) { - killed.push(id) - this.sessionProviders.delete(id) - } - } - return { alive, killed } + const result = await this.routing.reconcile(validWorktreeIds) + return result } dispose(): void { @@ -291,19 +290,19 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { disposeProviderOnly(): void { combineUnsubscribes(this.unsubscribers.splice(0))() + this.routing.dispose() } async shutdownFallbackSessions(): Promise { - return shutdownDegradedFallbackSessions(this.sessionProviders, this.fallback) + return await this.routing.shutdownFallbackSessions() } getCurrentDaemonSessionIds(): string[] { - return listProviderSessionIds(this.sessionProviders, this.current) + return this.routing.currentSessionIds() } fanoutCurrentDaemonSyntheticExits(code: number): void { - for (const id of this.getCurrentDaemonSessionIds()) { - this.sessionProviders.delete(id) + for (const id of this.routing.recordCurrentSyntheticExits()) { // Why: restart kills listed sessions even when the adapter did not track them active. // oxlint-disable-next-line unicorn/no-useless-spread -- copy-safe: listeners may unsubscribe during iteration for (const listener of [...this.exitListeners]) { @@ -329,29 +328,11 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { return this.allDaemonAdapters() } - private providerFor(sessionId: string): IPtyProvider { - return ( - this.sessionProviders.get(sessionId) ?? - this.findProviderForExistingSession(sessionId) ?? - this.fallback - ) - } - - private findProviderForExistingSession(sessionId: string): IPtyProvider | null { - for (const provider of this.allProviders()) { - if (provider.hasPty?.(sessionId) === true) { - this.sessionProviders.set(sessionId, provider) - return provider - } - } - return null - } - private allProviders(): IPtyProvider[] { - return [this.fallback, ...this.allDaemonAdapters()] + return this.routing.allProviders() } private allDaemonAdapters(): DaemonPtyAdapter[] { - return [this.current, ...this.legacy] + return [...this.routing.allAdapters()] } } diff --git a/src/main/daemon/degraded-daemon-session-routing.ts b/src/main/daemon/degraded-daemon-session-routing.ts index 2ff91cf128d..09c718eafdc 100644 --- a/src/main/daemon/degraded-daemon-session-routing.ts +++ b/src/main/daemon/degraded-daemon-session-routing.ts @@ -1,22 +1,312 @@ -import type { IPtyProvider } from '../providers/types' +import type { IPtyProvider, PtySpawnOptions, PtySpawnResult } from '../providers/types' import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import { shutdownDegradedFallbackSessions } from './degraded-daemon-fallback-shutdown' +import { DaemonSessionRouteTable } from './daemon-session-route-table' +import { DaemonSessionOwnerUnknownError, sameDaemonIncarnation } from './daemon-session-route' +import type { DaemonSnapshotAcknowledgementRoutes } from './daemon-snapshot-acknowledgement-routes' +import { DegradedFallbackSpawnRoutes } from './degraded-fallback-spawn-routes' -export function listProviderSessionIds( - sessionProviders: ReadonlyMap, - provider: IPtyProvider -): string[] { - return [...sessionProviders] - .filter(([, mappedProvider]) => mappedProvider === provider) - .map(([id]) => id) -} +export class DegradedDaemonSessionRouting { + private readonly adapters: DaemonPtyAdapter[] + private readonly adapterSet: ReadonlySet + private readonly routes: DaemonSessionRouteTable + private readonly fallbackSessions = new Map() + private readonly fallbackCollisions = new Set() + private readonly fallbackSpawns = new DegradedFallbackSpawnRoutes() + private readonly identityUnsubscribes: (() => void)[] = [] -export function findDaemonAdapter( - sessionProviders: ReadonlyMap, - daemonAdapters: readonly DaemonPtyAdapter[], - sessionId: string -): DaemonPtyAdapter | null { - const provider = sessionProviders.get(sessionId) - return provider && daemonAdapters.includes(provider as DaemonPtyAdapter) - ? (provider as DaemonPtyAdapter) - : null + constructor( + private readonly current: DaemonPtyAdapter, + legacy: readonly DaemonPtyAdapter[], + private readonly fallback: IPtyProvider, + private readonly snapshotAcks: DaemonSnapshotAcknowledgementRoutes + ) { + this.adapters = [current, ...legacy] + this.adapterSet = new Set(this.adapters) + this.routes = new DaemonSessionRouteTable(this.adapters) + for (const adapter of this.adapters) { + this.identityUnsubscribes.push( + adapter.onDaemonIdentityChanged(({ previous }) => { + this.snapshotAcks.dropAdapterIncarnation(adapter, previous) + this.routes.markOwnerUnavailable(adapter, previous) + }) + ) + } + } + + async discover(): Promise { + for (const adapter of this.adapters) { + try { + const before = adapter.getLastAuthenticatedDaemonIdentity() + const sessions = await adapter.listProcesses() + const incarnation = adapter.getLastAuthenticatedDaemonIdentity() + if (before && !sameDaemonIncarnation(before, incarnation)) { + throw new Error('daemon incarnation changed during session discovery') + } + this.routes.recordCompleteDiscovery( + adapter, + sessions.map((session) => session.id), + incarnation + ) + for (const session of sessions) { + if (this.fallbackSessions.has(session.id)) { + this.fallbackCollisions.add(session.id) + } + } + } catch (error) { + this.routes.recordDiscoveryFailure(adapter) + console.warn('[daemon] Failed to discover degraded daemon sessions', error) + } + } + } + + async spawn(opts: PtySpawnOptions): Promise { + const target = await this.spawnTarget(opts) + const inFlightFallbackId = target === this.fallback ? opts.sessionId : undefined + if (inFlightFallbackId) { + this.fallbackSpawns.begin(inFlightFallbackId) + } + try { + const result = await target.spawn(opts) + if (target === this.fallback) { + if (!this.fallbackSpawns.hasExited(result.id)) { + this.fallbackSessions.set(result.id, target) + } + if (this.routes.get(result.id)) { + this.fallbackCollisions.add(result.id) + } + } else { + this.routes.recordOwned(result.id, target as DaemonPtyAdapter) + } + return result + } finally { + if (inFlightFallbackId) { + this.fallbackSpawns.end(inFlightFallbackId) + } + } + } + + providerFor(sessionId: string): IPtyProvider { + if (this.fallbackCollisions.has(sessionId)) { + throw new DaemonSessionOwnerUnknownError(sessionId) + } + return this.fallbackSessions.get(sessionId) ?? this.routes.resolveOwnerSync(sessionId) + } + + ownerForHint(sessionId: string): IPtyProvider | null { + try { + return this.providerFor(sessionId) + } catch { + return null + } + } + + hasPty(sessionId: string): boolean { + if (this.fallbackCollisions.has(sessionId)) { + return true + } + const fallback = this.fallbackSessions.get(sessionId) + return fallback ? (fallback.hasPty?.(sessionId) ?? true) : this.routes.hasPty(sessionId) + } + + async probePtyLiveness(sessionId: string): Promise { + if (this.fallbackCollisions.has(sessionId)) { + return null + } + const fallback = this.fallbackSessions.get(sessionId) + if (!fallback) { + return await this.routes.probeLiveness(sessionId) + } + return fallback.probePtyLiveness + ? await fallback.probePtyLiveness(sessionId) + : (fallback.hasPty?.(sessionId) ?? 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) + return false + } + return shouldForward + } + const route = this.routes.get(sessionId) + if (!route) { + return ( + !this.fallbackSpawns.hasExited(sessionId) && + (this.fallbackSessions.get(sessionId) === provider || + this.fallbackSpawns.isInFlight(sessionId)) + ) + } + if ( + this.fallbackSessions.get(sessionId) === provider && + route.state === 'unavailable' && + !this.fallbackCollisions.has(sessionId) + ) { + this.fallbackCollisions.delete(sessionId) + return true + } + if (route) { + this.fallbackCollisions.add(sessionId) + return false + } + return true + } + + recordExit(provider: IPtyProvider, sessionId: string): boolean { + const adapter = this.daemonAdapter(provider) + if (!adapter) { + const isOwned = + this.fallbackSessions.get(sessionId) === provider || + this.fallbackSpawns.isInFlight(sessionId) + if (!isOwned) { + return false + } + this.fallbackSpawns.recordExit(sessionId) + this.fallbackSessions.delete(sessionId) + this.fallbackCollisions.delete(sessionId) + const route = this.routes.get(sessionId) + return !route || route.state === 'unavailable' + } + this.snapshotAcks.dropForProducer(sessionId, adapter) + if (this.fallbackSessions.has(sessionId)) { + this.routes.markUnavailable(sessionId, adapter) + if (this.routes.get(sessionId)?.state === 'unavailable') { + this.fallbackCollisions.delete(sessionId) + } + return false + } + const shouldForward = this.routes.shouldForwardEvent(sessionId, adapter) + this.routes.markUnavailable(sessionId, adapter) + return shouldForward + } + + recordShutdown(sessionId: string, provider: IPtyProvider, keepHistory?: boolean): void { + if (keepHistory) { + return + } + if (provider === this.fallback) { + this.fallbackSessions.delete(sessionId) + this.fallbackCollisions.delete(sessionId) + } else { + const adapter = provider as DaemonPtyAdapter + this.routes.markUnavailable(sessionId, adapter) + this.snapshotAcks.dropForProducer(sessionId, adapter) + } + } + + async reconcile(validWorktreeIds: Set): Promise<{ alive: string[]; killed: string[] }> { + const alive: string[] = [] + const killed: string[] = [] + const reconciled: { + adapter: DaemonPtyAdapter + incarnation: ReturnType + alive: string[] + killed: string[] + }[] = [] + for (const adapter of this.adapters) { + const before = adapter.getLastAuthenticatedDaemonIdentity() + const result = await adapter.reconcileOnStartup(validWorktreeIds) + const incarnation = adapter.getLastAuthenticatedDaemonIdentity() + if (before && !sameDaemonIncarnation(before, incarnation)) { + this.routes.markOwnerUnavailable(adapter, before) + throw new Error('daemon incarnation changed during startup reconciliation') + } + reconciled.push({ adapter, incarnation, ...result }) + for (const id of result.alive) { + alive.push(id) + } + for (const id of result.killed) { + killed.push(id) + } + } + for (const { adapter, incarnation } of reconciled) { + if (!sameDaemonIncarnation(incarnation, adapter.getLastAuthenticatedDaemonIdentity())) { + this.routes.markOwnerUnavailable(adapter, incarnation) + throw new Error('daemon incarnation changed during startup reconciliation') + } + } + for (const { adapter, incarnation, alive: sessionIds } of reconciled) { + for (const id of sessionIds) { + this.routes.recordOwnedIncarnation(id, adapter, incarnation) + if (this.fallbackSessions.has(id)) { + this.fallbackCollisions.add(id) + } + } + } + for (const { adapter, incarnation, killed: sessionIds } of reconciled) { + for (const id of sessionIds) { + this.routes.markUnavailable(id, adapter, incarnation) + this.snapshotAcks.dropForProducer(id, adapter, incarnation) + if (this.routes.get(id)?.state === 'unavailable') { + this.fallbackCollisions.delete(id) + } + } + } + return { alive, killed } + } + + clearTombstone(sessionId: string): void { + this.routes.getOwned(sessionId)?.clearTombstone(sessionId) + } + + currentSessionIds(): string[] { + return this.routes.getOwnedSessionIds(this.current) + } + + recordCurrentSyntheticExits(): string[] { + const forwarded: string[] = [] + for (const id of this.currentSessionIds()) { + if (this.routes.shouldForwardEvent(id, this.current)) { + forwarded.push(id) + } + this.routes.markUnavailable(id, this.current) + this.snapshotAcks.dropForProducer(id, this.current) + if (this.routes.get(id)?.state === 'unavailable') { + this.fallbackCollisions.delete(id) + } + } + return forwarded + } + + async shutdownFallbackSessions(): Promise { + return await shutdownDegradedFallbackSessions(this.fallbackSessions, this.fallback) + } + + allProviders(): IPtyProvider[] { + return [this.fallback, ...this.adapters] + } + + allAdapters(): readonly DaemonPtyAdapter[] { + return this.adapters + } + + dispose(): void { + for (const unsubscribe of this.identityUnsubscribes.splice(0)) { + unsubscribe() + } + this.fallbackSessions.clear() + this.fallbackCollisions.clear() + this.fallbackSpawns.clear() + this.snapshotAcks.clear() + } + + private async spawnTarget(opts: PtySpawnOptions): Promise { + if (opts.sessionId && this.fallbackCollisions.has(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)) + ) + } + + private daemonAdapter(provider: IPtyProvider): DaemonPtyAdapter | null { + return this.adapterSet.has(provider as DaemonPtyAdapter) ? (provider as DaemonPtyAdapter) : null + } } diff --git a/src/main/daemon/degraded-fallback-spawn-routes.ts b/src/main/daemon/degraded-fallback-spawn-routes.ts new file mode 100644 index 00000000000..6977a8f0dcb --- /dev/null +++ b/src/main/daemon/degraded-fallback-spawn-routes.ts @@ -0,0 +1,41 @@ +export class DegradedFallbackSpawnRoutes { + private readonly counts = new Map() + private readonly exited = new Set() + + begin(sessionId: string): void { + const count = this.counts.get(sessionId) ?? 0 + if (count === 0) { + this.exited.delete(sessionId) + } + this.counts.set(sessionId, count + 1) + } + + end(sessionId: string): void { + const remaining = (this.counts.get(sessionId) ?? 1) - 1 + if (remaining > 0) { + this.counts.set(sessionId, remaining) + return + } + this.counts.delete(sessionId) + this.exited.delete(sessionId) + } + + isInFlight(sessionId: string): boolean { + return this.counts.has(sessionId) + } + + hasExited(sessionId: string): boolean { + return this.exited.has(sessionId) + } + + recordExit(sessionId: string): void { + if (this.isInFlight(sessionId)) { + this.exited.add(sessionId) + } + } + + clear(): void { + this.counts.clear() + this.exited.clear() + } +} diff --git a/src/main/ipc/pty-management.test.ts b/src/main/ipc/pty-management.test.ts index ca5678ef462..4a63a08b211 100644 --- a/src/main/ipc/pty-management.test.ts +++ b/src/main/ipc/pty-management.test.ts @@ -383,6 +383,34 @@ describe('pty:management IPC handlers', () => { }) }) + it('does not count a replacement incarnation of the same id as remaining', async () => { + const sessionId = 'reused-id' + const current = makeAdapter(5, []) + vi.mocked(current.listSessions) + .mockResolvedValueOnce([ + makeSession(sessionId, { + incarnationId: 'original' + }) as Omit + ]) + .mockResolvedValue([ + makeSession(sessionId, { + incarnationId: 'replacement' + }) as Omit + ]) + const { registerDaemonManagementHandlers } = await importFresh() + getDaemonProviderMock.mockReturnValue(await makeRouter(current)) + registerDaemonManagementHandlers() + + const result = await runKillAllWithPolls(buildHandlerMap()['pty:management:killAll']) + + expect(result).toEqual({ + killedCount: 1, + remainingCount: 0, + killedSessionIds: [sessionId] + }) + expect(current.shutdown).toHaveBeenCalledExactlyOnceWith(sessionId, { immediate: true }) + }) + it('swallows per-session shutdown rejections without stopping the batch', async () => { const sessionsList = [makeSession('a'), makeSession('b')] const current = makeAdapter(5, []) diff --git a/src/main/ipc/pty-management.ts b/src/main/ipc/pty-management.ts index f53ff13646f..8f784fad459 100644 --- a/src/main/ipc/pty-management.ts +++ b/src/main/ipc/pty-management.ts @@ -61,13 +61,45 @@ async function collectSessions(adapters: DaemonPtyAdapter[]): Promise +} + +function indexSessionIncarnations( + sessions: readonly OwnedDaemonSession[] +): Map> { + const byOwner = new Map>() + for (const { owner, session } of sessions) { + let bySessionId = byOwner.get(owner) + if (!bySessionId) { + bySessionId = new Map() + byOwner.set(owner, bySessionId) + } + let bucket = bySessionId.get(session.sessionId) + if (!bucket) { + bucket = { hasUnqualifiedSession: false, incarnationIds: new Set() } + bySessionId.set(session.sessionId, bucket) + } + if (session.incarnationId === undefined) { + bucket.hasUnqualifiedSession = true + } else { + bucket.incarnationIds.add(session.incarnationId) + } + } + return byOwner +} + +function hasSessionIncarnation( + index: ReadonlyMap>, + original: OwnedDaemonSession +): boolean { + const bucket = index.get(original.owner)?.get(original.session.sessionId) return ( - left.owner === right.owner && - left.session.sessionId === right.session.sessionId && - (left.session.incarnationId === undefined || - right.session.incarnationId === undefined || - left.session.incarnationId === right.session.incarnationId) + bucket !== undefined && + (original.session.incarnationId === undefined || + bucket.hasUnqualifiedSession || + bucket.incarnationIds.has(original.session.incarnationId)) ) } @@ -120,20 +152,19 @@ export function registerDaemonManagementHandlers(): void { for (let attempt = 0; attempt < MAX_POLL_ATTEMPTS; attempt += 1) { await sleep(POLL_INTERVAL_MS) const current = await collectSessions(adapters) + const currentIndex = indexSessionIncarnations(current.sessions) remainingOriginal = initial.filter( (original) => current.unavailableOwners.has(original.owner) || - current.sessions.some((candidate) => sameSessionIncarnation(original, candidate)) + hasSessionIncarnation(currentIndex, original) ) if (remainingOriginal.length === 0) { break } } - const killed = initial.filter( - (original) => - !remainingOriginal.some((candidate) => sameSessionIncarnation(original, candidate)) - ) + const remainingSet = new Set(remainingOriginal) + const killed = initial.filter((original) => !remainingSet.has(original)) return { killedCount: killed.length, remainingCount: remainingOriginal.length,