mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 16:03:47 +00:00
fix(ssh): stop a failed relay --connect from orphaning the live relay (STA-1756)
When `--connect` to an existing detached relay failed, deploy removed the socket and launched a fresh relay over the same path. On Unix, unlinking a socket path does not close a listening socket the owner already holds: the old relay stayed alive, kept its PTYs and every shell/agent inside them, and became permanently unreachable. With the shipped `--grace-time 0` a relay holding a PTY arms no grace timer, so nothing ever reaped it. The relay already refuses this displacement itself — a fresh daemon that hits EADDRINUSE probe-connects the path and declines to steal it from a live owner. The unconditional `rm -f` destroyed exactly that evidence. Remove the socket only on positive proof that nothing is listening (a connect probe answered ECONNREFUSED/ENOENT). A live or unverifiable owner leaves the socket alone and fails the deploy with an actionable error, per the rule in docs/reference/ssh-execution-boundary.md that loss of contact is never evidence of process death. Stale sockets from a crashed relay still get cleaned up. Covered at three levels: the owner probe and its verdicts, the deploy wiring (no rm -f and no fresh launch unless the owner is proven gone), and a live oracle that runs real relay daemons holding a real PTY — including a control case that reproduces the orphan when the socket is unlinked first.
This commit is contained in:
@@ -186,27 +186,6 @@ describe('deployAndLaunchRelay', () => {
|
||||
expect(progress).toContain('Starting relay...')
|
||||
})
|
||||
|
||||
it('does not launch fresh after unconfirmed stale-socket cleanup', async () => {
|
||||
const conn = makeMockConnection()
|
||||
const unconfirmedCleanup = Object.assign(new Error('socket cleanup still running'), {
|
||||
sshChannelCloseConfirmed: false
|
||||
})
|
||||
vi.mocked(waitForSentinel).mockRejectedValueOnce(new Error('stale relay reconnect failed'))
|
||||
vi.mocked(execCommand)
|
||||
.mockResolvedValueOnce('__ORCA_REMOTE_PLATFORM__ Linux x86_64')
|
||||
.mockResolvedValueOnce('/home/user')
|
||||
.mockResolvedValueOnce('ORCA-NATIVE-DEPS-OK')
|
||||
.mockResolvedValueOnce('') // launch namespace marker
|
||||
.mockResolvedValueOnce('ALIVE')
|
||||
.mockRejectedValueOnce(unconfirmedCleanup)
|
||||
|
||||
await expect(deployAndLaunchRelay(conn)).rejects.toBe(unconfirmedCleanup)
|
||||
|
||||
const commands = vi.mocked(conn.exec).mock.calls.map(([command]) => command)
|
||||
expect(commands).toHaveLength(1)
|
||||
expect(commands.some((command) => command.includes('--detached'))).toBe(false)
|
||||
})
|
||||
|
||||
it('resolves the remote node path once per deploy', async () => {
|
||||
const conn = makeMockConnection()
|
||||
const mockExecCommand = vi.mocked(execCommand)
|
||||
|
||||
@@ -77,6 +77,7 @@ import {
|
||||
import { detectRemoteHostPlatform } from './ssh-remote-platform-detection'
|
||||
import { powerShellCommand, powerShellLiteral, powerShellNativeArg } from './ssh-remote-powershell'
|
||||
import { relaySocketNameForInstanceId } from './ssh-relay-instance-id'
|
||||
import { isRelaySocketOwnerLiveError, removeUnownedRelaySocket } from './ssh-relay-socket-owner'
|
||||
import { isSshSessionLimitError } from './ssh-session-limit-error'
|
||||
import {
|
||||
isWindowsRelayPipePath,
|
||||
@@ -1436,22 +1437,20 @@ async function launchRelay(
|
||||
} catch (err) {
|
||||
signal?.throwIfAborted()
|
||||
console.warn(
|
||||
'[ssh-relay] Socket reconnect failed, launching fresh relay:',
|
||||
'[ssh-relay] Socket reconnect failed:',
|
||||
err instanceof Error ? err.message : String(err)
|
||||
)
|
||||
// Why: stale socket from a crashed relay — remove it so the fresh launch can bind at the same path.
|
||||
await execCommand(conn, `rm -f ${shellEscape(sockFile)}`, { signal }).catch(
|
||||
(cleanupErr) => {
|
||||
if (isUnconfirmedSshCommandTermination(cleanupErr)) {
|
||||
throw cleanupErr
|
||||
}
|
||||
}
|
||||
)
|
||||
signal?.throwIfAborted()
|
||||
// Why: a refused --connect is loss of contact, never proof the relay died, so the socket is
|
||||
// removed only once a probe shows nothing is listening. Unlinking a live owner's path cannot
|
||||
// stop it — it just hides it from the fresh launch's EADDRINUSE check and orphans it and its
|
||||
// PTYs forever (STA-1756). See docs/reference/ssh-execution-boundary.md.
|
||||
await removeUnownedRelaySocket(conn, nodePath, sockFile, { signal, cause: err })
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
if (isUnconfirmedSshCommandTermination(err)) {
|
||||
if (isUnconfirmedSshCommandTermination(err) || isRelaySocketOwnerLiveError(err)) {
|
||||
// Why rethrow: falling through would launch a fresh relay over a socket path whose
|
||||
// owner is still listening, which is exactly the displacement this branch refused.
|
||||
throw err
|
||||
}
|
||||
signal?.throwIfAborted()
|
||||
|
||||
@@ -0,0 +1,279 @@
|
||||
// Runs real relay daemons on the host that runs the test, so socket-ownership
|
||||
// claims are settled by actual processes rather than by mocked exec output.
|
||||
//
|
||||
// Needs `pnpm build:relay` — callers skip when relayBundleDirForHost() is null.
|
||||
|
||||
import { spawn, type ChildProcess } from 'node:child_process'
|
||||
import { randomBytes } from 'node:crypto'
|
||||
import { EventEmitter } from 'node:events'
|
||||
import { cpSync, existsSync, mkdirSync, symlinkSync, writeFileSync } from 'node:fs'
|
||||
import { connect } from 'node:net'
|
||||
import { join } from 'node:path'
|
||||
import type { SshConnection } from './ssh-connection'
|
||||
|
||||
const FRAME_HEADER_BYTES = 13
|
||||
const MESSAGE_TYPE_REGULAR = 1
|
||||
|
||||
export function relayBundleDirForHost(repoRoot: string): string | null {
|
||||
const dir = join(repoRoot, 'out', 'relay', `${process.platform}-${process.arch}`)
|
||||
return existsSync(join(dir, 'relay.js')) ? dir : null
|
||||
}
|
||||
|
||||
export function delay(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms))
|
||||
}
|
||||
|
||||
export function isProcessAlive(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0)
|
||||
return true
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
export class LiveRelayFixture {
|
||||
readonly sockPath: string
|
||||
readonly credentialFile: string
|
||||
private readonly children = new Set<ChildProcess>()
|
||||
private readonly daemonPids = new Set<number>()
|
||||
|
||||
constructor(
|
||||
readonly relayDir: string,
|
||||
bundleDir: string,
|
||||
repoRoot: string
|
||||
) {
|
||||
mkdirSync(relayDir, { recursive: true })
|
||||
cpSync(bundleDir, relayDir, { recursive: true })
|
||||
// Why: node-pty is external to the relay bundle and must resolve beside relay.js.
|
||||
symlinkSync(join(repoRoot, 'node_modules'), join(relayDir, 'node_modules'))
|
||||
this.sockPath = join(relayDir, 'relay-ownership.sock')
|
||||
this.credentialFile = join(relayDir, 'relay-ownership.credential')
|
||||
writeFileSync(this.credentialFile, randomBytes(32).toString('base64url'), { mode: 0o600 })
|
||||
}
|
||||
|
||||
/** Launch a detached daemon exactly as ssh-relay-deploy does: --detached --grace-time 0. */
|
||||
launchDaemon(label: string): ChildProcess {
|
||||
const child = spawn(
|
||||
process.execPath,
|
||||
[
|
||||
join(this.relayDir, 'relay.js'),
|
||||
'--detached',
|
||||
'--grace-time',
|
||||
'0',
|
||||
'--sock-path',
|
||||
this.sockPath,
|
||||
'--credential-file',
|
||||
this.credentialFile,
|
||||
'--log-file',
|
||||
join(this.relayDir, `relay-${label}.log`)
|
||||
],
|
||||
{
|
||||
cwd: this.relayDir,
|
||||
detached: true,
|
||||
stdio: 'ignore',
|
||||
env: { ...process.env, ORCA_DISABLE_MACOS_LOGIN_SHELL: '1' }
|
||||
}
|
||||
)
|
||||
child.unref()
|
||||
if (child.pid) {
|
||||
this.daemonPids.add(child.pid)
|
||||
}
|
||||
this.children.add(child)
|
||||
return child
|
||||
}
|
||||
|
||||
/** 'READY' when a listener accepted, otherwise the connect error code. */
|
||||
connectProbe(): Promise<string> {
|
||||
return new Promise((resolve) => {
|
||||
const sock = connect(this.sockPath)
|
||||
sock.once('connect', () => {
|
||||
sock.destroy()
|
||||
resolve('READY')
|
||||
})
|
||||
sock.once('error', (err: NodeJS.ErrnoException) => resolve(err.code ?? 'ERROR'))
|
||||
})
|
||||
}
|
||||
|
||||
async waitForSocket(timeoutMs = 20_000): Promise<boolean> {
|
||||
const start = Date.now()
|
||||
while (Date.now() - start < timeoutMs) {
|
||||
if ((await this.connectProbe()) === 'READY') {
|
||||
return true
|
||||
}
|
||||
await delay(100)
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
openBridge(): RelayBridge {
|
||||
const bridge = new RelayBridge(this.relayDir, this.sockPath, this.credentialFile)
|
||||
this.children.add(bridge.process)
|
||||
return bridge
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
for (const child of this.children) {
|
||||
try {
|
||||
child.kill('SIGKILL')
|
||||
} catch {
|
||||
/* already gone */
|
||||
}
|
||||
}
|
||||
for (const pid of this.daemonPids) {
|
||||
try {
|
||||
process.kill(pid, 'SIGKILL')
|
||||
} catch {
|
||||
/* already gone */
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** A `relay.js --connect` bridge, driven over its stdio like the SSH exec channel does. */
|
||||
export class RelayBridge {
|
||||
readonly process: ChildProcess
|
||||
private buffer = Buffer.alloc(0)
|
||||
private sentinelSeen = false
|
||||
private nextId = 1
|
||||
private readonly pending = new Map<
|
||||
number,
|
||||
{ resolve: (value: never) => void; reject: (error: Error) => void }
|
||||
>()
|
||||
|
||||
constructor(relayDir: string, sockPath: string, credentialFile: string) {
|
||||
this.process = spawn(
|
||||
process.execPath,
|
||||
[
|
||||
join(relayDir, 'relay.js'),
|
||||
'--connect',
|
||||
'--sock-path',
|
||||
sockPath,
|
||||
'--credential-file',
|
||||
credentialFile
|
||||
],
|
||||
{ cwd: relayDir, stdio: ['pipe', 'pipe', 'pipe'] }
|
||||
)
|
||||
this.process.stderr?.resume()
|
||||
this.process.stdout?.on('data', (chunk: Buffer) => this.ingest(chunk))
|
||||
}
|
||||
|
||||
private ingest(chunk: Buffer): void {
|
||||
this.buffer = Buffer.concat([this.buffer, chunk])
|
||||
if (!this.sentinelSeen) {
|
||||
const newline = this.buffer.indexOf('\n')
|
||||
if (newline === -1) {
|
||||
return
|
||||
}
|
||||
this.sentinelSeen = true
|
||||
this.buffer = this.buffer.subarray(newline + 1)
|
||||
}
|
||||
while (this.buffer.length >= FRAME_HEADER_BYTES) {
|
||||
const length = this.buffer.readUInt32BE(9)
|
||||
if (this.buffer.length < FRAME_HEADER_BYTES + length) {
|
||||
return
|
||||
}
|
||||
const type = this.buffer[0]
|
||||
const payload = this.buffer.subarray(FRAME_HEADER_BYTES, FRAME_HEADER_BYTES + length)
|
||||
this.buffer = this.buffer.subarray(FRAME_HEADER_BYTES + length)
|
||||
if (type !== MESSAGE_TYPE_REGULAR) {
|
||||
continue
|
||||
}
|
||||
let message: { id?: number; result?: unknown; error?: unknown }
|
||||
try {
|
||||
message = JSON.parse(payload.toString('utf-8'))
|
||||
} catch {
|
||||
continue
|
||||
}
|
||||
const waiter = message.id === undefined ? undefined : this.pending.get(message.id)
|
||||
if (!waiter || message.id === undefined) {
|
||||
continue
|
||||
}
|
||||
this.pending.delete(message.id)
|
||||
if (message.error) {
|
||||
waiter.reject(new Error(JSON.stringify(message.error)))
|
||||
} else {
|
||||
waiter.resolve(message.result as never)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async waitForSentinel(timeoutMs = 20_000): Promise<void> {
|
||||
const start = Date.now()
|
||||
while (Date.now() - start < timeoutMs) {
|
||||
if (this.sentinelSeen) {
|
||||
return
|
||||
}
|
||||
await delay(50)
|
||||
}
|
||||
throw new Error('relay --connect never emitted its sentinel')
|
||||
}
|
||||
|
||||
request<T>(method: string, params: Record<string, unknown> = {}, timeoutMs = 20_000): Promise<T> {
|
||||
const id = this.nextId++
|
||||
const payload = Buffer.from(JSON.stringify({ jsonrpc: '2.0', id, method, params }), 'utf-8')
|
||||
const header = Buffer.alloc(FRAME_HEADER_BYTES)
|
||||
header[0] = MESSAGE_TYPE_REGULAR
|
||||
header.writeUInt32BE(0, 1)
|
||||
header.writeUInt32BE(0, 5)
|
||||
header.writeUInt32BE(payload.length, 9)
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
this.pending.delete(id)
|
||||
reject(new Error(`${method} timed out`))
|
||||
}, timeoutMs)
|
||||
this.pending.set(id, {
|
||||
resolve: ((value: T) => {
|
||||
clearTimeout(timer)
|
||||
resolve(value)
|
||||
}) as (value: never) => void,
|
||||
reject: (error: Error) => {
|
||||
clearTimeout(timer)
|
||||
reject(error)
|
||||
}
|
||||
})
|
||||
this.process.stdin?.write(Buffer.concat([header, payload]))
|
||||
})
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.process.kill('SIGKILL')
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* An SshConnection whose exec runs on this machine, so production command
|
||||
* strings and production execCommand parsing are both exercised for real.
|
||||
*/
|
||||
export function createLocalShellConnection(): SshConnection {
|
||||
return {
|
||||
usesSystemSshTransport: () => false,
|
||||
canRunConcurrentExecCommands: () => true,
|
||||
exec: (command: string) => Promise.resolve(createLocalShellChannel(command))
|
||||
} as unknown as SshConnection
|
||||
}
|
||||
|
||||
function createLocalShellChannel(command: string): EventEmitter & {
|
||||
stderr: EventEmitter
|
||||
close: () => void
|
||||
resume: () => void
|
||||
} {
|
||||
const channel = new EventEmitter() as EventEmitter & {
|
||||
stderr: EventEmitter & { resume: () => void }
|
||||
close: () => void
|
||||
resume: () => void
|
||||
}
|
||||
const stderr = new EventEmitter() as EventEmitter & { resume: () => void }
|
||||
stderr.resume = (): void => {}
|
||||
channel.stderr = stderr
|
||||
channel.resume = (): void => {}
|
||||
const child = spawn('/bin/sh', ['-c', command])
|
||||
channel.close = (): void => {
|
||||
child.kill('SIGKILL')
|
||||
}
|
||||
child.stdout.on('data', (chunk: Buffer) => channel.emit('data', chunk))
|
||||
child.stderr.on('data', (chunk: Buffer) => stderr.emit('data', chunk))
|
||||
child.on('error', (error: Error) => channel.emit('error', error))
|
||||
child.on('close', (code: number | null) => channel.emit('close', code ?? 0))
|
||||
return channel
|
||||
}
|
||||
@@ -0,0 +1,193 @@
|
||||
// STA-1756: a failed --connect used to `rm -f` the relay socket and launch a fresh
|
||||
// detached relay over the same path. Unlinking cannot stop a relay that already bound
|
||||
// that path — it only hides the live owner from the fresh launch's EADDRINUSE check,
|
||||
// so the old relay keeps running forever with its PTYs and every agent inside them.
|
||||
//
|
||||
// These pin the shipping wiring: deploy must not remove a socket, and must not launch
|
||||
// a replacement, unless the owner probe proves nothing is listening.
|
||||
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
vi.mock('electron', () => ({
|
||||
app: { getAppPath: () => '/mock/app' }
|
||||
}))
|
||||
|
||||
vi.mock('fs', () => ({
|
||||
existsSync: vi.fn().mockReturnValue(true),
|
||||
readFileSync: vi.fn().mockReturnValue('0.1.0+abcdef012345')
|
||||
}))
|
||||
|
||||
vi.mock('./relay-protocol', () => ({
|
||||
RELAY_VERSION: '0.1.0',
|
||||
RELAY_REMOTE_DIR: '.orca-remote',
|
||||
parseUnameToRelayPlatform: vi.fn().mockReturnValue('linux-x64'),
|
||||
RELAY_SENTINEL: 'ORCA-RELAY v0.1.0 READY\n',
|
||||
RELAY_SENTINEL_TIMEOUT_MS: 10_000
|
||||
}))
|
||||
|
||||
vi.mock('./ssh-relay-deploy-helpers', () => ({
|
||||
uploadDirectory: vi.fn().mockResolvedValue(undefined),
|
||||
waitForSentinel: vi.fn(),
|
||||
isUnconfirmedSshCommandTermination: (error: unknown) =>
|
||||
error instanceof Error &&
|
||||
(error as Error & { sshChannelCloseConfirmed?: boolean }).sshChannelCloseConfirmed === false,
|
||||
execCommand: vi.fn()
|
||||
}))
|
||||
|
||||
vi.mock('./ssh-remote-node-resolution', () => ({
|
||||
resolveRemoteNodePath: vi.fn().mockResolvedValue('/usr/bin/node')
|
||||
}))
|
||||
|
||||
vi.mock('./ssh-relay-endpoint-credential', () => ({
|
||||
writeRelayEndpointCredential: vi.fn().mockResolvedValue(undefined)
|
||||
}))
|
||||
|
||||
vi.mock('./ssh-relay-versioned-install', () => ({
|
||||
readLocalFullVersion: vi.fn().mockReturnValue('0.1.0+abcdef012345'),
|
||||
computeRemoteRelayDir: (home: string, v: string) => `${home}/.orca-remote/relay-${v}`,
|
||||
isRelayAlreadyInstalled: vi.fn().mockResolvedValue(true),
|
||||
finalizeInstall: vi.fn().mockResolvedValue(undefined),
|
||||
abandonInstall: vi.fn().mockResolvedValue(undefined),
|
||||
gcOldRelayVersions: vi.fn().mockResolvedValue(undefined)
|
||||
}))
|
||||
|
||||
vi.mock('./ssh-relay-install-lock', () => ({
|
||||
acquireInstallLock: vi.fn().mockResolvedValue(undefined),
|
||||
RELAY_INSTALL_LOCK_NAME: '.install-lock'
|
||||
}))
|
||||
|
||||
vi.mock('./ssh-relay-repair-lock', () => ({
|
||||
tryAcquireRelayRepairLock: vi.fn().mockResolvedValue('acquired')
|
||||
}))
|
||||
|
||||
import { deployAndLaunchRelay } from './ssh-relay-deploy'
|
||||
import { execCommand, waitForSentinel } from './ssh-relay-deploy-helpers'
|
||||
import { writeRelayEndpointCredential } from './ssh-relay-endpoint-credential'
|
||||
import { isRelaySocketOwnerLiveError } from './ssh-relay-socket-owner'
|
||||
import type { SshConnection } from './ssh-connection'
|
||||
|
||||
function makeMockConnection(): SshConnection {
|
||||
return {
|
||||
canRunConcurrentExecCommands: vi.fn().mockReturnValue(true),
|
||||
exec: vi.fn().mockResolvedValue({
|
||||
on: vi.fn(),
|
||||
stderr: { on: vi.fn() },
|
||||
stdin: {},
|
||||
stdout: { on: vi.fn() },
|
||||
close: vi.fn()
|
||||
}),
|
||||
writeFile: vi.fn().mockResolvedValue(undefined),
|
||||
sftp: vi.fn().mockResolvedValue({
|
||||
mkdir: vi.fn((_p: string, cb: (err: Error | null) => void) => cb(null)),
|
||||
end: vi.fn()
|
||||
})
|
||||
} as unknown as SshConnection
|
||||
}
|
||||
|
||||
/** Queue the exec replies deploy makes before it decides what to do with an occupied socket. */
|
||||
function queueDeployUpToOccupiedSocket(): void {
|
||||
vi.mocked(execCommand)
|
||||
.mockResolvedValueOnce('__ORCA_REMOTE_PLATFORM__ Linux x86_64')
|
||||
.mockResolvedValueOnce('/home/user')
|
||||
.mockResolvedValueOnce('ORCA-NATIVE-DEPS-OK')
|
||||
.mockResolvedValueOnce('') // launch namespace marker
|
||||
.mockResolvedValueOnce('ALIVE') // test -S: a socket inode is present
|
||||
}
|
||||
|
||||
function execCommands(): string[] {
|
||||
return vi.mocked(execCommand).mock.calls.map(([, command]) => command)
|
||||
}
|
||||
|
||||
function execChannelCommands(conn: SshConnection): string[] {
|
||||
return vi.mocked(conn.exec).mock.calls.map(([command]) => command)
|
||||
}
|
||||
|
||||
describe('deployAndLaunchRelay socket displacement', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
// Why: an ordered queue drives the probes below; the default only absorbs the
|
||||
// best-effort teardown commands deploy issues after the decision under test.
|
||||
vi.mocked(execCommand).mockReset().mockResolvedValue('')
|
||||
vi.mocked(waitForSentinel)
|
||||
.mockReset()
|
||||
.mockResolvedValue({ write: vi.fn(), onData: vi.fn(), onClose: vi.fn() } as never)
|
||||
})
|
||||
|
||||
it('never removes or replaces a socket a live relay still owns', async () => {
|
||||
const conn = makeMockConnection()
|
||||
vi.mocked(waitForSentinel).mockRejectedValueOnce(new Error('relay reconnect refused'))
|
||||
queueDeployUpToOccupiedSocket()
|
||||
vi.mocked(execCommand).mockResolvedValueOnce('LIVE') // owner probe
|
||||
|
||||
await expect(deployAndLaunchRelay(conn)).rejects.toSatisfy(isRelaySocketOwnerLiveError)
|
||||
|
||||
expect(execCommands().some((command) => command.startsWith('rm -f'))).toBe(false)
|
||||
expect(execChannelCommands(conn).some((command) => command.includes('--detached'))).toBe(false)
|
||||
// Why: the credential is rotated per fresh launch, so writing one would lock the
|
||||
// surviving relay out of every later --connect even if the socket were left alone.
|
||||
expect(writeRelayEndpointCredential).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('never removes or replaces a socket whose owner it could not probe', async () => {
|
||||
const conn = makeMockConnection()
|
||||
vi.mocked(waitForSentinel).mockRejectedValueOnce(new Error('relay reconnect refused'))
|
||||
queueDeployUpToOccupiedSocket()
|
||||
vi.mocked(execCommand).mockResolvedValueOnce('') // owner probe produced no verdict
|
||||
|
||||
await expect(deployAndLaunchRelay(conn)).rejects.toSatisfy(isRelaySocketOwnerLiveError)
|
||||
|
||||
expect(execCommands().some((command) => command.startsWith('rm -f'))).toBe(false)
|
||||
expect(execChannelCommands(conn).some((command) => command.includes('--detached'))).toBe(false)
|
||||
})
|
||||
|
||||
it('still clears a stale socket whose owner is proven gone', async () => {
|
||||
const conn = makeMockConnection()
|
||||
vi.mocked(waitForSentinel)
|
||||
.mockRejectedValueOnce(new Error('relay reconnect refused'))
|
||||
.mockResolvedValue({ write: vi.fn(), onData: vi.fn(), onClose: vi.fn() } as never)
|
||||
queueDeployUpToOccupiedSocket()
|
||||
vi.mocked(execCommand)
|
||||
.mockResolvedValueOnce('EXITED') // owner probe: connect refused
|
||||
.mockResolvedValueOnce('') // rm -f
|
||||
.mockResolvedValueOnce('READY') // socket readiness poll
|
||||
|
||||
await expect(deployAndLaunchRelay(conn)).resolves.toMatchObject({ nodePath: '/usr/bin/node' })
|
||||
|
||||
const removals = execCommands().filter((command) => command.startsWith('rm -f'))
|
||||
expect(removals).toHaveLength(1)
|
||||
expect(removals[0]).toContain('.sock')
|
||||
expect(execChannelCommands(conn).some((command) => command.includes('--detached'))).toBe(true)
|
||||
})
|
||||
|
||||
// Moved here from ssh-relay-deploy.test.ts: the removal it guards now lives behind the owner probe.
|
||||
it('does not launch fresh after unconfirmed stale-socket cleanup', async () => {
|
||||
const conn = makeMockConnection()
|
||||
const unconfirmedCleanup = Object.assign(new Error('socket cleanup still running'), {
|
||||
sshChannelCloseConfirmed: false
|
||||
})
|
||||
vi.mocked(waitForSentinel).mockRejectedValueOnce(new Error('stale relay reconnect failed'))
|
||||
queueDeployUpToOccupiedSocket()
|
||||
vi.mocked(execCommand)
|
||||
.mockResolvedValueOnce('EXITED') // owner probe
|
||||
.mockRejectedValueOnce(unconfirmedCleanup)
|
||||
|
||||
await expect(deployAndLaunchRelay(conn)).rejects.toBe(unconfirmedCleanup)
|
||||
|
||||
expect(execChannelCommands(conn)).toHaveLength(1)
|
||||
expect(execChannelCommands(conn).some((command) => command.includes('--detached'))).toBe(false)
|
||||
})
|
||||
|
||||
it('probes the socket by connecting to it, not by testing the inode again', async () => {
|
||||
const conn = makeMockConnection()
|
||||
vi.mocked(waitForSentinel).mockRejectedValueOnce(new Error('relay reconnect refused'))
|
||||
queueDeployUpToOccupiedSocket()
|
||||
vi.mocked(execCommand).mockResolvedValueOnce('LIVE')
|
||||
|
||||
await expect(deployAndLaunchRelay(conn)).rejects.toSatisfy(isRelaySocketOwnerLiveError)
|
||||
|
||||
const ownerProbe = execCommands().at(-1) ?? ''
|
||||
expect(ownerProbe).toContain('/usr/bin/node')
|
||||
expect(ownerProbe).toContain('connect')
|
||||
expect(ownerProbe).not.toContain('test -S')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,128 @@
|
||||
// STA-1756 oracle, run against real relay daemons rather than mocked exec output.
|
||||
//
|
||||
// The reported defect: a failed `--connect` unlinked the relay socket and launched a
|
||||
// fresh detached relay over the same path. Unlinking cannot stop a daemon that already
|
||||
// bound that path, so the old relay stayed alive forever with its PTYs and every agent
|
||||
// inside them — invisible to the app, and never reaped (`--grace-time 0` arms no timer
|
||||
// while a PTY is held).
|
||||
//
|
||||
// The control case below is latest-main's behaviour: it performs the raw `rm -f` and
|
||||
// asserts the orphan appears. The guarded case asserts removeUnownedRelaySocket refuses
|
||||
// the same displacement, which restores the relay's own EADDRINUSE interlock.
|
||||
//
|
||||
// Requires `pnpm build:relay`; skipped when the bundle for this host is absent.
|
||||
|
||||
import { mkdtempSync, rmSync, existsSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import {
|
||||
LiveRelayFixture,
|
||||
createLocalShellConnection,
|
||||
delay,
|
||||
isProcessAlive,
|
||||
relayBundleDirForHost
|
||||
} from './ssh-relay-live-daemon-harness'
|
||||
import {
|
||||
isRelaySocketOwnerLiveError,
|
||||
probeRelaySocketOwner,
|
||||
removeUnownedRelaySocket
|
||||
} from './ssh-relay-socket-owner'
|
||||
|
||||
const REPO_ROOT = process.cwd()
|
||||
const BUNDLE_DIR = relayBundleDirForHost(REPO_ROOT)
|
||||
|
||||
type RelayStatus = {
|
||||
pid: number
|
||||
ptys: { active: number }
|
||||
socket: { path: string; owned: boolean; listening: boolean }
|
||||
}
|
||||
|
||||
describe.skipIf(!BUNDLE_DIR || process.platform === 'win32')('relay socket ownership', () => {
|
||||
let fixture: LiveRelayFixture | null = null
|
||||
let root: string | null = null
|
||||
|
||||
afterEach(() => {
|
||||
fixture?.dispose()
|
||||
fixture = null
|
||||
if (root) {
|
||||
rmSync(root, { recursive: true, force: true })
|
||||
root = null
|
||||
}
|
||||
})
|
||||
|
||||
async function startRelayHoldingAPty(): Promise<{ fixture: LiveRelayFixture; pid: number }> {
|
||||
root = mkdtempSync(join(tmpdir(), 'orca-relay-ownership-'))
|
||||
const live = new LiveRelayFixture(join(root, 'relay'), BUNDLE_DIR!, REPO_ROOT)
|
||||
fixture = live
|
||||
live.launchDaemon('incumbent')
|
||||
expect(await live.waitForSocket()).toBe(true)
|
||||
|
||||
const bridge = live.openBridge()
|
||||
await bridge.waitForSentinel()
|
||||
await bridge.request('pty.spawn', { cwd: root, cols: 80, rows: 24 })
|
||||
const status = await bridge.request<RelayStatus>('relay.status')
|
||||
expect(status.ptys.active).toBe(1)
|
||||
// Why: the app walking away is what starts the incumbent's grace window; with
|
||||
// --grace-time 0 and a live PTY that window arms no timer, so nothing reaps it.
|
||||
bridge.close()
|
||||
await delay(500)
|
||||
return { fixture: live, pid: status.pid }
|
||||
}
|
||||
|
||||
it('reports a daemon that is listening as live, and a socket nobody holds as exited', async () => {
|
||||
const { fixture: live, pid } = await startRelayHoldingAPty()
|
||||
const conn = createLocalShellConnection()
|
||||
|
||||
await expect(probeRelaySocketOwner(conn, process.execPath, live.sockPath)).resolves.toBe('live')
|
||||
|
||||
// A crashed relay leaves the inode behind; that, and only that, is a stale socket.
|
||||
process.kill(pid, 'SIGKILL')
|
||||
await delay(500)
|
||||
expect(existsSync(live.sockPath)).toBe(true)
|
||||
await expect(probeRelaySocketOwner(conn, process.execPath, live.sockPath)).resolves.toBe(
|
||||
'exited'
|
||||
)
|
||||
})
|
||||
|
||||
it('refuses to unlink the socket of a relay that still holds a PTY', async () => {
|
||||
const { fixture: live, pid } = await startRelayHoldingAPty()
|
||||
const conn = createLocalShellConnection()
|
||||
|
||||
await expect(removeUnownedRelaySocket(conn, process.execPath, live.sockPath)).rejects.toSatisfy(
|
||||
isRelaySocketOwnerLiveError
|
||||
)
|
||||
|
||||
expect(existsSync(live.sockPath)).toBe(true)
|
||||
expect(isProcessAlive(pid)).toBe(true)
|
||||
|
||||
// The socket kept its owner, so a fresh daemon hits EADDRINUSE, finds a live
|
||||
// listener, and refuses to start — the interlock the unconditional rm -f defeated.
|
||||
live.launchDaemon('replacement')
|
||||
await delay(2_000)
|
||||
const bridge = live.openBridge()
|
||||
await bridge.waitForSentinel()
|
||||
const status = await bridge.request<RelayStatus>('relay.status')
|
||||
expect(status.pid).toBe(pid)
|
||||
expect(status.ptys.active).toBe(1)
|
||||
bridge.close()
|
||||
})
|
||||
|
||||
it('control: unlinking first orphans the incumbent relay and its PTY', async () => {
|
||||
const { fixture: live, pid } = await startRelayHoldingAPty()
|
||||
|
||||
rmSync(live.sockPath, { force: true })
|
||||
live.launchDaemon('replacement')
|
||||
expect(await live.waitForSocket()).toBe(true)
|
||||
const bridge = live.openBridge()
|
||||
await bridge.waitForSentinel()
|
||||
const status = await bridge.request<RelayStatus>('relay.status')
|
||||
|
||||
// The socket path now answers for a different daemon that owns no PTYs, while the
|
||||
// incumbent is still running with the user's shell and no way to ever be reached.
|
||||
expect(status.pid).not.toBe(pid)
|
||||
expect(status.ptys.active).toBe(0)
|
||||
expect(isProcessAlive(pid)).toBe(true)
|
||||
bridge.close()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,123 @@
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
vi.mock('./ssh-relay-deploy-helpers', () => ({
|
||||
execCommand: vi.fn(),
|
||||
isUnconfirmedSshCommandTermination: (error: unknown) =>
|
||||
error instanceof Error &&
|
||||
(error as Error & { sshChannelCloseConfirmed?: boolean }).sshChannelCloseConfirmed === false
|
||||
}))
|
||||
|
||||
import type { SshConnection } from './ssh-connection'
|
||||
import { execCommand } from './ssh-relay-deploy-helpers'
|
||||
import {
|
||||
isRelaySocketOwnerLiveError,
|
||||
parseRelaySocketOwnerProbe,
|
||||
probeRelaySocketOwner,
|
||||
relaySocketOwnerProbeCommand,
|
||||
removeUnownedRelaySocket
|
||||
} from './ssh-relay-socket-owner'
|
||||
|
||||
const SOCK = '/home/u/.orca-remote/relay-0.1.0+abc/relay-deadbeef.sock'
|
||||
const NODE = '/usr/bin/node'
|
||||
const conn = {} as SshConnection
|
||||
|
||||
function commands(): string[] {
|
||||
return vi.mocked(execCommand).mock.calls.map(([, command]) => command)
|
||||
}
|
||||
|
||||
describe('relay socket owner probe', () => {
|
||||
beforeEach(() => {
|
||||
vi.mocked(execCommand).mockReset()
|
||||
})
|
||||
|
||||
it('probes by connecting, not by testing the inode', () => {
|
||||
const command = relaySocketOwnerProbeCommand(NODE, SOCK)
|
||||
expect(command).toContain('net')
|
||||
expect(command).toContain('connect')
|
||||
// Why: `test -S` is what the deploy probe already does, and it cannot tell a
|
||||
// listening owner from an inode a crashed relay left behind.
|
||||
expect(command).not.toContain('test -S')
|
||||
expect(command).toContain(`'${SOCK}'`)
|
||||
})
|
||||
|
||||
it('reads ECONNREFUSED and ENOENT as the only proof of an absent owner', () => {
|
||||
expect(parseRelaySocketOwnerProbe('LIVE')).toBe('live')
|
||||
expect(parseRelaySocketOwnerProbe('EXITED\n')).toBe('exited')
|
||||
expect(parseRelaySocketOwnerProbe('UNVERIFIABLE')).toBe('unverifiable')
|
||||
})
|
||||
|
||||
it('refuses to guess from ambiguous or empty probe output', () => {
|
||||
expect(parseRelaySocketOwnerProbe('')).toBe('unverifiable')
|
||||
expect(parseRelaySocketOwnerProbe('LIVE\nEXITED')).toBe('unverifiable')
|
||||
expect(parseRelaySocketOwnerProbe('command not found: node')).toBe('unverifiable')
|
||||
})
|
||||
|
||||
it('answers unverifiable when the probe cannot run at all', async () => {
|
||||
vi.mocked(execCommand).mockRejectedValueOnce(new Error('channel open failure'))
|
||||
await expect(probeRelaySocketOwner(conn, NODE, SOCK)).resolves.toBe('unverifiable')
|
||||
})
|
||||
|
||||
it('propagates an abort instead of reporting a verdict', async () => {
|
||||
const controller = new AbortController()
|
||||
vi.mocked(execCommand).mockImplementationOnce(() => {
|
||||
controller.abort()
|
||||
return Promise.reject(new Error('aborted'))
|
||||
})
|
||||
await expect(probeRelaySocketOwner(conn, NODE, SOCK, controller.signal)).rejects.toThrow()
|
||||
})
|
||||
})
|
||||
|
||||
describe('removeUnownedRelaySocket', () => {
|
||||
beforeEach(() => {
|
||||
vi.mocked(execCommand).mockReset()
|
||||
})
|
||||
|
||||
it('removes the socket once the probe proves nothing is listening', async () => {
|
||||
vi.mocked(execCommand).mockResolvedValueOnce('EXITED').mockResolvedValueOnce('')
|
||||
|
||||
await removeUnownedRelaySocket(conn, NODE, SOCK)
|
||||
|
||||
expect(commands()[1]).toBe(`rm -f '${SOCK}'`)
|
||||
})
|
||||
|
||||
it('leaves the socket of a live owner in place', async () => {
|
||||
vi.mocked(execCommand).mockResolvedValueOnce('LIVE')
|
||||
|
||||
await expect(removeUnownedRelaySocket(conn, NODE, SOCK)).rejects.toSatisfy(
|
||||
isRelaySocketOwnerLiveError
|
||||
)
|
||||
|
||||
expect(commands().some((command) => command.startsWith('rm -f'))).toBe(false)
|
||||
})
|
||||
|
||||
it('leaves the socket in place when ownership is merely unverifiable', async () => {
|
||||
vi.mocked(execCommand).mockResolvedValueOnce('UNVERIFIABLE')
|
||||
|
||||
await expect(removeUnownedRelaySocket(conn, NODE, SOCK)).rejects.toSatisfy(
|
||||
isRelaySocketOwnerLiveError
|
||||
)
|
||||
|
||||
expect(commands().some((command) => command.startsWith('rm -f'))).toBe(false)
|
||||
})
|
||||
|
||||
it('rethrows an unconfirmed removal so no fresh relay launches over it', async () => {
|
||||
const unconfirmed = Object.assign(new Error('socket cleanup still running'), {
|
||||
sshChannelCloseConfirmed: false
|
||||
})
|
||||
vi.mocked(execCommand).mockResolvedValueOnce('EXITED').mockRejectedValueOnce(unconfirmed)
|
||||
|
||||
await expect(removeUnownedRelaySocket(conn, NODE, SOCK)).rejects.toBe(unconfirmed)
|
||||
})
|
||||
|
||||
it('names the socket and the recovery action in the error', async () => {
|
||||
vi.mocked(execCommand).mockResolvedValueOnce('LIVE')
|
||||
const cause = new Error('handshake refused')
|
||||
|
||||
const error = await removeUnownedRelaySocket(conn, NODE, SOCK, { cause }).catch((err) => err)
|
||||
|
||||
expect(error.sockPath).toBe(SOCK)
|
||||
expect(error.verdict).toBe('live')
|
||||
expect(error.cause).toBe(cause)
|
||||
expect(error.message).toContain('Reset remote relay')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,117 @@
|
||||
import type { SshConnection } from './ssh-connection'
|
||||
import { shellEscape } from './ssh-connection-utils'
|
||||
import { execCommand, isUnconfirmedSshCommandTermination } from './ssh-relay-deploy-helpers'
|
||||
|
||||
/**
|
||||
* Who owns a relay socket path, as observed from the client.
|
||||
*
|
||||
* The vocabulary is pinned by docs/reference/ssh-execution-boundary.md:
|
||||
* `exited` requires positive evidence of absence, and anything we could not
|
||||
* observe stays `unverifiable` — never collapsed into either neighbour.
|
||||
* Here the subject is the socket path: `exited` means the connect attempt was
|
||||
* actively refused, so no process is listening at that path any more.
|
||||
*/
|
||||
export type RelaySocketOwnerVerdict = 'live' | 'exited' | 'unverifiable'
|
||||
|
||||
const PROBE_TIMEOUT_MS = 3_000
|
||||
|
||||
// Why: probe via node (guaranteed present on the host — it is what runs the relay)
|
||||
// rather than lsof/fuser/socat, which vary by distro and are often absent.
|
||||
// Why: `test -S` only proves the inode exists; a connect-and-close is the only
|
||||
// thing that separates a listening owner from an inode a crashed relay left behind.
|
||||
const RELAY_SOCKET_OWNER_PROBE_SCRIPT =
|
||||
'var net=require("net"),done=false,' +
|
||||
'say=function(v){if(done)return;done=true;clearTimeout(t);process.stdout.write(v);s.destroy()},' +
|
||||
's=net.connect(process.argv[1]),' +
|
||||
`t=setTimeout(function(){say("UNVERIFIABLE")},${PROBE_TIMEOUT_MS});` +
|
||||
's.on("connect",function(){say("LIVE")});' +
|
||||
's.on("error",function(e){say(e.code==="ECONNREFUSED"||e.code==="ENOENT"?"EXITED":"UNVERIFIABLE")})'
|
||||
|
||||
export function relaySocketOwnerProbeCommand(nodePath: string, sockPath: string): string {
|
||||
// Why: pass the socket path as argv[1] rather than interpolating it into -e, to dodge quoting issues.
|
||||
return `${shellEscape(nodePath)} -e ${shellEscape(RELAY_SOCKET_OWNER_PROBE_SCRIPT)} ${shellEscape(sockPath)}`
|
||||
}
|
||||
|
||||
export function parseRelaySocketOwnerProbe(output: string): RelaySocketOwnerVerdict {
|
||||
const markers = new Set(
|
||||
output
|
||||
.split(/\r?\n/)
|
||||
.map((line) => line.trim())
|
||||
.filter((line) => line === 'LIVE' || line === 'EXITED' || line === 'UNVERIFIABLE')
|
||||
)
|
||||
if (markers.size !== 1) {
|
||||
return 'unverifiable'
|
||||
}
|
||||
if (markers.has('LIVE')) {
|
||||
return 'live'
|
||||
}
|
||||
return markers.has('EXITED') ? 'exited' : 'unverifiable'
|
||||
}
|
||||
|
||||
export async function probeRelaySocketOwner(
|
||||
conn: SshConnection,
|
||||
nodePath: string,
|
||||
sockPath: string,
|
||||
signal?: AbortSignal
|
||||
): Promise<RelaySocketOwnerVerdict> {
|
||||
try {
|
||||
return parseRelaySocketOwnerProbe(
|
||||
await execCommand(conn, relaySocketOwnerProbeCommand(nodePath, sockPath), { signal })
|
||||
)
|
||||
} catch {
|
||||
// Why: a probe we could not run answers nothing about the owner, and the
|
||||
// caller must not treat that silence as death.
|
||||
signal?.throwIfAborted()
|
||||
return 'unverifiable'
|
||||
}
|
||||
}
|
||||
|
||||
export class RelaySocketOwnerLiveError extends Error {
|
||||
readonly name = 'RelaySocketOwnerLiveError'
|
||||
|
||||
constructor(
|
||||
readonly sockPath: string,
|
||||
readonly verdict: Exclude<RelaySocketOwnerVerdict, 'exited'>,
|
||||
readonly cause?: unknown
|
||||
) {
|
||||
super(
|
||||
`Could not reach the relay at ${sockPath}, and its socket is ${
|
||||
verdict === 'live' ? 'still owned by a running relay' : 'of unproven ownership'
|
||||
}. Leaving it in place — replacing it would strand that relay and every terminal it holds. ` +
|
||||
'Retry, or use "Reset remote relay" on the SSH target to stop it.'
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
export function isRelaySocketOwnerLiveError(err: unknown): err is RelaySocketOwnerLiveError {
|
||||
return err instanceof RelaySocketOwnerLiveError
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove a relay socket so a fresh daemon can bind the same path — but only
|
||||
* once the probe proves nothing is listening there.
|
||||
*
|
||||
* Why this gate: unlinking cannot stop a relay that already bound the path. It
|
||||
* only hides the live owner from the fresh launch's EADDRINUSE stale-socket
|
||||
* check, which is the one interlock that refuses to displace a running daemon.
|
||||
* A failed `--connect` is loss of contact, never proof of death (STA-1756).
|
||||
*/
|
||||
export async function removeUnownedRelaySocket(
|
||||
conn: SshConnection,
|
||||
nodePath: string,
|
||||
sockPath: string,
|
||||
options?: { signal?: AbortSignal; cause?: unknown }
|
||||
): Promise<void> {
|
||||
const signal = options?.signal
|
||||
const verdict = await probeRelaySocketOwner(conn, nodePath, sockPath, signal)
|
||||
signal?.throwIfAborted()
|
||||
if (verdict !== 'exited') {
|
||||
throw new RelaySocketOwnerLiveError(sockPath, verdict, options?.cause)
|
||||
}
|
||||
await execCommand(conn, `rm -f ${shellEscape(sockPath)}`, { signal }).catch((cleanupErr) => {
|
||||
if (isUnconfirmedSshCommandTermination(cleanupErr)) {
|
||||
throw cleanupErr
|
||||
}
|
||||
})
|
||||
signal?.throwIfAborted()
|
||||
}
|
||||
Reference in New Issue
Block a user