fix(daemon): preserve fallback and pidfile fences

This commit is contained in:
Brennan Benson
2026-07-30 23:49:24 -07:00
parent f26c07b2eb
commit 126cf2bfcf
7 changed files with 275 additions and 82 deletions
+5 -7
View File
@@ -410,15 +410,13 @@ describe('killStaleDaemon pid identity guards', () => {
rmSync(dir, { recursive: true, force: true })
})
it('does not SIGTERM when the saved startedAtMs mismatches the current process', async () => {
it('removes a stale pidfile without signaling a mismatched process', async () => {
if (process.platform === 'win32') {
return
}
// Why: seed a pid file that claims the daemon is `process.pid` (us) but
// was started 1 hour ago. Our real start time is "now," so startTimeMatches
// returns false and isDaemonProcess rejects. killStaleDaemon must not call
// process.kill in that case.
// was started 1 hour ago. The identity mismatch must not signal us.
const bogusStartedAtMs = Date.now() - 60 * 60 * 1000
writeFileSync(
getDaemonPidPath(dir),
@@ -437,7 +435,7 @@ describe('killStaleDaemon pid identity guards', () => {
([, sig]) => sig === 'SIGTERM' || sig === 'SIGKILL'
)
expect(terminationSignals).toEqual([])
expect(parseDaemonPidFile(readFileSync(pidPath, 'utf8'))?.pid).toBe(process.pid)
expect(existsSync(pidPath)).toBe(false)
} finally {
killSpy.mockRestore()
}
@@ -587,7 +585,7 @@ describe('killStaleDaemon pid identity guards', () => {
}
})
it('preserves replacement identity files when the pid changes ownership during escalation', async () => {
it('removes the unchanged pidfile when the pid changes ownership during escalation', async () => {
if (process.platform === 'win32') {
return
}
@@ -639,7 +637,7 @@ describe('killStaleDaemon pid identity guards', () => {
await replacementObserved
await vi.advanceTimersByTimeAsync(3_100)
await expect(killing).resolves.toBe(true)
expect(parseDaemonPidFile(readFileSync(pidPath, 'utf8'))?.pid).toBe(pid)
expect(existsSync(pidPath)).toBe(false)
expect(existsSync(socketPath)).toBe(true)
} finally {
vi.useRealTimers()
+73 -59
View File
@@ -10,7 +10,7 @@ import {
} from '../../shared/process-output-field-scanner'
import { isStartupDiagnosticsEnabled, logStartupDiagnostic } from '../startup/startup-diagnostics'
import { encodeNdjson } from './ndjson'
import { getDaemonPidPath, unlinkOwnedDaemonPidFile } from './daemon-spawner'
import { getDaemonPidPath, unlinkUnchangedDaemonPidFile } from './daemon-spawner'
import {
PROTOCOL_VERSION,
type HelloMessage,
@@ -519,51 +519,68 @@ async function queryWindowsProcessIdentity(pid: number): Promise<WindowsProcessI
}
}
type DaemonProcessMatch = 'match' | 'stale' | 'unknown'
function probeProcessExistence(pid: number): 'present' | 'absent' | 'unknown' {
try {
process.kill(pid, 0)
return 'present'
} catch (error) {
return isNoSuchProcessError(error) ? 'absent' : 'unknown'
}
}
async function matchDaemonProcess(
pid: number,
socketPath: string,
tokenPath: string,
startedAtMs: number | null
): Promise<DaemonProcessMatch> {
const processExistence = probeProcessExistence(pid)
if (processExistence !== 'present') {
return processExistence === 'absent' ? 'stale' : 'unknown'
}
if (process.platform === 'win32') {
const identity = await queryWindowsProcessIdentity(pid)
if (identity === null) {
return probeProcessExistence(pid) === 'absent' ? 'stale' : 'unknown'
}
// Why: image names are too broad after PID reuse. Match the daemon entry
// plus the exact socket/token args so we only kill the daemon for this
// userData protocol endpoint.
return commandLineMatchesDaemon(identity.commandLine, socketPath, tokenPath) &&
startTimesWithinTolerance(identity.startedAtMs, startedAtMs, WIN32_START_TIME_TOLERANCE_MS)
? 'match'
: 'stale'
}
let commandLine: string
try {
commandLine = readFileSync(`/proc/${pid}/cmdline`, 'utf8')
} catch {
try {
commandLine = execFileSync('ps', ['-p', String(pid), '-o', 'command='], {
encoding: 'utf8',
timeout: 2_000
})
} catch {
return probeProcessExistence(pid) === 'absent' ? 'stale' : 'unknown'
}
}
return commandLineMatchesDaemon(commandLine, socketPath, tokenPath) &&
startTimeMatches(pid, startedAtMs)
? 'match'
: 'stale'
}
async function isDaemonProcess(
pid: number,
socketPath: string,
tokenPath: string,
startedAtMs: number | null
): Promise<boolean> {
try {
process.kill(pid, 0)
} catch {
return false
}
if (process.platform === 'win32') {
const identity = await queryWindowsProcessIdentity(pid)
if (identity === null) {
return false
}
// Why: image names are too broad after PID reuse. Match the daemon entry
// plus the exact socket/token args so we only kill the daemon for this
// userData protocol endpoint.
return (
commandLineMatchesDaemon(identity.commandLine, socketPath, tokenPath) &&
startTimesWithinTolerance(identity.startedAtMs, startedAtMs, WIN32_START_TIME_TOLERANCE_MS)
)
}
try {
const cmdline = readFileSync(`/proc/${pid}/cmdline`, 'utf8')
return (
commandLineMatchesDaemon(cmdline, socketPath, tokenPath) && startTimeMatches(pid, startedAtMs)
)
} catch {
try {
const output = execFileSync('ps', ['-p', String(pid), '-o', 'command='], {
encoding: 'utf8',
timeout: 2_000
})
return (
commandLineMatchesDaemon(output, socketPath, tokenPath) &&
startTimeMatches(pid, startedAtMs)
)
} catch {
return false
}
}
return (await matchDaemonProcess(pid, socketPath, tokenPath, startedAtMs)) === 'match'
}
async function getDaemonCommandLine(pid: number): Promise<string | null> {
@@ -670,17 +687,18 @@ 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
let identifiedPidContents: string | null = null
try {
const parsedPid = parseDaemonPidFile(readFileSync(pidPath, 'utf8'))
if (
parsedPid &&
(await isDaemonProcess(parsedPid.pid, socketPath, tokenPath, parsedPid.startedAtMs))
) {
const pidContents = readFileSync(pidPath, 'utf8')
const parsedPid = parseDaemonPidFile(pidContents)
const processMatch = parsedPid
? await matchDaemonProcess(parsedPid.pid, socketPath, tokenPath, parsedPid.startedAtMs)
: 'stale'
if (processMatch === 'stale') {
unlinkUnchangedDaemonPidFile(pidPath, pidContents)
} else if (parsedPid && processMatch === 'match') {
identifiedDaemon = true
identifiedPid = parsedPid.pid
identifiedLaunchNonce = parsedPid.launchNonce
identifiedPidContents = pidContents
const { pid, startedAtMs } = parsedPid
let exited = false
try {
@@ -709,11 +727,12 @@ export async function killStaleDaemon(
// window is long enough for the pid to be recycled if the original
// daemon died during the wait. Without this, we'd SIGKILL an unrelated
// process that happens to now own the same pid.
if (!(await isDaemonProcess(pid, socketPath, tokenPath, startedAtMs))) {
const currentMatch = await matchDaemonProcess(pid, socketPath, tokenPath, startedAtMs)
if (currentMatch !== 'match') {
console.warn('[daemon] Skipping SIGKILL for stale daemon: reason=pid_recycled')
identifiedDaemon = false
identifiedDaemon = currentMatch === 'stale'
exited = true
killedDaemon = true
killedDaemon = currentMatch === 'stale'
} else {
try {
process.kill(pid, 'SIGKILL')
@@ -729,13 +748,8 @@ export async function killStaleDaemon(
// PID file missing or process already dead
}
if (
killedDaemon &&
identifiedDaemon &&
identifiedPid !== null &&
identifiedLaunchNonce !== null
) {
unlinkOwnedDaemonPidFile(pidPath, identifiedPid, identifiedLaunchNonce)
if (killedDaemon && identifiedDaemon && identifiedPidContents !== null) {
unlinkUnchangedDaemonPidFile(pidPath, identifiedPidContents)
}
const socketIsLive = await canConnectSocket(socketPath)
+4
View File
@@ -126,6 +126,10 @@ export function unlinkOwnedDaemonPidFile(
})
}
export function unlinkUnchangedDaemonPidFile(pidPath: string, expectedContents: string): boolean {
return claimAndUnlinkOwnedFile(pidPath, (content) => content === expectedContents)
}
export function unlinkOwnedDaemonTokenFile(tokenPath: string, expectedToken: string): boolean {
return claimAndUnlinkOwnedFile(tokenPath, (content) => content.trim() === expectedToken)
}
@@ -0,0 +1,139 @@
import { spawn } from 'node:child_process'
import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { basename, join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { killStaleDaemon } from './daemon-health'
import {
getDaemonPidPath,
serializeDaemonPidFile,
unlinkUnchangedDaemonPidFile
} from './daemon-spawner'
function socketPathFor(dir: string): string {
return process.platform === 'win32'
? `\\\\.\\pipe\\${basename(dir)}-stale-pid.sock`
: join(dir, 'daemon.sock')
}
describe('stale daemon pidfile cleanup', () => {
let dir: string
let pidPath: string
let socketPath: string
let tokenPath: string
beforeEach(() => {
dir = mkdtempSync(join(tmpdir(), 'daemon-stale-pidfile-'))
pidPath = getDaemonPidPath(dir)
socketPath = socketPathFor(dir)
tokenPath = join(dir, 'daemon.token')
})
afterEach(() => {
vi.restoreAllMocks()
rmSync(dir, { recursive: true, force: true })
})
it('clears a dead process record so an exclusive relaunch can publish', async () => {
const child = spawn(process.execPath, ['-e', ''])
const deadPid = child.pid
expect(deadPid).toEqual(expect.any(Number))
await new Promise<void>((resolve) => child.once('exit', () => resolve()))
writeFileSync(
pidPath,
serializeDaemonPidFile({
pid: deadPid!,
startedAtMs: null
}),
{ mode: 0o600 }
)
await expect(killStaleDaemon(dir, socketPath, tokenPath)).resolves.toBe(false)
expect(existsSync(pidPath)).toBe(false)
const replacement = serializeDaemonPidFile({
pid: process.pid,
startedAtMs: null,
launchNonce: 'replacement'
})
expect(() => writeFileSync(pidPath, replacement, { mode: 0o600, flag: 'wx' })).not.toThrow()
expect(readFileSync(pidPath, 'utf8')).toBe(replacement)
})
it('removes a killed legacy record so an exclusive relaunch can publish', async () => {
const child = spawn(
process.execPath,
[
'-e',
"process.send?.('ready');setInterval(()=>{},1000)",
'daemon-entry',
socketPath,
tokenPath
],
{ stdio: ['ignore', 'ignore', 'ignore', 'ipc'] }
)
await new Promise<void>((resolve) => {
child.on('message', (message) => {
if (message === 'ready') {
resolve()
}
})
})
writeFileSync(
pidPath,
serializeDaemonPidFile({
pid: child.pid!,
startedAtMs: null
}),
{ mode: 0o600 }
)
try {
await expect(killStaleDaemon(dir, socketPath, tokenPath)).resolves.toBe(true)
expect(existsSync(pidPath)).toBe(false)
expect(() =>
writeFileSync(
pidPath,
serializeDaemonPidFile({
pid: process.pid,
startedAtMs: null,
launchNonce: 'replacement'
}),
{ mode: 0o600, flag: 'wx' }
)
).not.toThrow()
} finally {
if (child.exitCode === null) {
child.kill('SIGKILL')
}
}
})
it('preserves the exact record when process access is rejected', async () => {
const record = serializeDaemonPidFile({
pid: process.pid,
startedAtMs: null,
launchNonce: 'inaccessible'
})
writeFileSync(pidPath, record, { mode: 0o600 })
vi.spyOn(process, 'kill').mockImplementation(() => {
throw Object.assign(new Error('operation not permitted'), { code: 'EPERM' })
})
await expect(killStaleDaemon(dir, socketPath, tokenPath)).resolves.toBe(false)
expect(readFileSync(pidPath, 'utf8')).toBe(record)
})
it('preserves a replacement that no longer matches the stale contents', () => {
const stale = serializeDaemonPidFile({ pid: 1, startedAtMs: null })
const replacement = serializeDaemonPidFile({
pid: 2,
startedAtMs: null,
launchNonce: 'replacement'
})
writeFileSync(pidPath, replacement, { mode: 0o600 })
expect(unlinkUnchangedDaemonPidFile(pidPath, stale)).toBe(false)
expect(readFileSync(pidPath, 'utf8')).toBe(replacement)
})
})
@@ -622,7 +622,7 @@ describe('DegradedDaemonPtyProvider', () => {
expect(current.write).not.toHaveBeenCalled()
})
it('releases an in-flight fallback when the colliding daemon changes incarnation', async () => {
it('keeps an in-flight fallback fenced across identity changes until confirmation', async () => {
const sessionId = 'in-flight-incarnation-collision'
const previous = { pid: 1, startedAtMs: 1, launchNonce: 'previous' }
const replacement = { pid: 2, startedAtMs: 2, launchNonce: 'replacement' }
@@ -649,15 +649,45 @@ describe('DegradedDaemonPtyProvider', () => {
current.emitData(sessionId, 'foreign daemon output')
identity = replacement
notifyIdentityChange({ previous, current: replacement })
fallback.emitData(sessionId, 'fallback first')
fallback.emitData(sessionId, 'still fenced')
expect(data).not.toHaveBeenCalled()
expect(() => provider.write(sessionId, 'blocked input')).toThrow(
'daemon_session_routing_unavailable'
)
resolveSpawn({ id: sessionId })
await spawning
fallback.emitData(sessionId, 'fallback after confirmation')
provider.write(sessionId, 'fallback input')
expect(data).toHaveBeenCalledExactlyOnceWith({ id: sessionId, data: 'fallback first' })
expect(data).toHaveBeenCalledExactlyOnceWith({
id: sessionId,
data: 'fallback after confirmation'
})
expect(fallback.write).toHaveBeenCalledExactlyOnceWith(sessionId, 'fallback input')
})
it('drops an unconfirmed collision when fallback spawn fails', async () => {
const sessionId = 'rejected-fallback-collision'
const current = createDaemonAdapter('daemon')
const fallback = createProvider('fallback')
let rejectSpawn!: (error: Error) => void
vi.mocked(fallback.spawn).mockReturnValueOnce(
new Promise((_, reject) => {
rejectSpawn = reject
})
)
const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback })
const spawning = provider.spawn({ sessionId, isNewSession: true, cols: 80, rows: 24 })
current.emitData(sessionId, 'foreign daemon output')
rejectSpawn(new Error('spawn failed'))
await expect(spawning).rejects.toThrow('spawn failed')
expect(provider.hasPty(sessionId)).toBe(false)
})
it('rejects unknown fallback output after unavailable tombstones are evicted', async () => {
const current = createDaemonAdapter('daemon')
const fallback = createProvider('fallback')
@@ -25,7 +25,7 @@ export class DegradedDaemonSessionRouting {
this.adapters = [current, ...legacy]
this.adapterSet = new Set(this.adapters)
this.routes = new DaemonSessionRouteTable(this.adapters)
this.fallbackRoutes = new FallbackRoutes(this.fallbackOwners, this.routes, this.fallbackSpawns)
this.fallbackRoutes = new FallbackRoutes(this.fallbackOwners, this.routes)
for (const adapter of this.adapters) {
this.identityUnsubscribes.push(
adapter.onDaemonIdentityChanged(({ previous }) => {
@@ -72,13 +72,8 @@ export class DegradedDaemonSessionRouting {
try {
const result = await target.spawn(opts)
if (target === this.fallback) {
const fallbackSurvived = !this.fallbackSpawns.hasExited(result.id)
if (fallbackSurvived) {
this.fallbackRoutes.recordSession(result.id, target)
}
const daemonRoute = this.routes.get(result.id)
if (fallbackSurvived && daemonRoute && daemonRoute.state !== 'unavailable') {
this.fallbackRoutes.recordCollision(result.id)
if (!this.fallbackSpawns.hasExited(result.id)) {
this.fallbackRoutes.recordSpawnedSession(result.id, target)
}
} else {
this.routes.recordOwned(result.id, target as DaemonPtyAdapter)
@@ -87,6 +82,9 @@ export class DegradedDaemonSessionRouting {
} finally {
if (inFlightFallbackId) {
this.fallbackSpawns.end(inFlightFallbackId)
if (!this.fallbackSpawns.hasLiveCandidate(inFlightFallbackId)) {
this.fallbackRoutes.clearUnownedCollision(inFlightFallbackId)
}
}
}
}
@@ -2,7 +2,6 @@ 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'
import type { DegradedFallbackSpawnRoutes } from './degraded-fallback-spawn-routes'
type FallbackLivenessProbe = {
provider: IPtyProvider
@@ -16,8 +15,7 @@ export class DegradedFallbackSessionRoutes {
constructor(
private readonly sessions: Map<string, IPtyProvider>,
private readonly daemonRoutes: DaemonSessionRouteTable,
private readonly spawns: DegradedFallbackSpawnRoutes
private readonly daemonRoutes: DaemonSessionRouteTable
) {}
hasCollision(sessionId: string): boolean {
@@ -32,9 +30,21 @@ export class DegradedFallbackSessionRoutes {
this.collisionIds.delete(sessionId)
}
recordSession(sessionId: string, provider: IPtyProvider): void {
recordSpawnedSession(sessionId: string, provider: IPtyProvider): void {
this.sessions.set(sessionId, provider)
this.epochs.set(sessionId, Symbol())
const daemonRoute = this.daemonRoutes.get(sessionId)
if (daemonRoute?.state === 'unavailable') {
this.clearCollision(sessionId)
} else if (daemonRoute) {
this.recordCollision(sessionId)
}
}
clearUnownedCollision(sessionId: string): void {
if (!this.sessions.has(sessionId)) {
this.clearCollision(sessionId)
}
}
deleteSession(sessionId: string): void {
@@ -96,7 +106,7 @@ export class DegradedFallbackSessionRoutes {
owner: DaemonPtyAdapter,
incarnation: ReturnType<DaemonPtyAdapter['getLastAuthenticatedDaemonIdentity']>
): boolean {
if (!this.sessions.has(sessionId) && !this.spawns.hasLiveCandidate(sessionId)) {
if (!this.sessions.has(sessionId)) {
return false
}
const route = this.daemonRoutes.get(sessionId)