mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
feat(wsl): connect to a WSL distro's orcad over wsl.exe stdio only (WSL-via-orcad C5) (#27834)
This commit is contained in:
@@ -24,20 +24,22 @@ import type {
|
||||
StartedPortForward
|
||||
} from './ssh-port-forward-provider'
|
||||
|
||||
type OrcadStdioBridgeDependencies = {
|
||||
export type OrcadStdioBridgeDependencies = {
|
||||
resolveBridge: (conn: OrcadHostTransport, port: number) => Promise<OrcadStdioBridge>
|
||||
checkBridge: (conn: OrcadHostTransport, bridge: OrcadStdioBridge) => Promise<unknown>
|
||||
openChannel: (conn: OrcadHostTransport, bridge: OrcadStdioBridge) => Promise<RemoteCommandChannel>
|
||||
}
|
||||
|
||||
const defaultDependencies: OrcadStdioBridgeDependencies = {
|
||||
export const ORCAD_STDIO_BRIDGE_DEPENDENCIES: OrcadStdioBridgeDependencies = {
|
||||
resolveBridge: resolveOrcadStdioBridge,
|
||||
checkBridge: checkOrcadStdioBridge,
|
||||
openChannel: openOrcadStdioBridgeChannel
|
||||
}
|
||||
|
||||
export class OrcadStdioBridgePortForwardProvider implements SshPortForwardProvider {
|
||||
constructor(private readonly dependencies: OrcadStdioBridgeDependencies = defaultDependencies) {}
|
||||
constructor(
|
||||
private readonly dependencies: OrcadStdioBridgeDependencies = ORCAD_STDIO_BRIDGE_DEPENDENCIES
|
||||
) {}
|
||||
|
||||
// Why not system SSH: each bridge socket would be its own `ssh` process (a fresh login, and
|
||||
// perhaps a key prompt, when multiplexing is off), and its refusal can't be probed up front.
|
||||
@@ -50,7 +52,8 @@ export class OrcadStdioBridgePortForwardProvider implements SshPortForwardProvid
|
||||
}
|
||||
}
|
||||
|
||||
async function startOrcadStdioBridge(
|
||||
/** Transport-agnostic: any orcad host whose exec can run the bridge (SSH, a WSL distro). */
|
||||
export async function startOrcadStdioBridge(
|
||||
host: OrcadHostTransport,
|
||||
options: PortForwardStartOptions,
|
||||
deps: OrcadStdioBridgeDependencies
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
/**
|
||||
* Test-only: a `wsl.exe` spawner whose children run the stdio bridge in-process. Each child dials
|
||||
* the port the bridge command names on this machine and prints the real sentinel, so the listener
|
||||
* and WSL transport above it run unchanged.
|
||||
*/
|
||||
import { EventEmitter } from 'node:events'
|
||||
import { connect } from 'node:net'
|
||||
import { PassThrough } from 'node:stream'
|
||||
import type { PipedChildProcess, PipedProcessSpawner } from '@orca/process-host/process-spec'
|
||||
import {
|
||||
ORCAD_STDIO_BRIDGE_READY,
|
||||
ORCAD_STDIO_BRIDGE_REFUSED,
|
||||
orcadPosixStdioBridgeCommand
|
||||
} from '../ssh/orcad-stdio-bridge-script'
|
||||
import type { OrcadStdioBridge } from '../ssh/orcad-stdio-bridge'
|
||||
|
||||
export type FakeWslBridgeChildren = {
|
||||
spawn: PipedProcessSpawner
|
||||
/** Every spawned child's guest command, in order. */
|
||||
commands: string[]
|
||||
live: () => number
|
||||
maxLive: () => number
|
||||
}
|
||||
|
||||
/** The bridge the WSL path resolves for a Linux distro, without the platform probe. */
|
||||
export function posixOrcadStdioBridge(port: number): OrcadStdioBridge {
|
||||
return { command: orcadPosixStdioBridgeCommand(port), mode: 'raw', wrapCommand: true }
|
||||
}
|
||||
|
||||
export function fakeWslBridgeChildren(): FakeWslBridgeChildren {
|
||||
const commands: string[] = []
|
||||
let live = 0
|
||||
let maxLive = 0
|
||||
const spawn: PipedProcessSpawner = (spec) => {
|
||||
const command = spec.args?.at(-1) ?? ''
|
||||
commands.push(command)
|
||||
live += 1
|
||||
maxLive = Math.max(maxLive, live)
|
||||
const exitState: { exitCode: number | null; signalCode: NodeJS.Signals | null } = {
|
||||
exitCode: null,
|
||||
signalCode: null
|
||||
}
|
||||
const child = Object.assign(new EventEmitter(), exitState, {
|
||||
stdin: new PassThrough(),
|
||||
stdout: new PassThrough(),
|
||||
stderr: new PassThrough(),
|
||||
kill: () => {
|
||||
socket.destroy()
|
||||
finish(null, 'SIGTERM')
|
||||
return true
|
||||
}
|
||||
})
|
||||
let finished = false
|
||||
const finish = (code: number | null, signal: NodeJS.Signals | null = null): void => {
|
||||
if (finished) {
|
||||
return
|
||||
}
|
||||
finished = true
|
||||
live -= 1
|
||||
child.exitCode = code
|
||||
child.signalCode = signal
|
||||
child.emit('exit', code, signal)
|
||||
child.stdout.end()
|
||||
child.stderr.end()
|
||||
setImmediate(() => child.emit('close', code, signal))
|
||||
}
|
||||
const port = Number(/ (\d+); fi; done;/.exec(command)?.[1] ?? 0)
|
||||
const socket = connect({ host: '127.0.0.1', port })
|
||||
let connected = false
|
||||
socket.on('connect', () => {
|
||||
connected = true
|
||||
child.stdout.write(`${ORCAD_STDIO_BRIDGE_READY}\n`)
|
||||
child.stdin.pipe(socket)
|
||||
socket.pipe(child.stdout, { end: false })
|
||||
})
|
||||
socket.on('error', (error: NodeJS.ErrnoException) => {
|
||||
if (!connected) {
|
||||
child.stdout.write(`${ORCAD_STDIO_BRIDGE_REFUSED} ${error.code ?? 'ERROR'}\n`)
|
||||
}
|
||||
finish(0)
|
||||
})
|
||||
socket.on('close', () => finish(0))
|
||||
process.nextTick(() => child.emit('spawn'))
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the WSL channel reads only the streams, exit state, kill and events assigned above.
|
||||
return child as unknown as PipedChildProcess
|
||||
}
|
||||
return { spawn, commands, live: () => live, maxLive: () => maxLive }
|
||||
}
|
||||
@@ -0,0 +1,260 @@
|
||||
import { connect, type Server, type Socket } from 'node:net'
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from 'vitest'
|
||||
import { encodePairingOffer, parsePairingCode, PAIRING_OFFER_VERSION } from '../../shared/pairing'
|
||||
import type { ServeReadiness } from '../server/serve-readiness'
|
||||
import { OrcadManagedIdentityError } from '../ssh/orcad-managed-tunnel-identity'
|
||||
import { checkOrcadStdioBridge, openOrcadStdioBridgeChannel } from '../ssh/orcad-stdio-bridge'
|
||||
import { startEchoServer } from '../ssh/orcad-stdio-bridge-test-channel'
|
||||
import { fakeWslBridgeChildren, posixOrcadStdioBridge } from './wsl-orcad-bridge-test-child'
|
||||
import {
|
||||
WslOrcadConnectionManager,
|
||||
WslOrcadConnectionSupersededError,
|
||||
WslOrcadLaunchDeferredError,
|
||||
type WslOrcadConnection,
|
||||
type WslOrcadConnectionDependencies
|
||||
} from './wsl-orcad-connection'
|
||||
import type { WslOrcadLaunch } from './wsl-orcad-deploy'
|
||||
import { createWslOrcadHostTransport } from './wsl-orcad-host-transport'
|
||||
import { listenWslOrcadStdioBridge } from './wsl-orcad-stdio-bridge'
|
||||
|
||||
let first: { server: Server; port: number }
|
||||
let second: { server: Server; port: number }
|
||||
const managers: WslOrcadConnectionManager[] = []
|
||||
const clients: Socket[] = []
|
||||
|
||||
beforeAll(async () => {
|
||||
first = await startEchoServer()
|
||||
second = await startEchoServer()
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
for (const client of clients.splice(0)) {
|
||||
client.destroy()
|
||||
}
|
||||
await Promise.all(managers.splice(0).map((manager) => manager.dispose()))
|
||||
})
|
||||
|
||||
afterAll(() => {
|
||||
first.server.close()
|
||||
second.server.close()
|
||||
})
|
||||
|
||||
function readiness(port: number, deviceToken = 'device-token'): ServeReadiness {
|
||||
const endpoint = `ws://127.0.0.1:${port}`
|
||||
return {
|
||||
runtimeId: 'runtime-1',
|
||||
boundEndpoint: endpoint,
|
||||
advertisedEndpoint: null,
|
||||
managedWslCliReconciliation: 'settled',
|
||||
pairing: {
|
||||
available: true,
|
||||
url: encodePairingOffer({
|
||||
v: PAIRING_OFFER_VERSION,
|
||||
endpoint,
|
||||
deviceToken,
|
||||
publicKeyB64: 'public-key',
|
||||
pairedDeviceId: 'device-1'
|
||||
}),
|
||||
endpoint,
|
||||
deviceId: 'device-1',
|
||||
webClientUrl: null,
|
||||
scope: 'runtime',
|
||||
qr: null
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function ready(port: number, deviceToken?: string): WslOrcadLaunch {
|
||||
return {
|
||||
outcome: 'ready',
|
||||
distro: 'Ubuntu',
|
||||
distroId: null,
|
||||
port,
|
||||
fullVersion: '1.0.0',
|
||||
pid: 42,
|
||||
readiness: readiness(port, deviceToken),
|
||||
holder: 'adopted'
|
||||
}
|
||||
}
|
||||
|
||||
function setup(overrides: Partial<WslOrcadConnectionDependencies> = {}) {
|
||||
const children = fakeWslBridgeChildren()
|
||||
const launches: WslOrcadLaunch[] = []
|
||||
const deployAndLaunch = vi.fn(async () => launches.shift() ?? ready(first.port))
|
||||
const onChange = vi.fn()
|
||||
const manager = new WslOrcadConnectionManager({
|
||||
deployAndLaunch,
|
||||
createTransport: (target) => createWslOrcadHostTransport({ target, spawn: children.spawn }),
|
||||
listen: (transport, options) =>
|
||||
listenWslOrcadStdioBridge(transport, options, {
|
||||
resolveBridge: async (_conn, port) => posixOrcadStdioBridge(port),
|
||||
checkBridge: checkOrcadStdioBridge,
|
||||
openChannel: openOrcadStdioBridgeChannel
|
||||
}),
|
||||
verifyIdentity: async () => ({ verdict: 'verified' }),
|
||||
onChange,
|
||||
...overrides
|
||||
})
|
||||
managers.push(manager)
|
||||
return { manager, children, launches, deployAndLaunch, onChange }
|
||||
}
|
||||
|
||||
function dial(port: number): Promise<Socket> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const socket = connect(port, '127.0.0.1', () => resolve(socket))
|
||||
socket.once('error', reject)
|
||||
clients.push(socket)
|
||||
})
|
||||
}
|
||||
|
||||
function echoOf(socket: Socket, message: string): Promise<string> {
|
||||
return new Promise((resolve) => {
|
||||
let received = ''
|
||||
socket.on('data', (chunk: Buffer) => {
|
||||
received += chunk.toString('utf8')
|
||||
if (received.length >= message.length) {
|
||||
resolve(received)
|
||||
}
|
||||
})
|
||||
socket.write(message)
|
||||
})
|
||||
}
|
||||
|
||||
async function roundTrip(connection: WslOrcadConnection, message: string): Promise<string> {
|
||||
return echoOf(await dial(connection.localPort), message)
|
||||
}
|
||||
|
||||
async function until(condition: () => boolean): Promise<void> {
|
||||
for (let i = 0; i < 200 && !condition(); i++) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 5))
|
||||
}
|
||||
expect(condition()).toBe(true)
|
||||
}
|
||||
|
||||
describe('WslOrcadConnectionManager', () => {
|
||||
it('rewrites the pairing offer to the local bridge port and nothing else', async () => {
|
||||
const { manager, onChange } = setup()
|
||||
const connection = await manager.ensure('Ubuntu')
|
||||
|
||||
expect(connection).toMatchObject({
|
||||
distro: 'Ubuntu',
|
||||
environmentId: 'wsl:ubuntu',
|
||||
remotePort: first.port,
|
||||
fullVersion: '1.0.0'
|
||||
})
|
||||
expect(connection.localPort).not.toBe(first.port)
|
||||
expect(parsePairingCode(connection.pairingOffer)).toEqual({
|
||||
v: PAIRING_OFFER_VERSION,
|
||||
endpoint: `ws://127.0.0.1:${connection.localPort}/`,
|
||||
deviceToken: 'device-token',
|
||||
publicKeyB64: 'public-key',
|
||||
pairedDeviceId: 'device-1'
|
||||
})
|
||||
expect(onChange).toHaveBeenCalledWith('Ubuntu', connection)
|
||||
expect(await roundTrip(connection, 'hello')).toBe('hello')
|
||||
})
|
||||
|
||||
it('shares one deploy and one bridge between concurrent connects to a distro', async () => {
|
||||
const { manager, deployAndLaunch, children } = setup()
|
||||
const [a, b] = await Promise.all([manager.ensure('Ubuntu'), manager.ensure('ubuntu ')])
|
||||
expect(a).toBe(b)
|
||||
expect(deployAndLaunch).toHaveBeenCalledTimes(1)
|
||||
expect(manager.list()).toEqual([a])
|
||||
expect(manager.get('UBUNTU')).toBe(a)
|
||||
// Only the bridge check ran a child so far.
|
||||
expect(children.commands).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('keeps the bridge when orcad still listens on the same port', async () => {
|
||||
const { manager, children, launches, onChange } = setup()
|
||||
const connection = await manager.ensure('Ubuntu')
|
||||
launches.push(ready(first.port))
|
||||
expect(await manager.ensure('Ubuntu')).toBe(connection)
|
||||
expect(children.commands).toHaveLength(1)
|
||||
expect(onChange).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('reports a changed pairing offer without rebuilding the bridge', async () => {
|
||||
const { manager, launches, onChange } = setup()
|
||||
const connection = await manager.ensure('Ubuntu')
|
||||
launches.push(ready(first.port, 'rotated-token'))
|
||||
const refreshed = await manager.ensure('Ubuntu')
|
||||
expect(refreshed.localPort).toBe(connection.localPort)
|
||||
expect(parsePairingCode(refreshed.pairingOffer)?.deviceToken).toBe('rotated-token')
|
||||
expect(onChange).toHaveBeenLastCalledWith('Ubuntu', refreshed)
|
||||
})
|
||||
|
||||
it('moves the bridge to the new port after a relaunch, on the same local port', async () => {
|
||||
const { manager, launches, children } = setup()
|
||||
const before = await manager.ensure('Ubuntu')
|
||||
const stale = await dial(before.localPort)
|
||||
expect(await echoOf(stale, 'old')).toBe('old')
|
||||
const staleClosed = new Promise((resolve) => stale.once('close', resolve))
|
||||
|
||||
launches.push(ready(second.port))
|
||||
const after = await manager.ensure('Ubuntu')
|
||||
expect(after).not.toBe(before)
|
||||
expect(after.remotePort).toBe(second.port)
|
||||
expect(after.localPort).toBe(before.localPort)
|
||||
expect(after.pairingOffer).toBe(before.pairingOffer)
|
||||
await staleClosed
|
||||
expect(await roundTrip(after, 'new')).toBe('new')
|
||||
expect(children.commands.at(-1)).toContain(` ${second.port}; fi; done;`)
|
||||
})
|
||||
|
||||
it('throws the deferral and starts no bridge', async () => {
|
||||
const { manager, launches, children } = setup()
|
||||
launches.push({
|
||||
outcome: 'deferred',
|
||||
distro: 'Ubuntu',
|
||||
candidateVersion: '2.0.0',
|
||||
code: 'live_terminals',
|
||||
reason: 'Terminals are running.',
|
||||
forceable: true
|
||||
})
|
||||
await expect(manager.ensure('Ubuntu')).rejects.toBeInstanceOf(WslOrcadLaunchDeferredError)
|
||||
expect(children.commands).toHaveLength(0)
|
||||
expect(manager.get('Ubuntu')).toBeNull()
|
||||
})
|
||||
|
||||
it('refuses a foreign server behind the port and closes its bridge', async () => {
|
||||
const { manager, children } = setup({
|
||||
verifyIdentity: async () => ({ verdict: 'foreign', detail: 'another runtime' })
|
||||
})
|
||||
await expect(manager.ensure('Ubuntu')).rejects.toBeInstanceOf(OrcadManagedIdentityError)
|
||||
expect(manager.get('Ubuntu')).toBeNull()
|
||||
await until(() => children.live() === 0)
|
||||
})
|
||||
|
||||
it('drops a connect that a close overtook', async () => {
|
||||
let release: () => void = () => {}
|
||||
const gate = new Promise<void>((resolve) => {
|
||||
release = resolve
|
||||
})
|
||||
const { manager } = setup({
|
||||
deployAndLaunch: async () => {
|
||||
await gate
|
||||
return ready(first.port)
|
||||
}
|
||||
})
|
||||
const pending = manager.ensure('Ubuntu')
|
||||
await manager.close('Ubuntu')
|
||||
release()
|
||||
await expect(pending).rejects.toBeInstanceOf(WslOrcadConnectionSupersededError)
|
||||
expect(manager.get('Ubuntu')).toBeNull()
|
||||
})
|
||||
|
||||
it('tears down every bridge child and the listener on close', async () => {
|
||||
const { manager, children, onChange } = setup()
|
||||
const connection = await manager.ensure('Ubuntu')
|
||||
const sockets = await Promise.all([1, 2, 3].map(() => dial(connection.localPort)))
|
||||
await Promise.all(sockets.map((socket, i) => echoOf(socket, `s${i}`)))
|
||||
expect(children.live()).toBe(3)
|
||||
|
||||
await connection.close()
|
||||
await until(() => children.live() === 0)
|
||||
expect(manager.get('Ubuntu')).toBeNull()
|
||||
expect(onChange).toHaveBeenLastCalledWith('Ubuntu', null)
|
||||
await expect(dial(connection.localPort)).rejects.toThrow()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,295 @@
|
||||
/**
|
||||
* The desktop's connection to one WSL distro's orcad: deploy and launch (C4), the `wsl.exe` stdio
|
||||
* bridge listener, and orcad's pairing offer rewritten to that listener's loopback port. One per
|
||||
* distro; a relaunch that moved orcad's port moves the bridge with it. In memory only: the
|
||||
* deployment record (C7) persists what `onChange` reports.
|
||||
*/
|
||||
import { parsePairingCode } from '../../shared/pairing'
|
||||
import { pairedRuntimeConnectionFromOffer } from '../../shared/paired-runtime-connection'
|
||||
import type { ServeReadiness } from '../server/serve-readiness'
|
||||
import type { OrcadHostTransport } from '../ssh/orcad-host-transport'
|
||||
import {
|
||||
OrcadManagedIdentityError,
|
||||
verifyOrcadPairingIdentity,
|
||||
type OrcadTunnelIdentity
|
||||
} from '../ssh/orcad-managed-tunnel-identity'
|
||||
import { tunneledOrcadPairingCode } from '../ssh/orcad-tunneled-pairing'
|
||||
import type { StartedPortForward } from '../ssh/ssh-port-forward-provider'
|
||||
import type { WslDistroHolder } from './wsl-distro-holder'
|
||||
import { wslDistroHolderKey } from './wsl-distro-holder-record'
|
||||
import {
|
||||
defaultWslOrcadEnvironmentId,
|
||||
deployAndLaunchWslOrcad,
|
||||
type WslOrcadDeployOptions,
|
||||
type WslOrcadLaunch
|
||||
} from './wsl-orcad-deploy'
|
||||
import type { WslOrcadHostTarget } from './wsl-orcad-host-target'
|
||||
import { createWslOrcadHostTransport } from './wsl-orcad-host-transport'
|
||||
import {
|
||||
listenWslOrcadStdioBridge,
|
||||
type WslOrcadStdioBridgeOptions
|
||||
} from './wsl-orcad-stdio-bridge'
|
||||
|
||||
export type WslOrcadConnection = {
|
||||
distro: string
|
||||
environmentId: string
|
||||
/** The loopback port on this machine the pairing offer points at. */
|
||||
localPort: number
|
||||
/** orcad's port inside the distro. */
|
||||
remotePort: number
|
||||
fullVersion: string
|
||||
/** orcad's pairing code, with its endpoint rewritten to `ws://127.0.0.1:<localPort>`. */
|
||||
pairingOffer: string
|
||||
close(): Promise<void>
|
||||
}
|
||||
|
||||
type WslOrcadDeferredLaunch = Extract<WslOrcadLaunch, { outcome: 'deferred' }>
|
||||
|
||||
/** orcad installed an update it may not activate yet (live terminals); nothing is listening. */
|
||||
export class WslOrcadLaunchDeferredError extends Error {
|
||||
constructor(readonly launch: WslOrcadDeferredLaunch) {
|
||||
super(launch.reason)
|
||||
this.name = 'WslOrcadLaunchDeferredError'
|
||||
}
|
||||
}
|
||||
|
||||
export class WslOrcadConnectionSupersededError extends Error {
|
||||
constructor(distro: string) {
|
||||
super(`The connection to the WSL distro ${distro} was closed while it was being opened.`)
|
||||
this.name = 'WslOrcadConnectionSupersededError'
|
||||
}
|
||||
}
|
||||
|
||||
export type WslOrcadConnectOptions = {
|
||||
environmentId?: string
|
||||
label?: string
|
||||
/** Rebinds a port a persisted pairing offer already names; 0 or absent picks a free one. */
|
||||
localPort?: number
|
||||
force?: boolean
|
||||
signal?: AbortSignal
|
||||
}
|
||||
|
||||
export type WslOrcadConnectionDependencies = {
|
||||
holder?: WslDistroHolder
|
||||
deployAndLaunch?: (distro: string, options: WslOrcadDeployOptions) => Promise<WslOrcadLaunch>
|
||||
createTransport?: (target: WslOrcadHostTarget) => OrcadHostTransport
|
||||
listen?: (
|
||||
transport: OrcadHostTransport,
|
||||
options: WslOrcadStdioBridgeOptions
|
||||
) => Promise<StartedPortForward>
|
||||
verifyIdentity?: (pairingOffer: string, readiness: ServeReadiness) => Promise<OrcadTunnelIdentity>
|
||||
/** A connection (re)opened or its offer changed; null once it closed. */
|
||||
onChange?: (distro: string, connection: WslOrcadConnection | null) => void
|
||||
}
|
||||
|
||||
type ActiveWslOrcadConnection = {
|
||||
connection: WslOrcadConnection
|
||||
forward: StartedPortForward
|
||||
}
|
||||
|
||||
export class WslOrcadConnectionManager {
|
||||
private readonly active = new Map<string, ActiveWslOrcadConnection>()
|
||||
private readonly inFlight = new Map<string, Promise<WslOrcadConnection>>()
|
||||
private readonly generations = new Map<string, number>()
|
||||
|
||||
constructor(private readonly dependencies: WslOrcadConnectionDependencies = {}) {}
|
||||
|
||||
/** Joins a connect already running for this distro rather than racing a second bridge. */
|
||||
ensure(distro: string, options: WslOrcadConnectOptions = {}): Promise<WslOrcadConnection> {
|
||||
const name = distro.trim()
|
||||
if (!name) {
|
||||
return Promise.reject(new Error('A WSL distro name is required'))
|
||||
}
|
||||
const key = wslDistroHolderKey(name)
|
||||
const pending = this.inFlight.get(key)
|
||||
if (pending) {
|
||||
return pending
|
||||
}
|
||||
const operation = this.connect(name, key, options).finally(() => {
|
||||
if (this.inFlight.get(key) === operation) {
|
||||
this.inFlight.delete(key)
|
||||
}
|
||||
})
|
||||
this.inFlight.set(key, operation)
|
||||
return operation
|
||||
}
|
||||
|
||||
get(distro: string): WslOrcadConnection | null {
|
||||
return this.active.get(wslDistroHolderKey(distro))?.connection ?? null
|
||||
}
|
||||
|
||||
list(): WslOrcadConnection[] {
|
||||
return [...this.active.values()].map((entry) => entry.connection)
|
||||
}
|
||||
|
||||
async close(distro: string): Promise<void> {
|
||||
const key = wslDistroHolderKey(distro)
|
||||
this.generations.set(key, this.generation(key) + 1)
|
||||
const entry = this.active.get(key)
|
||||
if (!entry) {
|
||||
return
|
||||
}
|
||||
this.active.delete(key)
|
||||
await entry.forward.close()
|
||||
this.dependencies.onChange?.(entry.connection.distro, null)
|
||||
}
|
||||
|
||||
async dispose(): Promise<void> {
|
||||
const keys = new Set([...this.active.keys(), ...this.inFlight.keys()])
|
||||
await Promise.all([...keys].map((key) => this.close(key)))
|
||||
}
|
||||
|
||||
private generation(key: string): number {
|
||||
return this.generations.get(key) ?? 0
|
||||
}
|
||||
|
||||
private async connect(
|
||||
distro: string,
|
||||
key: string,
|
||||
options: WslOrcadConnectOptions
|
||||
): Promise<WslOrcadConnection> {
|
||||
const generation = this.generation(key)
|
||||
const environmentId = options.environmentId ?? defaultWslOrcadEnvironmentId(distro)
|
||||
const label = options.label ?? distro
|
||||
const transport = (this.dependencies.createTransport ?? defaultTransport)({
|
||||
kind: 'wsl',
|
||||
id: environmentId,
|
||||
distro,
|
||||
label
|
||||
})
|
||||
// Every call relaunches a stopped orcad, so a distro restart is recovered here.
|
||||
const launch = await (this.dependencies.deployAndLaunch ?? deployAndLaunchWslOrcad)(distro, {
|
||||
environmentId,
|
||||
label,
|
||||
force: options.force,
|
||||
signal: options.signal,
|
||||
holder: this.dependencies.holder,
|
||||
createTransport: () => transport
|
||||
})
|
||||
if (launch.outcome === 'deferred') {
|
||||
throw new WslOrcadLaunchDeferredError(launch)
|
||||
}
|
||||
if (this.generation(key) !== generation) {
|
||||
throw new WslOrcadConnectionSupersededError(distro)
|
||||
}
|
||||
const current = this.active.get(key)
|
||||
if (
|
||||
current &&
|
||||
current.connection.remotePort === launch.port &&
|
||||
current.connection.environmentId === environmentId
|
||||
) {
|
||||
return this.refreshOffer(key, current, launch)
|
||||
}
|
||||
// orcad moved ports (or this is the first connect): the old bridge dials a dead port.
|
||||
const preferredLocalPort = current?.connection.localPort ?? options.localPort ?? 0
|
||||
if (current) {
|
||||
this.active.delete(key)
|
||||
await current.forward.close()
|
||||
this.dependencies.onChange?.(distro, null)
|
||||
}
|
||||
const forward = await this.listen(transport, {
|
||||
environmentId,
|
||||
remotePort: launch.port,
|
||||
localPort: preferredLocalPort,
|
||||
label: `Orca server in WSL: ${label}`
|
||||
})
|
||||
try {
|
||||
if (this.generation(key) !== generation) {
|
||||
throw new WslOrcadConnectionSupersededError(distro)
|
||||
}
|
||||
const pairingOffer = tunneledOrcadPairingCode(launch.readiness, forward.entry.localPort)
|
||||
const identity = await (this.dependencies.verifyIdentity ?? verifyWslOrcadIdentity)(
|
||||
pairingOffer,
|
||||
launch.readiness
|
||||
)
|
||||
if (identity.verdict === 'foreign') {
|
||||
throw new OrcadManagedIdentityError(launch.port, identity.detail)
|
||||
}
|
||||
if (identity.verdict === 'unreachable') {
|
||||
// Why not fail: a server that is still starting is not a wrong server.
|
||||
console.warn(`[wsl] Orca server in ${distro} did not answer its bridge: ${identity.detail}`)
|
||||
}
|
||||
if (this.generation(key) !== generation) {
|
||||
throw new WslOrcadConnectionSupersededError(distro)
|
||||
}
|
||||
const entry: ActiveWslOrcadConnection = {
|
||||
forward,
|
||||
connection: {
|
||||
distro,
|
||||
environmentId,
|
||||
localPort: forward.entry.localPort,
|
||||
remotePort: launch.port,
|
||||
fullVersion: launch.fullVersion,
|
||||
pairingOffer,
|
||||
close: () => this.closeEntry(key, entry)
|
||||
}
|
||||
}
|
||||
this.active.set(key, entry)
|
||||
this.dependencies.onChange?.(distro, entry.connection)
|
||||
return entry.connection
|
||||
} catch (error) {
|
||||
await forward.close()
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
private refreshOffer(
|
||||
key: string,
|
||||
entry: ActiveWslOrcadConnection,
|
||||
launch: Extract<WslOrcadLaunch, { outcome: 'ready' }>
|
||||
): WslOrcadConnection {
|
||||
const pairingOffer = tunneledOrcadPairingCode(launch.readiness, entry.connection.localPort)
|
||||
if (
|
||||
pairingOffer === entry.connection.pairingOffer &&
|
||||
launch.fullVersion === entry.connection.fullVersion
|
||||
) {
|
||||
return entry.connection
|
||||
}
|
||||
entry.connection = { ...entry.connection, pairingOffer, fullVersion: launch.fullVersion }
|
||||
this.active.set(key, entry)
|
||||
this.dependencies.onChange?.(entry.connection.distro, entry.connection)
|
||||
return entry.connection
|
||||
}
|
||||
|
||||
/** A persisted local port may be taken by now; a free one beats no connection. */
|
||||
private async listen(
|
||||
transport: OrcadHostTransport,
|
||||
options: WslOrcadStdioBridgeOptions
|
||||
): Promise<StartedPortForward> {
|
||||
const listen = this.dependencies.listen ?? listenWslOrcadStdioBridge
|
||||
try {
|
||||
return await listen(transport, options)
|
||||
} catch (error) {
|
||||
if (options.localPort === 0) {
|
||||
throw error
|
||||
}
|
||||
console.warn(
|
||||
`[wsl] Could not rebind local port ${options.localPort}; picking another:`,
|
||||
error instanceof Error ? error.message : error
|
||||
)
|
||||
return listen(transport, { ...options, localPort: 0 })
|
||||
}
|
||||
}
|
||||
|
||||
private async closeEntry(key: string, entry: ActiveWslOrcadConnection): Promise<void> {
|
||||
if (this.active.get(key) === entry) {
|
||||
await this.close(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function verifyWslOrcadIdentity(
|
||||
pairingOffer: string,
|
||||
readiness: ServeReadiness
|
||||
): Promise<OrcadTunnelIdentity> {
|
||||
const offer = parsePairingCode(pairingOffer)
|
||||
return offer
|
||||
? verifyOrcadPairingIdentity(pairedRuntimeConnectionFromOffer(offer), {
|
||||
runtimeId: readiness.runtimeId || null
|
||||
})
|
||||
: { verdict: 'foreign', detail: 'The Orca server in WSL published an invalid pairing offer.' }
|
||||
}
|
||||
|
||||
function defaultTransport(target: WslOrcadHostTarget): OrcadHostTransport {
|
||||
return createWslOrcadHostTransport({ target })
|
||||
}
|
||||
@@ -71,6 +71,11 @@ export type WslOrcadDeployOptions = {
|
||||
readRegistryId?: (distro: string) => Promise<string | null>
|
||||
}
|
||||
|
||||
/** Until the persisted deployment record (C7) assigns one, the environment is keyed per distro. */
|
||||
export function defaultWslOrcadEnvironmentId(distro: string): string {
|
||||
return `wsl:${wslDistroHolderKey(distro)}`
|
||||
}
|
||||
|
||||
/** WSL 1 runs no Linux kernel, so the pinned Node and its pty cannot run there. */
|
||||
export function isWsl1KernelRelease(release: string): boolean {
|
||||
return /^\d+\.\d+\.\d+-\d+-Microsoft$/.test(release.trim())
|
||||
@@ -86,7 +91,7 @@ export async function deployAndLaunchWslOrcad(
|
||||
}
|
||||
const target: WslOrcadHostTarget = {
|
||||
kind: 'wsl',
|
||||
id: options.environmentId ?? `wsl:${wslDistroHolderKey(name)}`,
|
||||
id: options.environmentId ?? defaultWslOrcadEnvironmentId(name),
|
||||
distro: name,
|
||||
label: options.label ?? name
|
||||
}
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
import { connect, type Server, type Socket } from 'node:net'
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from 'vitest'
|
||||
import { checkOrcadStdioBridge, openOrcadStdioBridgeChannel } from '../ssh/orcad-stdio-bridge'
|
||||
import { ORCAD_STDIO_BRIDGE_MAX_CHANNELS } from '../ssh/orcad-stdio-bridge-listener'
|
||||
import { startEchoServer } from '../ssh/orcad-stdio-bridge-test-channel'
|
||||
import type { StartedPortForward } from '../ssh/ssh-port-forward-provider'
|
||||
import { fakeWslBridgeChildren, posixOrcadStdioBridge } from './wsl-orcad-bridge-test-child'
|
||||
import { createWslOrcadHostTransport } from './wsl-orcad-host-transport'
|
||||
import { listenWslOrcadStdioBridge } from './wsl-orcad-stdio-bridge'
|
||||
|
||||
let echo: { server: Server; port: number }
|
||||
const started: StartedPortForward[] = []
|
||||
const clients: Socket[] = []
|
||||
|
||||
beforeAll(async () => {
|
||||
echo = await startEchoServer()
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
for (const client of clients.splice(0)) {
|
||||
client.destroy()
|
||||
}
|
||||
await Promise.all(started.splice(0).map((forward) => forward.close()))
|
||||
})
|
||||
|
||||
afterAll(() => {
|
||||
echo.server.close()
|
||||
})
|
||||
|
||||
async function startBridge(remotePort = echo.port) {
|
||||
const children = fakeWslBridgeChildren()
|
||||
const transport = createWslOrcadHostTransport({
|
||||
target: { kind: 'wsl', id: 'wsl:ubuntu', distro: 'Ubuntu', label: 'Ubuntu' },
|
||||
spawn: children.spawn
|
||||
})
|
||||
const forward = await listenWslOrcadStdioBridge(
|
||||
transport,
|
||||
{ environmentId: 'wsl:ubuntu', remotePort, localPort: 0, label: 'Ubuntu' },
|
||||
{
|
||||
resolveBridge: async (_conn, port) => posixOrcadStdioBridge(port),
|
||||
checkBridge: checkOrcadStdioBridge,
|
||||
openChannel: openOrcadStdioBridgeChannel
|
||||
}
|
||||
)
|
||||
started.push(forward)
|
||||
return { forward, children }
|
||||
}
|
||||
|
||||
function dial(port: number): Promise<Socket> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const socket = connect(port, '127.0.0.1', () => resolve(socket))
|
||||
socket.once('error', reject)
|
||||
clients.push(socket)
|
||||
})
|
||||
}
|
||||
|
||||
function echoOf(socket: Socket, message: string): Promise<string> {
|
||||
return new Promise((resolve) => {
|
||||
let received = ''
|
||||
const onData = (chunk: Buffer): void => {
|
||||
received += chunk.toString('utf8')
|
||||
if (received.length >= message.length) {
|
||||
socket.off('data', onData)
|
||||
resolve(received)
|
||||
}
|
||||
}
|
||||
socket.on('data', onData)
|
||||
socket.write(message)
|
||||
})
|
||||
}
|
||||
|
||||
async function until(condition: () => boolean): Promise<void> {
|
||||
for (let i = 0; i < 200 && !condition(); i++) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 5))
|
||||
}
|
||||
expect(condition()).toBe(true)
|
||||
}
|
||||
|
||||
describe('listenWslOrcadStdioBridge (real listener, fake wsl.exe echo children)', () => {
|
||||
it('rides each accepted socket on its own wsl.exe bridge child', async () => {
|
||||
const { forward, children } = await startBridge()
|
||||
// The up-front check is one child; it has already exited.
|
||||
expect(children.commands).toHaveLength(1)
|
||||
await until(() => children.live() === 0)
|
||||
|
||||
const first = await dial(forward.entry.localPort)
|
||||
const second = await dial(forward.entry.localPort)
|
||||
expect(await echoOf(first, 'one')).toBe('one')
|
||||
expect(await echoOf(second, 'two')).toBe('two')
|
||||
expect(children.commands).toHaveLength(3)
|
||||
expect(children.live()).toBe(2)
|
||||
for (const command of children.commands) {
|
||||
expect(command).toContain("'.orca-remote/runtimes'/node-*/bin/node")
|
||||
expect(command).toContain(` ${echo.port}; fi; done;`)
|
||||
}
|
||||
})
|
||||
|
||||
it(`queues past ${ORCAD_STDIO_BRIDGE_MAX_CHANNELS} sockets and reuses freed slots`, async () => {
|
||||
const { forward, children } = await startBridge()
|
||||
await until(() => children.live() === 0)
|
||||
const count = ORCAD_STDIO_BRIDGE_MAX_CHANNELS + 4
|
||||
const sockets = await Promise.all(
|
||||
Array.from({ length: count }, () => dial(forward.entry.localPort))
|
||||
)
|
||||
const first = sockets.slice(0, ORCAD_STDIO_BRIDGE_MAX_CHANNELS)
|
||||
await Promise.all(first.map((socket, i) => echoOf(socket, `m${i}`)))
|
||||
expect(children.live()).toBe(ORCAD_STDIO_BRIDGE_MAX_CHANNELS)
|
||||
|
||||
for (const socket of first) {
|
||||
socket.destroy()
|
||||
}
|
||||
const rest = sockets.slice(ORCAD_STDIO_BRIDGE_MAX_CHANNELS)
|
||||
expect(await Promise.all(rest.map((socket, i) => echoOf(socket, `q${i}`)))).toEqual(
|
||||
rest.map((_, i) => `q${i}`)
|
||||
)
|
||||
expect(children.maxLive()).toBe(ORCAD_STDIO_BRIDGE_MAX_CHANNELS)
|
||||
expect(children.commands).toHaveLength(count + 1)
|
||||
})
|
||||
|
||||
it('drops a socket whose bridge reports a refused port', async () => {
|
||||
const { forward, children } = await startBridge(1)
|
||||
const socket = await dial(forward.entry.localPort)
|
||||
await new Promise((resolve) => socket.once('close', resolve))
|
||||
await until(() => children.live() === 0)
|
||||
})
|
||||
|
||||
it('kills every bridge child on close and stops listening', async () => {
|
||||
const { forward, children } = await startBridge()
|
||||
const sockets = await Promise.all([1, 2, 3].map(() => dial(forward.entry.localPort)))
|
||||
await Promise.all(sockets.map((socket, i) => echoOf(socket, `c${i}`)))
|
||||
expect(children.live()).toBe(3)
|
||||
|
||||
await forward.close()
|
||||
await until(() => children.live() === 0)
|
||||
await expect(dial(forward.entry.localPort)).rejects.toThrow()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,42 @@
|
||||
/**
|
||||
* The desktop's only way into a WSL distro's orcad (design D2): a loopback listener whose every
|
||||
* accepted socket rides its own `wsl.exe` child running the stdio bridge on the distro's pinned
|
||||
* Node. WSL localhost forwarding is never used: it refuses connections for 0.2–1 s after a listen
|
||||
* and depends on the distro's networking mode.
|
||||
*/
|
||||
import type { OrcadHostTransport } from '../ssh/orcad-host-transport'
|
||||
import {
|
||||
ORCAD_STDIO_BRIDGE_DEPENDENCIES,
|
||||
startOrcadStdioBridge,
|
||||
type OrcadStdioBridgeDependencies
|
||||
} from '../ssh/orcad-stdio-bridge-provider'
|
||||
import type { StartedPortForward } from '../ssh/ssh-port-forward-provider'
|
||||
|
||||
export type WslOrcadStdioBridgeOptions = {
|
||||
environmentId: string
|
||||
/** orcad's loopback port inside the distro. */
|
||||
remotePort: number
|
||||
/** 0 picks a free one. */
|
||||
localPort: number
|
||||
label: string
|
||||
}
|
||||
|
||||
export function listenWslOrcadStdioBridge(
|
||||
transport: OrcadHostTransport,
|
||||
options: WslOrcadStdioBridgeOptions,
|
||||
dependencies: OrcadStdioBridgeDependencies = ORCAD_STDIO_BRIDGE_DEPENDENCIES
|
||||
): Promise<StartedPortForward> {
|
||||
return startOrcadStdioBridge(
|
||||
transport,
|
||||
{
|
||||
id: `wsl-orcad:${options.environmentId}`,
|
||||
connectionId: options.environmentId,
|
||||
localHost: '127.0.0.1',
|
||||
localPort: options.localPort,
|
||||
remoteHost: '127.0.0.1',
|
||||
remotePort: options.remotePort,
|
||||
label: options.label
|
||||
},
|
||||
dependencies
|
||||
)
|
||||
}
|
||||
Reference in New Issue
Block a user