fix(daemon): close routing fence review gaps

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