diff --git a/src/main/ssh/ssh-connection-channel-lifetime.test.ts b/src/main/ssh/ssh-connection-channel-lifetime.test.ts new file mode 100644 index 000000000000..56b5a84f92a9 --- /dev/null +++ b/src/main/ssh/ssh-connection-channel-lifetime.test.ts @@ -0,0 +1,65 @@ +import { EventEmitter } from 'node:events' +import { expect, it } from 'vitest' +import { trackSshConnectionChannelLifetime } from './ssh-connection-channel-lifetime' +import { SshConnectionWorkLedger } from './ssh-connection-work-ledger' + +const signal = () => new AbortController().signal + +it('settles the opening only when the channel emits close', async () => { + const ledger = new SshConnectionWorkLedger() + const channel = new EventEmitter() + trackSshConnectionChannelLifetime(ledger.beginChannelOpen(), channel) + const fence = ledger.fenceForReset() + channel.emit('end') + expect(fence.assertDrained).toThrow('ssh_connection_work_not_drained') + channel.emit('close') + await fence.drain(signal()) + expect(channel.listenerCount('error')).toBe(0) +}) + +it('treats an already closed channel as settled', async () => { + const ledger = new SshConnectionWorkLedger() + trackSshConnectionChannelLifetime( + ledger.beginChannelOpen(), + Object.assign(new EventEmitter(), { closed: true }) + ) + await ledger.fenceForReset().drain(signal()) +}) + +it('keeps a fenced channel error unverifiable even after it closes', async () => { + const ledger = new SshConnectionWorkLedger() + const channel = new EventEmitter() + trackSshConnectionChannelLifetime(ledger.beginChannelOpen(), channel) + const fence = ledger.fenceForReset() + const failure = new Error('channel write failed') + channel.emit('error', failure) + channel.emit('close') + await expect(fence.drain(signal())).rejects.toBe(failure) +}) + +it('does not carry a pre-fence error from a channel that closed before the fence', async () => { + const ledger = new SshConnectionWorkLedger() + const channel = new EventEmitter() + trackSshConnectionChannelLifetime(ledger.beginChannelOpen(), channel) + channel.emit('error', new Error('closed before reset')) + channel.emit('close') + await ledger.fenceForReset().drain(signal()) +}) + +it.each([undefined, null, {}, { on: () => {} }])( + 'marks an untrackable open result (%s) unverifiable instead of settled', + async (value) => { + const ledger = new SshConnectionWorkLedger() + trackSshConnectionChannelLifetime(ledger.beginChannelOpen(), value) + await expect(ledger.fenceForReset().drain(signal())).rejects.toThrow( + 'ssh_connection_channel_lifetime_unverifiable' + ) + } +) + +it('runs the injected admission check before admitting work', () => { + const ledger = new SshConnectionWorkLedger(undefined, () => { + throw new Error('connection disposed') + }) + expect(() => ledger.beginChannelOpen()).toThrow('connection disposed') +}) diff --git a/src/main/ssh/ssh-connection-channel-lifetime.ts b/src/main/ssh/ssh-connection-channel-lifetime.ts new file mode 100644 index 000000000000..c6ad86100bcf --- /dev/null +++ b/src/main/ssh/ssh-connection-channel-lifetime.ts @@ -0,0 +1,60 @@ +import type { + SshConnectionWorkChannel, + SshConnectionWorkLedger +} from './ssh-connection-work-ledger' + +type TrackableChannel = SshConnectionWorkChannel & { + closed?: unknown +} + +function isTrackableChannel(value: unknown): value is TrackableChannel { + if (typeof value !== 'object' || value === null) { + return false + } + return ( + 'on' in value && + typeof value.on === 'function' && + 'once' in value && + typeof value.once === 'function' && + 'removeListener' in value && + typeof value.removeListener === 'function' + ) +} + +export function openTrackedSshSocket( + ledger: SshConnectionWorkLedger, + open: () => T +): T { + const work = ledger.beginChannelOpen() + try { + const socket = open() + trackSshConnectionChannelLifetime(work, socket) + return socket + } catch (error) { + work.markUnverifiable(error instanceof Error ? error : new Error(String(error))) + throw error + } +} + +/** Start tracking before the open callback hands the channel to another owner. */ +export function trackSshConnectionChannelLifetime( + work: ReturnType, + value: unknown +): void { + if (!isTrackableChannel(value)) { + work.markUnverifiable(new Error('ssh_connection_channel_lifetime_unverifiable')) + return + } + const channel = value + work.bind(channel) + if (channel.closed === true) { + work.close() + return + } + const onError = (error: Error) => work.markUnverifiable(error) + channel.on('error', onError) + channel.once('close', () => { + channel.removeListener('error', onError) + work.close() + }) +} diff --git a/src/main/ssh/ssh-connection-work-ledger.test.ts b/src/main/ssh/ssh-connection-work-ledger.test.ts new file mode 100644 index 000000000000..25fa62e36531 --- /dev/null +++ b/src/main/ssh/ssh-connection-work-ledger.test.ts @@ -0,0 +1,256 @@ +import { EventEmitter } from 'node:events' +import { expect, it } from 'vitest' +import { SshConnectionWorkLedger } from './ssh-connection-work-ledger' + +const signal = () => new AbortController().signal + +it('allows admitted work to open nested channels after fencing and drains children independently', async () => { + const ledger = new SshConnectionWorkLedger() + const continueOperation = Promise.withResolvers() + let child!: ReturnType + const parent = ledger.run(async () => { + await continueOperation.promise + await ledger.run(async () => { + child = ledger.beginChannelOpen() + child.bind(new EventEmitter()) + }) + }) + const fence = ledger.fenceForReset() + let drained = false + const draining = fence.drain(signal()).then(() => { + drained = true + }) + continueOperation.resolve() + await parent + expect(drained).toBe(false) + expect(fence.assertDrained).toThrow('ssh_connection_work_not_drained') + child.close() + await draining + fence.assertDrained() +}) + +it('refuses stale async scope work once its admitted parent has settled', async () => { + const ledger = new SshConnectionWorkLedger() + const releaseStale = Promise.withResolvers() + const staleResult = Promise.withResolvers() + await ledger.run(async () => { + void releaseStale.promise.then(() => { + try { + ledger.beginChannelOpen() + staleResult.resolve('incorrectly admitted') + } catch (error) { + staleResult.resolve(error) + } + }) + }) + const fence = ledger.fenceForReset() + releaseStale.resolve() + expect(await staleResult.promise).toMatchObject({ + message: 'ssh_connection_work_admission_closed' + }) + await fence.drain(signal()) +}) + +it('exempts only the exact captured control channel and keeps ordinary admission fenced', async () => { + const ledger = new SshConnectionWorkLedger() + const control = ledger.beginChannelOpen() + const resource = new EventEmitter() + control.bind(resource) + const transfer = ledger.beginChannelOpen() + transfer.bind(new EventEmitter()) + expect(() => ledger.fenceForReset(new EventEmitter())).toThrow( + 'ssh_connection_work_control_channel_unproven' + ) + const fence = ledger.fenceForReset(resource) + expect(() => ledger.beginChannelOpen()).toThrow('ssh_connection_work_admission_closed') + await expect(ledger.run(async () => undefined)).rejects.toThrow( + 'ssh_connection_work_admission_closed' + ) + expect(fence.assertDrained).toThrow('ssh_connection_work_not_drained') + transfer.close() + await fence.drain(signal()) + fence.assertDrained() + control.close(new Error('expected reset transport close')) + fence.assertDrained() + expect(() => ledger.fenceForReset(resource)).toThrow('ssh_connection_work_already_fenced') +}) + +it('refuses ambiguous control-channel bindings without publishing a fence', async () => { + const ledger = new SshConnectionWorkLedger() + const resource = new EventEmitter() + const first = ledger.beginChannelOpen() + const second = ledger.beginChannelOpen() + first.bind(resource) + second.bind(resource) + expect(() => ledger.fenceForReset(resource)).toThrow( + 'ssh_connection_work_control_channel_unproven' + ) + await ledger.run(async () => undefined) + second.close() + await ledger.fenceForReset(resource).drain(signal()) +}) + +it('keeps uncertain pre-fence openings unconfirmed after late channel close', async () => { + const ledger = new SshConnectionWorkLedger() + const opening = ledger.beginChannelOpen() + const uncertainty = new Error('open timed out without closure evidence') + opening.markUnverifiable(uncertainty) + const fence = ledger.fenceForReset() + await expect(fence.drain(signal())).rejects.toBe(uncertainty) + opening.bind(new EventEmitter()) + opening.close() + expect(fence.assertDrained).toThrow(uncertainty) + await expect(fence.drain(signal())).rejects.toBe(uncertainty) +}) + +it('retains operation failures observed while draining', async () => { + const ledger = new SshConnectionWorkLedger() + const operation = Promise.withResolvers() + const failure = new Error('upload failed') + const running = ledger.run(() => operation.promise).catch((error: unknown) => error) + const fence = ledger.fenceForReset() + const draining = fence.drain(signal()).catch((error: unknown) => error) + operation.reject(failure) + expect(await running).toBe(failure) + expect(await draining).toBe(failure) + expect(fence.assertDrained).toThrow(failure) + await expect(fence.drain(signal())).rejects.toBe(failure) +}) + +it('retains a channel failure even after repeated successful close calls', async () => { + const ledger = new SshConnectionWorkLedger() + const channel = ledger.beginChannelOpen() + const fence = ledger.fenceForReset() + const failure = new Error('channel failed') + channel.close(failure) + channel.close() + expect(fence.assertDrained).toThrow(failure) + await expect(fence.drain(signal())).rejects.toBe(failure) +}) + +it('wakes an in-flight drain when an opening becomes unverifiable without waiting for close', async () => { + const ledger = new SshConnectionWorkLedger() + const channel = ledger.beginChannelOpen() + const fence = ledger.fenceForReset() + const failure = new Error('opening outcome became unverifiable') + let observed: unknown + const draining = fence.drain(signal()).catch((error: unknown) => { + observed = error + }) + channel.markUnverifiable(failure) + await new Promise((resolve) => setImmediate(resolve)) + const observedBeforeClose = observed + channel.close() + await draining + expect(observedBeforeClose).toBe(failure) +}) + +it('reports one failed channel without waiting for unrelated admitted channels to close', async () => { + const ledger = new SshConnectionWorkLedger() + const failedChannel = ledger.beginChannelOpen() + const liveChannel = ledger.beginChannelOpen() + const fence = ledger.fenceForReset() + const failure = new Error('selected transfer failed') + let observed: unknown + const draining = fence.drain(signal()).catch((error: unknown) => { + observed = error + }) + failedChannel.close(failure) + await new Promise((resolve) => setImmediate(resolve)) + const observedBeforeOtherClose = observed + liveChannel.close() + await draining + expect(observedBeforeOtherClose).toBe(failure) +}) + +it('cancels only the wait and permits exact drain retry without reopening admission', async () => { + const ledger = new SshConnectionWorkLedger() + const opening = ledger.beginChannelOpen() + const fence = ledger.fenceForReset() + const controller = new AbortController() + const cancellation = new Error('user cancelled waiting') + const draining = fence.drain(controller.signal).catch((error: unknown) => error) + controller.abort(cancellation) + expect(await draining).toBe(cancellation) + expect(fence.assertDrained).toThrow('ssh_connection_work_not_drained') + expect(() => ledger.beginChannelOpen()).toThrow('ssh_connection_work_admission_closed') + opening.bind(new EventEmitter()) + opening.close() + await fence.drain(signal()) + fence.assertDrained() +}) + +it('isolates admission, async ancestry, failures, and control identities between connections', async () => { + const first = new SshConnectionWorkLedger() + const second = new SshConnectionWorkLedger() + const resource = new EventEmitter() + first.beginChannelOpen().bind(resource) + const firstFence = first.fenceForReset(resource) + await second.run(async () => { + expect(() => first.beginChannelOpen()).toThrow('ssh_connection_work_admission_closed') + }) + expect(() => second.fenceForReset(resource)).toThrow( + 'ssh_connection_work_control_channel_unproven' + ) + const channel = second.beginChannelOpen() + const secondFence = second.fenceForReset() + channel.close(new Error('second connection failure')) + await expect(secondFence.drain(signal())).rejects.toThrow('second connection failure') + await firstFence.drain(signal()) +}) + +it('rejects rebinding and binding an already settled opening', () => { + const ledger = new SshConnectionWorkLedger() + const channel = ledger.beginChannelOpen() + channel.bind(new EventEmitter()) + expect(() => channel.bind(new EventEmitter())).toThrow( + 'ssh_connection_work_channel_binding_changed' + ) + const closed = ledger.beginChannelOpen() + closed.close() + expect(() => closed.bind(new EventEmitter())).toThrow( + 'ssh_connection_work_channel_binding_changed' + ) +}) + +it.each([false, true])( + 'preserves drained proof after exact exempt control closes (error=%s)', + async (withError) => { + const ledger = new SshConnectionWorkLedger() + const control = ledger.beginChannelOpen() + const resource = new EventEmitter() + control.bind(resource) + const fence = ledger.fenceForReset(resource) + await fence.drain(signal()) + if (withError) { + control.markUnverifiable(new Error('control stream closed after response')) + } + control.close(withError ? new Error('control closed') : undefined) + fence.assertDrained() + await fence.drain(signal()) + expect(() => ledger.beginChannelOpen()).toThrow('admission_closed') + } +) + +it('control closure cannot erase retained unrelated work or transport failure', async () => { + for (const transportFailure of [false, true]) { + const ledger = new SshConnectionWorkLedger() + const control = ledger.beginChannelOpen() + const resource = new EventEmitter() + control.bind(resource) + const work = ledger.beginChannelOpen() + const fence = ledger.fenceForReset(resource) + if (transportFailure) { + work.close() + await fence.drain(signal()) + ledger.markTransportUnverifiable() + } else { + work.close(new Error('unrelated write failed')) + } + control.close() + expect(fence.assertDrained).toThrow( + transportFailure ? 'transport_unverifiable' : 'unrelated write failed' + ) + await expect(fence.drain(signal())).rejects.toThrow() + } +}) diff --git a/src/main/ssh/ssh-connection-work-ledger.ts b/src/main/ssh/ssh-connection-work-ledger.ts new file mode 100644 index 000000000000..183bc270a396 --- /dev/null +++ b/src/main/ssh/ssh-connection-work-ledger.ts @@ -0,0 +1,208 @@ +import { AsyncLocalStorage } from 'node:async_hooks' +import { waitForPromiseWithSignal } from '../../shared/abort-signal-reason' + +/** Identity of a channel or socket; only compared by reference. */ +export type SshConnectionWorkChannel = Pick + +type WorkEntry = { + done: Promise + finish: () => void + resource?: SshConnectionWorkChannel + failure?: Error +} + +export class SshConnectionWorkAdmissionClosedError extends Error { + constructor() { + super('ssh_connection_work_admission_closed') + this.name = 'SshConnectionWorkAdmissionClosedError' + } +} + +/** Tracks local operation/channel lifetime; channel closure is not remote process-exit proof. */ +export class SshConnectionWorkLedger { + private readonly pending = new Set() + private readonly scope = new AsyncLocalStorage() + private fenced = false + private exempt: WorkEntry | undefined + private failure: Error | undefined + private readonly failureWake = Promise.withResolvers() + + constructor( + private readonly onChange: () => void = () => {}, + private readonly assertAdmission: () => void = () => {} + ) {} + + isDrained(): boolean { + return this.pending.size === 0 && !this.failure + } + + private recordFailure(error: Error): void { + this.failure ??= error + this.failureWake.resolve(this.failure) + } + + markTransportUnverifiable(): void { + if (this.fenced) { + this.recordFailure(new Error('ssh_connection_reset_transport_unverifiable')) + } + } + + private admit(): WorkEntry { + this.assertAdmission() + const parent = this.scope.getStore() + if (this.fenced && (!parent || !this.pending.has(parent) || parent === this.exempt)) { + throw new SshConnectionWorkAdmissionClosedError() + } + const completion = Promise.withResolvers() + const entry: WorkEntry = { done: completion.promise, finish: completion.resolve } + this.pending.add(entry) + return entry + } + + private settle(entry: WorkEntry, error?: unknown): void { + if (!this.pending.delete(entry)) { + return + } + if (error !== undefined) { + entry.failure = error instanceof Error ? error : new Error(String(error)) + } + if (this.fenced && entry !== this.exempt && entry.failure) { + this.recordFailure(entry.failure) + } + entry.finish() + this.onChange() + } + + async run(operation: () => Promise): Promise { + const entry = this.admit() + return this.scope.run(entry, async () => { + try { + const result = await operation() + this.settle(entry) + return result + } catch (error) { + this.settle(entry, error) + throw error + } + }) + } + + beginSession() { + const entry = this.admit() + let closed = false + let closeFailed = false + let active = 0 + const finish = () => { + if (closed && !closeFailed && active === 0) { + this.settle(entry) + } + } + return { + run: async (operation: () => Promise): Promise => { + if (closed) { + throw new Error('ssh_connection_work_session_closed') + } + active++ + try { + return await this.scope.run(entry, () => this.run(operation)) + } catch (error) { + entry.failure ??= error instanceof Error ? error : new Error(String(error)) + if (this.fenced) { + this.recordFailure(entry.failure) + } + throw error + } finally { + active-- + finish() + } + }, + close: (error?: unknown) => { + closed = true + if (error !== undefined) { + // Failed closure must remain observable even if reset has not started yet. + closeFailed = true + entry.failure ??= error instanceof Error ? error : new Error(String(error)) + if (this.fenced) { + this.recordFailure(entry.failure) + } + } + finish() + } + } + } + + beginChannelOpen() { + const entry = this.admit() + let bound = false + return { + bind: (resource: SshConnectionWorkChannel) => { + if (bound || !this.pending.has(entry)) { + throw new Error('ssh_connection_work_channel_binding_changed') + } + entry.resource = resource + bound = true + }, + markUnverifiable: (error: Error) => { + if (this.pending.has(entry)) { + entry.failure ??= error + if (this.fenced && entry !== this.exempt) { + this.recordFailure(error) + } + } + }, + close: (error?: unknown) => this.settle(entry, error) + } + } + + fenceForReset(controlChannel?: SshConnectionWorkChannel) { + if (this.fenced) { + throw new Error('ssh_connection_work_already_fenced') + } + const matches = controlChannel + ? [...this.pending].filter((entry) => entry.resource === controlChannel) + : [] + if (controlChannel && matches.length !== 1) { + throw new Error('ssh_connection_work_control_channel_unproven') + } + this.exempt = matches[0] + this.fenced = true + const assertDrained = () => { + if (this.failure) { + throw this.failure + } + if ([...this.pending].some((entry) => entry !== this.exempt)) { + throw new Error('ssh_connection_work_not_drained') + } + } + return { + drain: async (signal: AbortSignal) => { + signal.throwIfAborted() + for (;;) { + if (this.failure) { + throw this.failure + } + const pending = [...this.pending].filter((entry) => entry !== this.exempt) + for (const entry of pending) { + if (entry.failure) { + throw entry.failure + } + } + if (pending.length === 0) { + assertDrained() + return + } + await waitForPromiseWithSignal( + Promise.race([ + Promise.all(pending.map((entry) => entry.done)), + this.failureWake.promise.then((error) => { + throw error + }) + ]), + signal + ) + } + }, + assertDrained + } + } +} diff --git a/src/main/ssh/ssh-forward-socket-admission.test.ts b/src/main/ssh/ssh-forward-socket-admission.test.ts new file mode 100644 index 000000000000..e337eab68658 --- /dev/null +++ b/src/main/ssh/ssh-forward-socket-admission.test.ts @@ -0,0 +1,35 @@ +import { EventEmitter } from 'node:events' +import { expect, it, vi } from 'vitest' +import { openTrackedSshSocket } from './ssh-connection-channel-lifetime' +import { SshConnectionWorkLedger } from './ssh-connection-work-ledger' + +it('acquires admission before a socket factory can connect', () => { + const ledger = new SshConnectionWorkLedger() + const open = vi.fn(() => new EventEmitter()) + ledger.fenceForReset() + expect(() => openTrackedSshSocket(ledger, open)).toThrow('admission_closed') + expect(open).not.toHaveBeenCalled() +}) + +it('waits for actual close rather than the destroyed flag', async () => { + const ledger = new SshConnectionWorkLedger() + const socket = openTrackedSshSocket(ledger, () => + Object.assign(new EventEmitter(), { destroyed: false }) + ) + const fence = ledger.fenceForReset() + socket.destroyed = true + expect(fence.assertDrained).toThrow('not_drained') + socket.emit('close') + await fence.drain(new AbortController().signal) +}) + +it('retains socket-construction uncertainty without accepting a later reset', async () => { + const ledger = new SshConnectionWorkLedger() + const failure = new Error('socket setup failed') + expect(() => + openTrackedSshSocket(ledger, () => { + throw failure + }) + ).toThrow(failure) + await expect(ledger.fenceForReset().drain(new AbortController().signal)).rejects.toBe(failure) +}) diff --git a/src/main/ssh/ssh-transport-close-ledger.test.ts b/src/main/ssh/ssh-transport-close-ledger.test.ts new file mode 100644 index 000000000000..b126c9cb0b10 --- /dev/null +++ b/src/main/ssh/ssh-transport-close-ledger.test.ts @@ -0,0 +1,69 @@ +import { EventEmitter } from 'node:events' +import { expect, it, vi } from 'vitest' +import { SshTransportCloseLedger } from './ssh-transport-close-ledger' + +function fixture() { + const ledger = new SshTransportCloseLedger() + const allocate = () => { + const client = new EventEmitter() + ledger.track(client) + return client + } + return { ledger, allocate } +} + +it('waits for all allocated clients, including failed authentication attempts', async () => { + const { ledger, allocate } = fixture() + const failed = allocate() + const successful = allocate() + const done = vi.fn() + const draining = ledger.drain(new AbortController().signal).then(done) + failed.emit('end') + successful.emit('close') + await Promise.resolve() + await Promise.resolve() + expect(done).not.toHaveBeenCalled() + failed.emit('close') + await draining + expect(done).toHaveBeenCalledOnce() + expect(failed.listenerCount('close')).toBe(0) + expect(successful.listenerCount('close')).toBe(0) +}) + +it('remembers closure before drain and does not accumulate listeners', async () => { + const { ledger, allocate } = fixture() + for (let i = 0; i < 100; i++) { + const client = allocate() + ledger.track(client) + expect(client.listenerCount('close')).toBe(1) + client.emit('close') + expect(client.listenerCount('close')).toBe(0) + } + await ledger.drain(new AbortController().signal) + await ledger.drain(new AbortController().signal) +}) + +it('retains physical closure evidence across canceled waits', async () => { + const { ledger, allocate } = fixture() + const client = allocate() + const controller = new AbortController() + const draining = ledger.drain(controller.signal) + controller.abort() + await expect(draining).rejects.toThrow() + expect(client.listenerCount('close')).toBe(1) + const retry = ledger.drain(new AbortController().signal) + client.emit('close') + await retry + expect(client.listenerCount('close')).toBe(0) +}) + +it('refuses new outstanding allocations made after the drain snapshot', async () => { + const { ledger, allocate } = fixture() + const original = allocate() + const draining = ledger.drain(new AbortController().signal) + const late = allocate() + original.emit('close') + await expect(draining).rejects.toThrow('ssh_client_close_allocations_changed') + late.emit('close') + await ledger.drain(new AbortController().signal) +}) diff --git a/src/main/ssh/ssh-transport-close-ledger.ts b/src/main/ssh/ssh-transport-close-ledger.ts new file mode 100644 index 000000000000..b9721ea71fc7 --- /dev/null +++ b/src/main/ssh/ssh-transport-close-ledger.ts @@ -0,0 +1,39 @@ +import type { EventEmitter } from 'node:events' +import { waitForPromiseWithSignal } from '../../shared/abort-signal-reason' + +/** Authentication settlement does not establish physical socket closure. */ +export class SshTransportCloseLedger { + private readonly pending = new Map>() + + constructor(private readonly onChange: () => void = () => {}) {} + + isClosed(): boolean { + return this.pending.size === 0 + } + + track(client: EventEmitter): void { + if (this.pending.has(client)) { + return + } + let resolveClose!: () => void + const closed = new Promise((resolve) => { + resolveClose = resolve + }) + const onClose = (): void => { + client.off('close', onClose) + this.pending.delete(client) + resolveClose() + this.onChange() + } + this.pending.set(client, closed) + client.on('close', onClose) + } + + /** Caller must fence new allocations before taking this snapshot. */ + async drain(signal: AbortSignal): Promise { + await waitForPromiseWithSignal(Promise.all(this.pending.values()), signal) + if (this.pending.size !== 0) { + throw new Error('ssh_client_close_allocations_changed') + } + } +} diff --git a/src/main/ssh/ssh-transport-close-observation.test.ts b/src/main/ssh/ssh-transport-close-observation.test.ts new file mode 100644 index 000000000000..c12950110b10 --- /dev/null +++ b/src/main/ssh/ssh-transport-close-observation.test.ts @@ -0,0 +1,33 @@ +import { EventEmitter } from 'node:events' +import { expect, it, vi } from 'vitest' +import { observeSshTransportClose } from './ssh-transport-close-observation' + +it('resolves only after every observed resource emits close', async () => { + const client = new EventEmitter() + client.on('error', () => {}) + const proxy = new EventEmitter() + const observation = observeSshTransportClose([client, proxy]) + const done = vi.fn() + const waiting = observation.wait(new AbortController().signal).then(done) + client.emit('end') + client.emit('error', new Error('reset by peer')) + client.emit('close') + await Promise.resolve() + expect(done).not.toHaveBeenCalled() + proxy.emit('close') + await waiting + expect(done).toHaveBeenCalledOnce() +}) + +it('cancels only the wait and removes its listeners on dispose', async () => { + const client = new EventEmitter() + client.on('error', () => {}) + const observation = observeSshTransportClose([client]) + const controller = new AbortController() + const waiting = observation.wait(controller.signal) + controller.abort(new Error('caller gave up')) + await expect(waiting).rejects.toThrow('caller gave up') + expect(client.listenerCount('close')).toBe(1) + observation.dispose() + expect(client.listenerCount('close')).toBe(0) +}) diff --git a/src/main/ssh/ssh-transport-close-observation.ts b/src/main/ssh/ssh-transport-close-observation.ts new file mode 100644 index 000000000000..5f604303d421 --- /dev/null +++ b/src/main/ssh/ssh-transport-close-observation.ts @@ -0,0 +1,25 @@ +import type { EventEmitter } from 'node:events' +import { waitForPromiseWithSignal } from '../../shared/abort-signal-reason' + +/** Register before initiating close; 'end', 'error' and destroy requests are not closure. */ +export function observeSshTransportClose(resources: readonly EventEmitter[]) { + const listeners: { resource: EventEmitter; close: () => void }[] = [] + const closed = Promise.all( + resources.map( + (resource) => + new Promise((resolve) => { + const close = () => resolve() + listeners.push({ resource, close }) + resource.once('close', close) + }) + ) + ) + return { + wait: (signal: AbortSignal) => waitForPromiseWithSignal(closed, signal), + dispose: () => { + for (const { resource, close } of listeners) { + resource.removeListener('close', close) + } + } + } +}