From 6f7d7e17c4aac05d66ce4b9ac9a6dae65b37fe0e Mon Sep 17 00:00:00 2001 From: m4air Date: Wed, 30 Sep 2026 12:08:07 -0700 Subject: [PATCH 1/2] feat(ssh): add the SSH connection work ledger and channel lifetime tracking Dormant foundation for the T2 SSH core port of #16741 (design D9.3). No production module imports it yet; only tests do, so SSH behavior is unchanged. The next T2 slice wires SshConnection.exec/sftp/channel opens and forwarded sockets through it. Adds: - SshConnectionWorkLedger: tracks local operation and channel lifetimes per connection. fenceForReset closes admission (except work nested under an already-admitted operation and one exactly-identified control channel) and drain() waits for the rest, failing sticky on any failure or unverifiable opening observed after the fence. Channel closure is local lifetime evidence only, never proof that a remote process exited. - trackSshConnectionChannelLifetime / openTrackedSshSocket: bind an opened channel or socket to its ledger entry and settle it only on 'close' ('end', 'error' and the destroyed flag are not closure). Porting note (source: #16741 head a68b6f3531, merge-base 277c289bd4; T2 per the #24137 track manifest): - Taken verbatim: ssh-connection-work-ledger behavior and tests, ssh-forward-socket-admission.test.ts, openTrackedSshSocket. - Adapted: the ledger no longer calls assertProfileLifetimeAdmission on every admit. That module is an opt-in native-addon profile lock (addon dropped in T0) that overlaps the SQLite authority from #22612; callers inject their own admission check through the existing constructor argument instead (SshConnection will pass its disposed check). The broad `object` resource type became SshConnectionWorkChannel (on/once/ removeListener), and trackSshConnectionChannelLifetime uses a type guard instead of a cast, so a value without removeListener is now reported unverifiable rather than tracked. Added ssh-connection-channel-lifetime tests, since #16741 covered it only through SshConnection. - Dropped: none in these files. - Not in this slice: SshConnection wiring (ssh-connection.ts), forwarded channel lifetime (T4), upload session lifetime (T6), profile lifetime admission/lock/participation (needs a D9.4 decision against #22612). No wire change: nothing is sent to relays or clients. --- .../ssh-connection-channel-lifetime.test.ts | 65 +++++ .../ssh/ssh-connection-channel-lifetime.ts | 60 ++++ .../ssh/ssh-connection-work-ledger.test.ts | 256 ++++++++++++++++++ src/main/ssh/ssh-connection-work-ledger.ts | 208 ++++++++++++++ .../ssh/ssh-forward-socket-admission.test.ts | 35 +++ 5 files changed, 624 insertions(+) create mode 100644 src/main/ssh/ssh-connection-channel-lifetime.test.ts create mode 100644 src/main/ssh/ssh-connection-channel-lifetime.ts create mode 100644 src/main/ssh/ssh-connection-work-ledger.test.ts create mode 100644 src/main/ssh/ssh-connection-work-ledger.ts create mode 100644 src/main/ssh/ssh-forward-socket-admission.test.ts 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) +}) From a5540c4a62418da67fbf9b71a9ae6a6bede9f766 Mon Sep 17 00:00:00 2001 From: m4air Date: Wed, 30 Sep 2026 12:08:18 -0700 Subject: [PATCH 2/2] feat(ssh): add SSH transport close ledger and close observation Dormant foundation for the T2 SSH core port of #16741. No production module imports it yet, so SSH behavior is unchanged; the next T2 slice has SshConnection track every ssh2 client it allocates and gate its transport-closure subscribers on this ledger plus the work ledger. Adds: - SshTransportCloseLedger: remembers every allocated client (including ones whose authentication failed) until it emits 'close'. drain() waits for the snapshot and fails with ssh_client_close_allocations_changed if a client was allocated after the caller should have fenced allocation. Authentication settlement, 'end' and 'error' are not physical closure. - observeSshTransportClose: registers 'close' listeners on a set of transport resources before close is initiated; wait() is cancelable without dropping the evidence, dispose() removes the listeners. Porting note (source: #16741 head a68b6f3531, merge-base 277c289bd4; T2 per the #24137 track manifest): - Taken verbatim: both modules and ssh-transport-close-ledger.test.ts. - Adapted: the ledger test no longer casts EventEmitter to ssh2 Client, since track() already accepts an EventEmitter. Added ssh-transport-close-observation tests, since #16741 covered it only through SshConnection.dispose. - Dropped: none. No wire change. Loss of transport is still reported as unverifiable, never as remote process exit. --- .../ssh/ssh-transport-close-ledger.test.ts | 69 +++++++++++++++++++ src/main/ssh/ssh-transport-close-ledger.ts | 39 +++++++++++ .../ssh-transport-close-observation.test.ts | 33 +++++++++ .../ssh/ssh-transport-close-observation.ts | 25 +++++++ 4 files changed, 166 insertions(+) create mode 100644 src/main/ssh/ssh-transport-close-ledger.test.ts create mode 100644 src/main/ssh/ssh-transport-close-ledger.ts create mode 100644 src/main/ssh/ssh-transport-close-observation.test.ts create mode 100644 src/main/ssh/ssh-transport-close-observation.ts 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) + } + } + } +}