mirror of
https://github.com/stablyai/orca.git
synced 2026-10-19 08:03:13 +00:00
* fix(ssh): handle owner displacement and graceful shutdown SSH connections can reconnect with valid session proof after network loss or device sleep. When the incumbent owner is still half-open, allow the reconnecting client to displace it outright rather than wait for socket closure — a window that may never close. Retain displaced deliveries for the new owner to rotate. During app shutdown, drain SSH sessions without terminating recovery operations, and retry pending owner grants in case a replacement commits mid-drain. * fix(ssh): handle owner displacement and graceful shutdown Make QuitTeardownStartGate a shared singleton so SSH connects use the same shutdown fence as the main quit path. Track test-connection probes to ensure they complete before final teardown. Guard owner displacement to prevent stale owners from clearing recovery state claimed by newer owners. * fix(ssh): fix flaky test sync and add error code safety check Test was using tick-based Promise.resolve() loops which don't guarantee the async operation has started. Replace with signal-based synchronization that waits for the actual lease flush. Also add nullish-coalescing to error code check to prevent crashes if error is null or undefined. * fix(ssh): fence reset transport opens during shutdown * fix(ssh): keep recovery leases stable across reconnects * fix(ssh): close transports owned by cancelled connect attempts When a connect is cancelled after its transport has opened, that cancelled attempt still owns the transport and must close it — otherwise it leaks. Add disconnectConnection() to close by identity (not by target ID) so a cancelled attempt closes only the transport it minted, without tearing down its replacement's live transport. Track priorConnection to detect whether this attempt opened a new transport or reused an existing one, and close only on abandonment if this attempt owns the session. * fix(ssh): fence old owner proofs and close superseded transports When an owner reconnects with a new proof while an old one is still live, the old proof is now fenced with SUPERSEDED_ERROR instead of retrying indefinitely. The relay also closes stale transports to signal that their recovery generation has been overtaken by a newer one. This ensures overlapping reconnect scenarios complete with the newest proof rather than getting blocked by stale recovery attempts. * fix(relay): re-pin stdin/stdout fds after closing to prevent recycling When the relay closes stdin/stdout to signal EOF to the SSH peer, the OS can recycle those fds (0 and 1) for new sockets or files. If Node still treats process.stdin/stdout as those numbers, subsequent operations corrupt socket clients and trigger shutdown errors. Re-pin the fds by opening /dev/null to keep them occupied and prevent recycling.
121 lines
3.8 KiB
TypeScript
121 lines
3.8 KiB
TypeScript
import type { SshTarget, SshConnectionState } from '../../shared/ssh-types'
|
|
import { SshConnection, type SshConnectionCallbacks } from './ssh-connection'
|
|
|
|
// ── Connection Manager ──────────────────────────────────────────────
|
|
// Why: extracted from ssh-connection.ts to keep each file under the
|
|
// 300-line oxlint max-lines threshold while preserving a clear
|
|
// single-responsibility boundary (connection lifecycle vs. pool management).
|
|
|
|
export class SshConnectionManager {
|
|
private connections = new Map<string, SshConnection>()
|
|
private callbacks: SshConnectionCallbacks
|
|
// Why: attempt identity lets disconnect unblock a replacement without the
|
|
// cancelled attempt later clearing the replacement's state.
|
|
private connectingTargets = new Map<string, symbol>()
|
|
|
|
constructor(callbacks: SshConnectionCallbacks) {
|
|
this.callbacks = callbacks
|
|
}
|
|
|
|
setCallbacks(callbacks: SshConnectionCallbacks): void {
|
|
this.callbacks = callbacks
|
|
for (const connection of this.connections.values()) {
|
|
connection.setCallbacks(callbacks)
|
|
}
|
|
}
|
|
|
|
async connect(target: SshTarget): Promise<SshConnection> {
|
|
const existing = this.connections.get(target.id)
|
|
if (existing?.getState().status === 'connected') {
|
|
return existing
|
|
}
|
|
|
|
if (this.connectingTargets.has(target.id)) {
|
|
throw new Error(`Connection to ${target.label} is already in progress`)
|
|
}
|
|
|
|
const attempt = Symbol(target.id)
|
|
this.connectingTargets.set(target.id, attempt)
|
|
|
|
try {
|
|
if (existing) {
|
|
await existing.disconnect()
|
|
}
|
|
|
|
const conn = new SshConnection(target, this.callbacks)
|
|
this.connections.set(target.id, conn)
|
|
|
|
try {
|
|
await conn.connect()
|
|
} catch (err) {
|
|
if (this.connections.get(target.id) === conn) {
|
|
this.connections.delete(target.id)
|
|
}
|
|
throw err
|
|
}
|
|
|
|
return conn
|
|
} finally {
|
|
if (this.connectingTargets.get(target.id) === attempt) {
|
|
this.connectingTargets.delete(target.id)
|
|
}
|
|
}
|
|
}
|
|
|
|
async disconnect(targetId: string): Promise<void> {
|
|
// Why: disconnect invalidates the old attempt immediately so a reconnect
|
|
// need not wait for the cancelled socket's late completion.
|
|
this.connectingTargets.delete(targetId)
|
|
const conn = this.connections.get(targetId)
|
|
if (!conn) {
|
|
return
|
|
}
|
|
await conn.disconnect()
|
|
if (this.connections.get(targetId) === conn) {
|
|
this.connections.delete(targetId)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Close one specific connection, clearing the pool entry only when it is still the registered one.
|
|
* Why: a cancelled connect whose transport opened late owns that exact connection — disconnecting
|
|
* by target id would tear down the replacement's live transport instead.
|
|
*/
|
|
async disconnectConnection(targetId: string, conn: SshConnection): Promise<void> {
|
|
await conn.disconnect()
|
|
if (this.connections.get(targetId) === conn) {
|
|
this.connections.delete(targetId)
|
|
}
|
|
}
|
|
|
|
async reconnect(targetId: string): Promise<void> {
|
|
const conn = this.connections.get(targetId)
|
|
if (!conn) {
|
|
return
|
|
}
|
|
await conn.reconnect()
|
|
}
|
|
|
|
getConnection(targetId: string): SshConnection | undefined {
|
|
return this.connections.get(targetId)
|
|
}
|
|
|
|
getState(targetId: string): SshConnectionState | null {
|
|
return this.connections.get(targetId)?.getState() ?? null
|
|
}
|
|
|
|
getAllStates(): Map<string, SshConnectionState> {
|
|
const states = new Map<string, SshConnectionState>()
|
|
for (const [id, conn] of this.connections) {
|
|
states.set(id, conn.getState())
|
|
}
|
|
return states
|
|
}
|
|
|
|
async disconnectAll(): Promise<void> {
|
|
const disconnects = Array.from(this.connections.values()).map((c) => c.disconnect())
|
|
await Promise.allSettled(disconnects)
|
|
this.connections.clear()
|
|
}
|
|
}
|