mirror of
https://github.com/stablyai/orca.git
synced 2026-10-17 00:03:08 +00:00
fix(ssh): make session-limited (MaxSessions=1) relay deploys actually succeed
Review of #7180 verified the parent fallback end-to-end against a real MaxSessions=1 sshd and found the connect still failed. Four gaps, in order of discovery: - isSshSessionLimitError missed stock OpenSSH, which refuses session channels over MaxSessions with SSH2_OPEN_CONNECT_FAILED (2) and 'open failed' — reason 4 never matched, so the fallback never triggered. - execCommand settled aborted commands before the channel finished closing, so the sequential fallback reissued execs while sshd still counted the old session. - SshConnection.waitForSshCallback rejected aborts mid-channel-open immediately, leaking a confirmed-late channel that held the only session slot; it now settles after the late channel closes (bounded) and drains its streams so ssh2 emits 'close'. - Session channel opens now retry transient session-limit refusals (sshd frees the slot only after processing our close-ack, which the next open can beat by microseconds), and the remote orca CLI shim install is non-fatal like the managed-hook install — after the relay bridge occupies the sole session slot, raw-connection extras must degrade instead of failing the connection. Verified live against Docker sshd (OpenSSH 9.2, MaxSessions=1): fresh deploy (upload + native deps + launch), reconnect cycles, and a PTY round-trip all succeed; unrestricted-sshd regression run also passes. Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
@@ -22,6 +22,7 @@ type MockSshClient = {
|
||||
_sock: Socket | undefined
|
||||
lastExecCommand?: string
|
||||
lastConnectConfig?: unknown
|
||||
exec: (cmd: string, cb: (err: Error | undefined, channel: unknown) => void) => void
|
||||
}
|
||||
let clientInstances: MockSshClient[] = []
|
||||
|
||||
@@ -707,6 +708,84 @@ describe('SshConnection', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('retries a session-limit-refused exec open and succeeds on a later attempt', async () => {
|
||||
const conn = new SshConnection(createTarget(), createCallbacks())
|
||||
await conn.connect()
|
||||
const channel = { close: vi.fn() }
|
||||
const execMock = vi
|
||||
.fn<(cmd: string, cb: (err: Error | undefined, ch: unknown) => void) => void>()
|
||||
.mockImplementationOnce((_cmd, cb) => {
|
||||
cb(
|
||||
Object.assign(new Error('(SSH) Channel open failure: open failed'), { reason: 2 }),
|
||||
undefined
|
||||
)
|
||||
})
|
||||
.mockImplementation((_cmd, cb) => cb(undefined, channel))
|
||||
clientInstances[0].exec = execMock as never
|
||||
|
||||
await expect(conn.exec('printf ready')).resolves.toBe(channel)
|
||||
expect(execMock).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('surfaces the session-limit error once open retries are exhausted', async () => {
|
||||
const conn = new SshConnection(createTarget(), createCallbacks())
|
||||
await conn.connect()
|
||||
const refusal = Object.assign(new Error('(SSH) Channel open failure: open failed'), {
|
||||
reason: 2
|
||||
})
|
||||
const execMock = vi
|
||||
.fn<(cmd: string, cb: (err: Error | undefined, ch: unknown) => void) => void>()
|
||||
.mockImplementation((_cmd, cb) => cb(refusal, undefined))
|
||||
clientInstances[0].exec = execMock as never
|
||||
|
||||
await expect(conn.exec('printf ready')).rejects.toBe(refusal)
|
||||
expect(execMock).toHaveBeenCalledTimes(4)
|
||||
})
|
||||
|
||||
it('does not retry non-session-limit exec open failures', async () => {
|
||||
const conn = new SshConnection(createTarget(), createCallbacks())
|
||||
await conn.connect()
|
||||
const failure = new Error('Not connected')
|
||||
const execMock = vi
|
||||
.fn<(cmd: string, cb: (err: Error | undefined, ch: unknown) => void) => void>()
|
||||
.mockImplementation((_cmd, cb) => cb(failure, undefined))
|
||||
clientInstances[0].exec = execMock as never
|
||||
|
||||
await expect(conn.exec('printf ready')).rejects.toBe(failure)
|
||||
expect(execMock).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('settles an abort during channel open only after the late channel closes', async () => {
|
||||
const conn = new SshConnection(createTarget(), createCallbacks())
|
||||
await conn.connect()
|
||||
execBehavior = 'pending'
|
||||
const controller = new AbortController()
|
||||
const lateChannel = Object.assign(new EventEmitter(), {
|
||||
close: vi.fn(),
|
||||
resume: vi.fn(),
|
||||
stderr: { resume: vi.fn() }
|
||||
})
|
||||
|
||||
const outcomePromise = conn
|
||||
.exec('printf ready', { signal: controller.signal })
|
||||
.then(() => 'opened')
|
||||
.catch((error: Error) => error.name)
|
||||
|
||||
await Promise.resolve()
|
||||
controller.abort()
|
||||
pendingExecCallback?.(undefined, lateChannel)
|
||||
|
||||
// Why: the sshd session slot is freed only when the channel finishes
|
||||
// closing — settling before 'close' lets the next open race the close.
|
||||
const early = await Promise.race([outcomePromise, Promise.resolve('pending')])
|
||||
expect(early).toBe('pending')
|
||||
expect(lateChannel.close).toHaveBeenCalledTimes(1)
|
||||
expect(lateChannel.resume).toHaveBeenCalled()
|
||||
|
||||
lateChannel.emit('close')
|
||||
await expect(outcomePromise).resolves.toBe('AbortError')
|
||||
})
|
||||
|
||||
it('times out when ssh2 never opens an SFTP channel', async () => {
|
||||
const conn = new SshConnection(createTarget(), createCallbacks())
|
||||
await conn.connect()
|
||||
|
||||
+102
-16
@@ -35,12 +35,24 @@ import {
|
||||
type SshConnectionCallbacks
|
||||
} from './ssh-connection-utils'
|
||||
import type { RemoteHostPlatform } from './ssh-remote-platform'
|
||||
import { isSshSessionLimitError } from './ssh-session-limit-error'
|
||||
export type { SshConnectionCallbacks } from './ssh-connection-utils'
|
||||
|
||||
type SshRemoteFileOptions = {
|
||||
hostPlatform?: RemoteHostPlatform
|
||||
}
|
||||
|
||||
// Upper bound on waiting for an aborted late-opened channel to finish closing
|
||||
// before rejecting anyway. Normal closes complete in one network round-trip.
|
||||
const ABORTED_CHANNEL_CLOSE_GRACE_MS = 5_000
|
||||
|
||||
// Why: on session-limited servers (MaxSessions), a channel open can be refused
|
||||
// transiently — e.g. our next CHANNEL_OPEN reaching sshd a few microseconds
|
||||
// before it finishes processing the previous channel's close. A refused open
|
||||
// never started the command, so retrying is always safe.
|
||||
const SESSION_LIMIT_OPEN_RETRIES = 4
|
||||
const SESSION_LIMIT_OPEN_RETRY_DELAY_MS = 150
|
||||
|
||||
function cloneResolvedConfig(config: SshResolvedConfig | null): SshResolvedConfig | null {
|
||||
if (!config) {
|
||||
return null
|
||||
@@ -132,10 +144,14 @@ export class SshConnection {
|
||||
}
|
||||
const client = this.client
|
||||
const remoteCommand = options?.wrapCommand === false ? cmd : wrapRemoteCommandForPosixShell(cmd)
|
||||
return this.waitForSshCallback(
|
||||
'SSH exec channel timed out',
|
||||
(callback) => client.exec(remoteCommand, callback),
|
||||
(channel) => channel.close(),
|
||||
return this.openSessionChannelWithRetry(
|
||||
() =>
|
||||
this.waitForSshCallback(
|
||||
'SSH exec channel timed out',
|
||||
(callback) => client.exec(remoteCommand, callback),
|
||||
(channel) => channel.close(),
|
||||
options?.signal
|
||||
),
|
||||
options?.signal
|
||||
)
|
||||
}
|
||||
@@ -148,13 +164,39 @@ export class SshConnection {
|
||||
throw new Error('Not connected')
|
||||
}
|
||||
const client = this.client
|
||||
return this.waitForSshCallback(
|
||||
'SSH SFTP channel timed out',
|
||||
(callback) => client.sftp(callback),
|
||||
(sftp) => sftp.end()
|
||||
return this.openSessionChannelWithRetry(() =>
|
||||
this.waitForSshCallback(
|
||||
'SSH SFTP channel timed out',
|
||||
(callback) => client.sftp(callback),
|
||||
(sftp) => sftp.end()
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
private async openSessionChannelWithRetry<T>(
|
||||
open: () => Promise<T>,
|
||||
signal?: AbortSignal
|
||||
): Promise<T> {
|
||||
let lastError: unknown
|
||||
for (let attempt = 0; attempt < SESSION_LIMIT_OPEN_RETRIES; attempt++) {
|
||||
if (attempt > 0) {
|
||||
await new Promise((resolve) => setTimeout(resolve, SESSION_LIMIT_OPEN_RETRY_DELAY_MS))
|
||||
if (signal?.aborted) {
|
||||
throw createSshOperationAbortError()
|
||||
}
|
||||
}
|
||||
try {
|
||||
return await open()
|
||||
} catch (err) {
|
||||
if (!isSshSessionLimitError(err)) {
|
||||
throw err
|
||||
}
|
||||
lastError = err
|
||||
}
|
||||
}
|
||||
throw lastError
|
||||
}
|
||||
|
||||
private waitForSshCallback<T>(
|
||||
timeoutMessage: string,
|
||||
register: (callback: (error: Error | undefined, value: T) => void) => void,
|
||||
@@ -163,23 +205,57 @@ export class SshConnection {
|
||||
): Promise<T> {
|
||||
return new Promise((resolve, reject) => {
|
||||
let settled = false
|
||||
// Why: rejecting the instant the signal aborts lets the caller proceed
|
||||
// while the in-flight channel open completes in the background and holds
|
||||
// a server-side session slot (MaxSessions). Mark the abort and settle
|
||||
// from the open callback, after the late resource has been closed.
|
||||
let abortRequested = false
|
||||
const cleanup = (): void => {
|
||||
clearTimeout(timer)
|
||||
signal?.removeEventListener('abort', onAbort)
|
||||
}
|
||||
const onAbort = (): void => {
|
||||
if (settled) {
|
||||
return
|
||||
}
|
||||
settled = true
|
||||
cleanup()
|
||||
reject(createSshOperationAbortError())
|
||||
abortRequested = true
|
||||
}
|
||||
const timer = setTimeout(() => {
|
||||
settled = true
|
||||
signal?.removeEventListener('abort', onAbort)
|
||||
reject(new Error(timeoutMessage))
|
||||
reject(abortRequested ? createSshOperationAbortError() : new Error(timeoutMessage))
|
||||
}, CONNECT_TIMEOUT_MS)
|
||||
const rejectAfterClose = (value: T): void => {
|
||||
const abortError = createSshOperationAbortError()
|
||||
const emitter = value as Partial<NodeJS.EventEmitter> & {
|
||||
resume?: () => void
|
||||
stderr?: { resume?: () => void }
|
||||
}
|
||||
let finished = false
|
||||
const done = (): void => {
|
||||
if (finished) {
|
||||
return
|
||||
}
|
||||
finished = true
|
||||
clearTimeout(closeGraceTimer)
|
||||
reject(abortError)
|
||||
}
|
||||
// Why: bounded — a remote that never confirms the close must not hang
|
||||
// the aborted operation forever.
|
||||
const closeGraceTimer = setTimeout(done, ABORTED_CHANNEL_CLOSE_GRACE_MS)
|
||||
if (typeof emitter.once === 'function') {
|
||||
emitter.once('close', done)
|
||||
}
|
||||
// Why: ssh2 withholds the 'close' event until the channel's streams
|
||||
// are drained; nobody else will ever read this discarded channel.
|
||||
emitter.resume?.()
|
||||
emitter.stderr?.resume?.()
|
||||
try {
|
||||
cleanupLateValue?.(value)
|
||||
} catch {
|
||||
/* best effort */
|
||||
}
|
||||
if (typeof emitter.once !== 'function') {
|
||||
done()
|
||||
}
|
||||
}
|
||||
const finish = (error: Error | undefined, value?: T): void => {
|
||||
if (settled) {
|
||||
// Why: ssh2 can invoke the open callback after our timeout has
|
||||
@@ -196,6 +272,14 @@ export class SshConnection {
|
||||
}
|
||||
settled = true
|
||||
cleanup()
|
||||
if (abortRequested) {
|
||||
if (!error && value !== undefined) {
|
||||
rejectAfterClose(value)
|
||||
} else {
|
||||
reject(createSshOperationAbortError())
|
||||
}
|
||||
return
|
||||
}
|
||||
if (error) {
|
||||
reject(error)
|
||||
return
|
||||
@@ -203,7 +287,9 @@ export class SshConnection {
|
||||
resolve(value as T)
|
||||
}
|
||||
if (signal?.aborted) {
|
||||
onAbort()
|
||||
// No open is in flight yet, so failing fast leaks nothing.
|
||||
cleanup()
|
||||
reject(createSshOperationAbortError())
|
||||
return
|
||||
}
|
||||
signal?.addEventListener('abort', onAbort, { once: true })
|
||||
|
||||
@@ -230,8 +230,22 @@ describe('execCommand', () => {
|
||||
await Promise.resolve()
|
||||
controller.abort()
|
||||
|
||||
await expect(commandPromise).rejects.toMatchObject({ name: 'AbortError' })
|
||||
// Why: sshd frees the MaxSessions slot only when the channel finishes
|
||||
// closing, so abort must request close and settle from the 'close' event —
|
||||
// settling immediately lets the sequential fallback race the closing
|
||||
// channel and get refused.
|
||||
expect(channel.close).toHaveBeenCalledOnce()
|
||||
const settledEarly = await Promise.race([
|
||||
commandPromise.then(
|
||||
() => 'settled',
|
||||
() => 'settled'
|
||||
),
|
||||
Promise.resolve('pending')
|
||||
])
|
||||
expect(settledEarly).toBe('pending')
|
||||
channel.emit('close', 0)
|
||||
|
||||
await expect(commandPromise).rejects.toMatchObject({ name: 'AbortError' })
|
||||
expect(channel.listenerCount('error')).toBe(0)
|
||||
expect(channel.listenerCount('data')).toBe(0)
|
||||
expect(channel.listenerCount('close')).toBe(0)
|
||||
@@ -257,8 +271,13 @@ describe('execCommand', () => {
|
||||
controller.abort()
|
||||
resolveExec(channel)
|
||||
|
||||
await expect(commandPromise).rejects.toMatchObject({ name: 'AbortError' })
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
expect(channel.close).toHaveBeenCalledOnce()
|
||||
// Nonzero exit from the killed command must still surface as AbortError.
|
||||
channel.emit('close', 1)
|
||||
|
||||
await expect(commandPromise).rejects.toMatchObject({ name: 'AbortError' })
|
||||
expect(channel.listenerCount('error')).toBe(0)
|
||||
expect(channel.listenerCount('data')).toBe(0)
|
||||
expect(channel.listenerCount('close')).toBe(0)
|
||||
|
||||
@@ -275,11 +275,16 @@ export async function execCommand(
|
||||
fn(val as never)
|
||||
}
|
||||
const fail = (err: Error): void => {
|
||||
settle(reject, err)
|
||||
settle(reject, abortRequested ? createSshOperationAbortError() : err)
|
||||
}
|
||||
// Why: sshd counts the session against MaxSessions until CHANNEL_CLOSE
|
||||
// completes. Settling on abort before the channel actually closes lets the
|
||||
// concurrent-bootstrap sequential fallback reissue an exec while the slot
|
||||
// is still held, so it gets refused again. Close and settle from onClose.
|
||||
let abortRequested = false
|
||||
const onAbort = (): void => {
|
||||
abortRequested = true
|
||||
channel.close()
|
||||
settle(reject, createSshOperationAbortError())
|
||||
}
|
||||
const onStdoutData = (data: Buffer): void => {
|
||||
stdout += data.toString('utf-8')
|
||||
@@ -288,7 +293,9 @@ export async function execCommand(
|
||||
stderr += data.toString('utf-8')
|
||||
}
|
||||
const onClose = (code: number): void => {
|
||||
if (code !== 0) {
|
||||
if (abortRequested) {
|
||||
settle(reject, createSshOperationAbortError())
|
||||
} else if (code !== 0) {
|
||||
const output = stderr.trim() || stdout.trim()
|
||||
settle(reject, new Error(`Command "${command}" failed (exit ${code}): ${output}`))
|
||||
} else {
|
||||
@@ -297,7 +304,12 @@ export async function execCommand(
|
||||
}
|
||||
const timeout = setTimeout(() => {
|
||||
channel.close()
|
||||
settle(reject, new Error(`Command "${command}" timed out after ${timeoutMs / 1000}s`))
|
||||
settle(
|
||||
reject,
|
||||
abortRequested
|
||||
? createSshOperationAbortError()
|
||||
: new Error(`Command "${command}" timed out after ${timeoutMs / 1000}s`)
|
||||
)
|
||||
}, timeoutMs)
|
||||
|
||||
// Why: remote reboot tears down exec channels with stream errors. Without
|
||||
|
||||
@@ -531,7 +531,19 @@ export class SshRelaySession {
|
||||
return false
|
||||
}
|
||||
|
||||
await this.installRemoteOrcaCliShim()
|
||||
try {
|
||||
await this.installRemoteOrcaCliShim()
|
||||
} catch (error) {
|
||||
// Why: the remote `orca` CLI shim is a convenience bridge. On session-
|
||||
// limited remotes (MaxSessions=1) the relay bridge holds the only slot,
|
||||
// so this raw-connection install can fail — that must not fail the
|
||||
// whole connection, matching the managed-hook install above.
|
||||
console.warn(
|
||||
`[ssh-relay-session] remote orca CLI shim install failed for ${this.targetId}: ${
|
||||
error instanceof Error ? error.message : String(error)
|
||||
}`
|
||||
)
|
||||
}
|
||||
if (shouldContinue && !shouldContinue()) {
|
||||
return false
|
||||
}
|
||||
|
||||
@@ -10,6 +10,16 @@ describe('isSshSessionLimitError', () => {
|
||||
).toBe(true)
|
||||
})
|
||||
|
||||
it('matches the OpenSSH MaxSessions rejection (SSH2_OPEN_CONNECT_FAILED + "open failed")', () => {
|
||||
// Why: stock OpenSSH sshd refuses session channels over MaxSessions with
|
||||
// reason 2, not resource-shortage — observed against OpenSSH 9.2.
|
||||
expect(
|
||||
isSshSessionLimitError(
|
||||
Object.assign(new Error('(SSH) Channel open failure: open failed'), { reason: 2 })
|
||||
)
|
||||
).toBe(true)
|
||||
})
|
||||
|
||||
it('matches OpenSSH mux and MaxSessions failures', () => {
|
||||
expect(
|
||||
isSshSessionLimitError(
|
||||
@@ -29,11 +39,6 @@ describe('isSshSessionLimitError', () => {
|
||||
Object.assign(new Error('(SSH) Channel open failure: open failed'), { reason: 1 })
|
||||
)
|
||||
).toBe(false)
|
||||
expect(
|
||||
isSshSessionLimitError(
|
||||
Object.assign(new Error('(SSH) Channel open failure: open failed'), { reason: 2 })
|
||||
)
|
||||
).toBe(false)
|
||||
expect(
|
||||
isSshSessionLimitError(
|
||||
Object.assign(new Error('(SSH) Channel open failure: open failed'), { reason: 3 })
|
||||
|
||||
@@ -4,8 +4,11 @@ export function isSshSessionLimitError(err: unknown): boolean {
|
||||
}
|
||||
const reason = (err as { reason?: unknown }).reason
|
||||
const message = err.message.toLowerCase()
|
||||
// Why: OpenSSH rejects session channels over MaxSessions with
|
||||
// SSH2_OPEN_CONNECT_FAILED (2) and the literal description "open failed";
|
||||
// reason 4 (resource shortage) covers other server implementations.
|
||||
if (
|
||||
reason === 4 &&
|
||||
(reason === 2 || reason === 4) &&
|
||||
(message.includes('channel open failure') || message.includes('open failed'))
|
||||
) {
|
||||
return true
|
||||
|
||||
Reference in New Issue
Block a user