mirror of
https://github.com/stablyai/orca.git
synced 2026-10-21 00:03:46 +00:00
fix(daemon): make legacy retirement fail closed
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
/* eslint-disable max-lines -- Why: covers daemon-init's full restart flow (7-step sequence per docs/daemon-staleness-ux.md §Phase 1 + coalescer); one describe block keeps shared mocks in one place. */
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { join } from 'node:path'
|
||||
import { PROTOCOL_VERSION } from './types'
|
||||
import { CLEAN_DISCONNECT_PROTOCOL_VERSION, PROTOCOL_VERSION } from './types'
|
||||
import { WEDGED_DAEMON_GRACE_RETRIES } from './daemon-init'
|
||||
|
||||
const FAKE_USER_DATA_PATH = '/fake/userData'
|
||||
@@ -140,7 +140,10 @@ const {
|
||||
const disconnectOnlyErrors: Error[] = []
|
||||
const routerSubscriptionError: { current: Error | null } = { current: null }
|
||||
// Why: legacy adapters retire only when provably empty, so tests opt in rather than defaulting to it.
|
||||
const retireEmptyLegacyDaemons: { current: boolean } = { current: false }
|
||||
const retireEmptyLegacyDaemons: {
|
||||
current: boolean
|
||||
pendingByProtocol: Map<number, Promise<boolean>>
|
||||
} = { current: false, pendingByProtocol: new Map() }
|
||||
// Same for DaemonPtyAdapter — tests assert the replacement adapter is fresh but its respawn closure targets the *original* spawner.
|
||||
const adapterInstances: MockAdapter[] = []
|
||||
// Why: adapters are built inside initDaemonPtyProvider, so tests set this before init to make listSessions report live sessions.
|
||||
@@ -252,6 +255,7 @@ type MockAdapter = {
|
||||
shutdown: ReturnType<typeof vi.fn>
|
||||
dispose: ReturnType<typeof vi.fn>
|
||||
disconnectOnly: ReturnType<typeof vi.fn>
|
||||
retireIfEmpty: ReturnType<typeof vi.fn>
|
||||
onData: ReturnType<typeof vi.fn>
|
||||
onExit: ReturnType<typeof vi.fn>
|
||||
// Why: the router calls onData/onExit on each adapter; the stub returns a no-op unsubscribe so router subscription doesn't explode.
|
||||
@@ -410,7 +414,15 @@ vi.mock('./daemon-pty-adapter', () => ({
|
||||
}
|
||||
})
|
||||
// Default preserves the daemon: retirement is opt-in per test, and fail-closed is the production rule.
|
||||
this.retireIfEmpty = vi.fn(async () => retireEmptyLegacyDaemons.current)
|
||||
this.retireIfEmpty = vi.fn(async () => {
|
||||
if (this.protocolVersion < CLEAN_DISCONNECT_PROTOCOL_VERSION) {
|
||||
return false
|
||||
}
|
||||
return (
|
||||
retireEmptyLegacyDaemons.pendingByProtocol.get(this.protocolVersion) ??
|
||||
retireEmptyLegacyDaemons.current
|
||||
)
|
||||
})
|
||||
this.onData = vi.fn(() => {
|
||||
if (routerSubscriptionError.current) {
|
||||
const error = routerSubscriptionError.current
|
||||
@@ -440,6 +452,7 @@ async function importFresh() {
|
||||
lifecycleLeaseErrors.length = 0
|
||||
disconnectOnlyErrors.length = 0
|
||||
retireEmptyLegacyDaemons.current = false
|
||||
retireEmptyLegacyDaemons.pendingByProtocol.clear()
|
||||
routerSubscriptionError.current = null
|
||||
adapterInstances.length = 0
|
||||
defaultListSessionsSessions.length = 0
|
||||
@@ -514,9 +527,9 @@ function mockConnectedAdoptionClientOnce(): void {
|
||||
})
|
||||
}
|
||||
|
||||
function mockOnlyDaemonSocketAlive(socketSuffix: string): void {
|
||||
function mockOnlyDaemonSocketAlive(...socketSuffixes: string[]): void {
|
||||
netConnectMock.mockImplementation((options?: { path?: string }) => {
|
||||
const live = options?.path?.endsWith(socketSuffix) ?? false
|
||||
const live = socketSuffixes.some((suffix) => options?.path?.endsWith(suffix))
|
||||
const handlers: Record<string, (() => void)[]> = { connect: [], error: [] }
|
||||
return {
|
||||
on(event: string, callback: () => void) {
|
||||
@@ -1044,6 +1057,54 @@ describe('daemon-init: runRestartDaemon (7-step sequence)', () => {
|
||||
expect(adapterInstances.some((instance) => instance.protocolVersion === 9)).toBe(true)
|
||||
})
|
||||
|
||||
it('preserves a pre-clean-disconnect daemon even when it is empty', async () => {
|
||||
const mod = await importFresh()
|
||||
const protocolVersion = CLEAN_DISCONNECT_PROTOCOL_VERSION - 1
|
||||
probeSocketExistsMock.mockImplementation(
|
||||
(path?: string) => path?.endsWith(`daemon-v${protocolVersion}.sock`) ?? false
|
||||
)
|
||||
mockOnlyDaemonSocketAlive(`daemon-v${protocolVersion}.sock`)
|
||||
retireEmptyLegacyDaemons.current = true
|
||||
|
||||
const adapters = await mod.createLegacyDaemonAdapters('/fake/daemon')
|
||||
|
||||
expect(adapters.map((candidate) => candidate.protocolVersion)).toEqual([protocolVersion])
|
||||
expect(adapterInstances[0].retireIfEmpty).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('checks supported legacy generations concurrently', async () => {
|
||||
const mod = await importFresh()
|
||||
const protocolVersions = [
|
||||
CLEAN_DISCONNECT_PROTOCOL_VERSION,
|
||||
CLEAN_DISCONNECT_PROTOCOL_VERSION + 1
|
||||
]
|
||||
probeSocketExistsMock.mockImplementation((path?: string) =>
|
||||
protocolVersions.some((version) => path?.endsWith(`daemon-v${version}.sock`))
|
||||
)
|
||||
mockOnlyDaemonSocketAlive(...protocolVersions.map((version) => `daemon-v${version}.sock`))
|
||||
const releases: ((retired: boolean) => void)[] = []
|
||||
for (const protocolVersion of protocolVersions) {
|
||||
retireEmptyLegacyDaemons.pendingByProtocol.set(
|
||||
protocolVersion,
|
||||
new Promise<boolean>((resolve) => releases.push(resolve))
|
||||
)
|
||||
}
|
||||
|
||||
const discovery = mod.createLegacyDaemonAdapters('/fake/daemon')
|
||||
await vi.waitFor(() => {
|
||||
expect(
|
||||
adapterInstances
|
||||
.filter((candidate) => protocolVersions.includes(candidate.protocolVersion))
|
||||
.map((candidate) => candidate.retireIfEmpty.mock.calls.length)
|
||||
).toEqual([1, 1])
|
||||
})
|
||||
for (const release of releases) {
|
||||
release(true)
|
||||
}
|
||||
|
||||
await expect(discovery).resolves.toEqual([])
|
||||
})
|
||||
|
||||
it('restart path with no legacy adapters yields a bare DaemonPtyAdapter (not wrapped in a router)', async () => {
|
||||
const mod = await importFresh()
|
||||
await mod.initDaemonPtyProvider()
|
||||
|
||||
@@ -1339,50 +1339,46 @@ export async function createLegacyDaemonAdapters(
|
||||
runtimeDir: string,
|
||||
historyPath = getHistoryDir()
|
||||
): Promise<DaemonPtyAdapter[]> {
|
||||
const adapters: DaemonPtyAdapter[] = []
|
||||
for (const protocolVersion of PREVIOUS_DAEMON_PROTOCOL_VERSIONS) {
|
||||
const socketPath = getDaemonSocketPath(runtimeDir, protocolVersion)
|
||||
const tokenPath = getDaemonTokenPath(runtimeDir, protocolVersion)
|
||||
if (!(await probeSocket(socketPath))) {
|
||||
// Why: a recycled stale pid later turns an identity check into a PowerShell spawn, so delete leaked pid/token files — but only when the pid-process is provably gone (a live daemon can transiently fail the probe, and dropping its token makes its sessions permanently unadoptable).
|
||||
if (!legacyDaemonProcessMayBeAlive(runtimeDir, protocolVersion)) {
|
||||
for (const stalePath of [
|
||||
getDaemonPidPath(runtimeDir, protocolVersion),
|
||||
getDaemonTokenPath(runtimeDir, protocolVersion)
|
||||
]) {
|
||||
try {
|
||||
unlinkSync(stalePath)
|
||||
} catch {
|
||||
// Best-effort
|
||||
const adapters = await Promise.all(
|
||||
PREVIOUS_DAEMON_PROTOCOL_VERSIONS.map(async (protocolVersion) => {
|
||||
const socketPath = getDaemonSocketPath(runtimeDir, protocolVersion)
|
||||
const tokenPath = getDaemonTokenPath(runtimeDir, protocolVersion)
|
||||
if (!(await probeSocket(socketPath))) {
|
||||
// Why: a recycled stale pid later turns an identity check into a PowerShell spawn, so delete leaked pid/token files — but only when the pid-process is provably gone (a live daemon can transiently fail the probe, and dropping its token makes its sessions permanently unadoptable).
|
||||
if (!legacyDaemonProcessMayBeAlive(runtimeDir, protocolVersion)) {
|
||||
for (const stalePath of [
|
||||
getDaemonPidPath(runtimeDir, protocolVersion),
|
||||
getDaemonTokenPath(runtimeDir, protocolVersion)
|
||||
]) {
|
||||
try {
|
||||
unlinkSync(stalePath)
|
||||
} catch {
|
||||
// Best-effort
|
||||
}
|
||||
}
|
||||
}
|
||||
if (process.platform !== 'win32' && existsSync(socketPath)) {
|
||||
try {
|
||||
unlinkSync(socketPath)
|
||||
} catch {
|
||||
// Best-effort
|
||||
if (process.platform !== 'win32' && existsSync(socketPath)) {
|
||||
try {
|
||||
unlinkSync(socketPath)
|
||||
} catch {
|
||||
// Best-effort
|
||||
}
|
||||
}
|
||||
}
|
||||
return null
|
||||
}
|
||||
continue
|
||||
}
|
||||
// Keep old-protocol PTYs routed to their original daemon during upgrade; legacy adapters never respawn (new code would recreate stale env semantics).
|
||||
// historyPath is still needed for cleanup — without it a later v4 session reusing the same ID could false-restore stale scrollback.bin.
|
||||
const adapter = new DaemonPtyAdapter({
|
||||
socketPath,
|
||||
tokenPath,
|
||||
pidPath: getDaemonPidPath(runtimeDir, protocolVersion),
|
||||
profileScope: runtimeDir,
|
||||
protocolVersion,
|
||||
historyPath
|
||||
// Keep old-protocol PTYs routed to their original daemon during upgrade; legacy adapters never respawn (new code would recreate stale env semantics).
|
||||
// historyPath is still needed for cleanup — without it a later v4 session reusing the same ID could false-restore stale scrollback.bin.
|
||||
const adapter = new DaemonPtyAdapter({
|
||||
socketPath,
|
||||
tokenPath,
|
||||
pidPath: getDaemonPidPath(runtimeDir, protocolVersion),
|
||||
profileScope: runtimeDir,
|
||||
protocolVersion,
|
||||
historyPath
|
||||
})
|
||||
// Why: bounded parallel retirement keeps multiple stranded generations off the serial startup path.
|
||||
return (await adapter.retireIfEmpty()) ? null : adapter
|
||||
})
|
||||
// Why: a superseded daemon has no retirement path on win32, so one strands per protocol bump and
|
||||
// lingers for the machine's uptime. Retiring the provably empty ones keeps the stranded set to the
|
||||
// daemons that still own something; a non-empty or unverifiable daemon is preserved and routed.
|
||||
if (await adapter.retireIfEmpty()) {
|
||||
continue
|
||||
}
|
||||
adapters.push(adapter)
|
||||
}
|
||||
return adapters
|
||||
)
|
||||
return adapters.filter((candidate): candidate is DaemonPtyAdapter => candidate !== null)
|
||||
}
|
||||
|
||||
@@ -2247,13 +2247,10 @@ export class DaemonPtyAdapter implements IPtyProvider {
|
||||
await this.respawnPromise
|
||||
}
|
||||
|
||||
/**
|
||||
* Retire a superseded daemon that owns nothing. Windows has no endpoint-ownership retirement, so one
|
||||
* daemon strands per protocol bump and each becomes an unmonitored owner of whatever it still holds.
|
||||
* Fail closed: an unverifiable inventory preserves the daemon, and `shutdownIfIdle` re-proves emptiness
|
||||
* daemon-side before it exits. Bounded because this runs on the startup path.
|
||||
*/
|
||||
async retireIfEmpty(budgetMs = 1_000): Promise<boolean> {
|
||||
if (this.protocolVersion < CLEAN_DISCONNECT_PROTOCOL_VERSION) {
|
||||
return false
|
||||
}
|
||||
const deadlineMs = Date.now() + budgetMs
|
||||
const remaining = (): number => Math.max(1, deadlineMs - Date.now())
|
||||
try {
|
||||
@@ -2268,10 +2265,20 @@ export class DaemonPtyAdapter implements IPtyProvider {
|
||||
if (result.sessions.some((session) => session.isAlive)) {
|
||||
return false
|
||||
}
|
||||
// Why: only this response atomically fences creates after the daemon-side empty proof.
|
||||
const retirement = await this.client.request<{ retiring: boolean }>(
|
||||
'shutdownIfIdle',
|
||||
undefined,
|
||||
remaining()
|
||||
)
|
||||
if (!retirement.retiring) {
|
||||
return false
|
||||
}
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
await this.disconnectOnly()
|
||||
this.respawnAdoptionClosed = true
|
||||
this.client.disconnect()
|
||||
return true
|
||||
}
|
||||
|
||||
|
||||
@@ -3,12 +3,14 @@
|
||||
* lingers for the machine's uptime — five accumulated on one host, the oldest 12 days old. A legacy
|
||||
* daemon that owns nothing must retire; one that still owns sessions, or cannot be inventoried, must not.
|
||||
*/
|
||||
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import { createServer, type Socket } from 'node:net'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import type { DaemonClient } from './client'
|
||||
import { DaemonPtyAdapter } from './daemon-pty-adapter'
|
||||
import { CLEAN_DISCONNECT_PROTOCOL_VERSION, PROTOCOL_VERSION } from './daemon-protocol-version'
|
||||
import { DaemonServer } from './daemon-server'
|
||||
import { getDaemonSocketPath } from './daemon-spawner'
|
||||
import type { SubprocessHandle } from './session'
|
||||
@@ -57,10 +59,11 @@ describe('legacy daemon retirement', () => {
|
||||
rmSync(dir, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
async function startServer(): Promise<DaemonServer> {
|
||||
async function startServer(protocolVersion = PROTOCOL_VERSION): Promise<DaemonServer> {
|
||||
const started = new DaemonServer({
|
||||
socketPath,
|
||||
tokenPath,
|
||||
protocolVersion,
|
||||
spawnSubprocess: () => fixtureSubprocess()
|
||||
})
|
||||
await started.start()
|
||||
@@ -68,10 +71,11 @@ describe('legacy daemon retirement', () => {
|
||||
return started
|
||||
}
|
||||
|
||||
function createAdapter(): DaemonPtyAdapter {
|
||||
function createAdapter(protocolVersion = PROTOCOL_VERSION): DaemonPtyAdapter {
|
||||
adapter = new DaemonPtyAdapter({
|
||||
socketPath,
|
||||
tokenPath,
|
||||
protocolVersion,
|
||||
profileScope: dir,
|
||||
historyPath: join(dir, 'history')
|
||||
})
|
||||
@@ -101,6 +105,30 @@ describe('legacy daemon retirement', () => {
|
||||
expect(await owner.listSessions()).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('preserves an empty daemon when another client wins the retirement race', async () => {
|
||||
await startServer()
|
||||
owner = new DaemonPtyAdapter({ socketPath, tokenPath, profileScope: dir })
|
||||
await owner.establishLifecycleLease()
|
||||
const candidate = createAdapter()
|
||||
|
||||
const retired = await candidate.retireIfEmpty()
|
||||
await owner.spawn({ sessionId: 'created-after-refusal', cols: 80, rows: 24 })
|
||||
|
||||
expect(retired).toBe(false)
|
||||
expect(await candidate.listSessions()).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('preserves protocols that predate atomic idle retirement', async () => {
|
||||
const protocolVersion = CLEAN_DISCONNECT_PROTOCOL_VERSION - 1
|
||||
await startServer(protocolVersion)
|
||||
const candidate = createAdapter(protocolVersion)
|
||||
|
||||
const retired = await candidate.retireIfEmpty()
|
||||
|
||||
expect(retired).toBe(false)
|
||||
expect(await candidate.listSessions()).toEqual([])
|
||||
})
|
||||
|
||||
it('fails closed when the daemon cannot be reached at all', async () => {
|
||||
// No server listening: the inventory is unverifiable, so the daemon must be preserved and routed.
|
||||
const retired = await createAdapter().retireIfEmpty()
|
||||
@@ -123,7 +151,7 @@ describe('legacy daemon retirement', () => {
|
||||
const retired = await createAdapter().retireIfEmpty(200)
|
||||
|
||||
expect(retired).toBe(false)
|
||||
expect(Date.now() - startedAt).toBeLessThan(5_000)
|
||||
expect(Date.now() - startedAt).toBeLessThan(1_000)
|
||||
} finally {
|
||||
// close() alone waits on live connections, and the adapter's sockets are still open.
|
||||
for (const socket of accepted) {
|
||||
@@ -132,4 +160,39 @@ describe('legacy daemon retirement', () => {
|
||||
await new Promise<void>((resolve) => wedged.close(() => resolve()))
|
||||
}
|
||||
})
|
||||
|
||||
it('shares one budget across inventory and a wedged atomic shutdown', async () => {
|
||||
const started = await startServer()
|
||||
const serverInternals = started as unknown as {
|
||||
routeRequest(clientId: string, request: { type: string }): Promise<unknown>
|
||||
}
|
||||
const routeRequest = serverInternals.routeRequest.bind(started)
|
||||
serverInternals.routeRequest = async (clientId, request) => {
|
||||
if (request.type === 'listSessions') {
|
||||
await new Promise((resolve) => setTimeout(resolve, 120))
|
||||
} else if (request.type === 'shutdownIfIdle') {
|
||||
return new Promise<never>(() => {})
|
||||
}
|
||||
return routeRequest(clientId, request)
|
||||
}
|
||||
const candidate = createAdapter()
|
||||
const client = (candidate as unknown as { client: DaemonClient }).client
|
||||
const request = vi.spyOn(client, 'request')
|
||||
const startedAt = Date.now()
|
||||
|
||||
let retired: boolean
|
||||
try {
|
||||
retired = await candidate.retireIfEmpty(200)
|
||||
} finally {
|
||||
serverInternals.routeRequest = routeRequest
|
||||
}
|
||||
|
||||
const elapsedMs = Date.now() - startedAt
|
||||
const shutdownTimeoutMs = request.mock.calls.find(([type]) => type === 'shutdownIfIdle')?.[2]
|
||||
expect(retired).toBe(false)
|
||||
expect(shutdownTimeoutMs).toEqual(expect.any(Number))
|
||||
expect(shutdownTimeoutMs).toBeGreaterThan(0)
|
||||
expect(shutdownTimeoutMs).toBeLessThan(150)
|
||||
expect(elapsedMs).toBeLessThan(300)
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user