feat(ssh): port the SSH connection work ledger and transport close ledger (#16741 T2)

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')
+})

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<T extends NodeJS.EventEmitter>(
+  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<SshConnectionWorkLedger['beginChannelOpen']>,
+  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()
+  })
+}

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<void>()
+  let child!: ReturnType<typeof ledger.beginChannelOpen>
+  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<void>()
+  const staleResult = Promise.withResolvers<unknown>()
+  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<void>()
+  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<void>((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<void>((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()
+  }
+})