fix(e2b): fence teardown and terminal publication

2026-07-29 07:42:25 +08:00
parent e9bef2967d
commit 022841027f
9 changed files with 237 additions and 22 deletions
@@ -28,6 +28,8 @@ export class E2BSubprocessService extends SubprocessService {

  private readonly live = new Set<E2BSubprocessHandle>()
  private readonly terminals = new Set<SubprocessTerminalHandle>()
+  private readonly terminalSetups = new Set<Promise<void>>()
+  private disposing = false

  /** @inheritdoc */
  readonly cwd: string
@@ -41,12 +43,17 @@ export class E2BSubprocessService extends SubprocessService {
    this.cwd = ctx.e2b.cwd
    this.runtimeRoot = ctx.e2b.runtimeRoot
    ctx.effect(() => async () => {
+      this.disposing = true
+      await Promise.all([...this.terminalSetups])
      const handles = [...this.live]
      const terminals = [...this.terminals]
      const pending: Promise<unknown>[] = []
      for (const handle of handles) {
        handle.terminate()
-        pending.push(handle.waitForExit().then(() => { this.live.delete(handle) }))
+        pending.push(handle.waitForExit().then(async () => {
+          await handle.done.catch(() => undefined)
+          this.live.delete(handle)
+        }))
      }
      for (const terminal of terminals) {
        terminal.terminate()
@@ -89,6 +96,7 @@ export class E2BSubprocessService extends SubprocessService {

  /** @inheritdoc */
  spawn(spec: SubprocessSpawnSpec): SubprocessHandle {
+    if (this.isDisposing()) throw new Error('subprocess-e2b: service is disposing')
    const program = spec.argv[0]
    if (program === undefined || program.length === 0) {
      throw new Error('invalid argv: expected a non-empty program name at argv[0]')
@@ -112,6 +120,7 @@ export class E2BSubprocessService extends SubprocessService {

  /** @inheritdoc */
  async spawnTerminal(spec: SubprocessTerminalSpawnSpec): Promise<SubprocessTerminalHandle> {
+    if (this.isDisposing()) throw new Error('subprocess-e2b: service is disposing')
    const program = spec.argv[0]
    if (program === undefined || program.length === 0) {
      throw new Error('subprocess-e2b: terminal argv must contain a program')
@@ -123,14 +132,31 @@ export class E2BSubprocessService extends SubprocessService {
    }
    spec.signal?.throwIfAborted()
    const stateDir = posix.join(this.runtimeRoot, 'terminals', randomUUID())
-    const terminal = await spawnE2BTerminal(this.ctx.e2b, spec, stateDir)
-    this.terminals.add(terminal)
-    const release = async (): Promise<void> => {
-      await terminal.waitForExit()
-      this.terminals.delete(terminal)
+    const setup = Promise.withResolvers<void>()
+    this.terminalSetups.add(setup.promise)
+    try {
+      const terminal = await spawnE2BTerminal(this.ctx.e2b, spec, stateDir)
+      this.terminals.add(terminal)
+      if (this.isDisposing()) {
+        terminal.terminate()
+        await terminal.waitForExit()
+        this.terminals.delete(terminal)
+        throw new Error('subprocess-e2b: service disposed during terminal setup')
+      }
+      const release = async (): Promise<void> => {
+        await terminal.waitForExit()
+        this.terminals.delete(terminal)
+      }
+      void terminal.done.then(release, release).catch(() => {})
+      return terminal
+    } finally {
+      this.terminalSetups.delete(setup.promise)
+      setup.resolve()
    }
-    void terminal.done.then(release, release).catch(() => {})
-    return terminal
+  }
+
+  private isDisposing(): boolean {
+    return this.disposing
  }
 }

@@ -213,6 +213,7 @@ export class E2BSubprocessHandle implements SubprocessHandle {
  private invalidHandleQuiescent = false
  private provisionalHandleQuiescent = false
  private terminationStarted = false
+  private terminationSucceeded = false
  private terminationAttempt: Promise<void> | undefined
  private terminationFailure: Error | undefined
  private terminationSignal: NodeJS.Signals | null = null
@@ -264,13 +265,14 @@ export class E2BSubprocessHandle implements SubprocessHandle {

  /** @inheritdoc */
  terminate(): void {
-    if (this.terminationAttempt !== undefined) return
+    if (this.terminationSucceeded || this.terminationAttempt !== undefined) return
    this.terminationStarted = true
    this.terminationFailure = undefined
    const attempt = this.terminateRemote()
    this.terminationAttempt = attempt
    void attempt.then(
      () => {
+        this.terminationSucceeded = true
        this.terminationAttempt = undefined
      },
      (error: unknown) => {
@@ -1,6 +1,7 @@
 /** E2B PTY allocation and process-session ownership for the subprocess seam. */

 import { Buffer } from 'node:buffer'
+import { randomUUID } from 'node:crypto'
 import { PassThrough } from 'node:stream'
 import { posix } from 'node:path'
 import {
@@ -28,11 +29,13 @@ const TERMINAL_RUNNER_SOURCE = [
  'dsh_state=$1',
  'mapfile -d \'\' -t dsh_env < "$dsh_state/environment"',
  'mapfile -d \'\' -t dsh_argv < "$dsh_state/argv"',
-  'rm -f -- "$dsh_state/environment" "$dsh_state/argv" "$dsh_state/runner.bash"',
+  'dsh_output_marker=$(<"$dsh_state/output-marker")',
+  'rm -f -- "$dsh_state/environment" "$dsh_state/argv" "$dsh_state/output-marker" "$dsh_state/runner.bash"',
  'if (( ${#dsh_argv[@]} == 0 )); then',
  "  printf 'terminal runner received empty argv\\n' >&2",
  '  exit 125',
  'fi',
+  'printf \'%s\' "$dsh_output_marker"',
  "printf 'ready\\n' > \"$dsh_state/ready\"",
  'exec env -i "${dsh_env[@]}" "${dsh_argv[@]}"',
  '',
@@ -42,6 +45,7 @@ interface TerminalPaths {
  runner: string
  environment: string
  argv: string
+  outputMarker: string
  ready: string
 }

@@ -53,6 +57,73 @@ function delay(ms: number): Promise<void> {
  return new Promise(resolve => setTimeout(resolve, ms))
 }

+class BootstrapOutputFilter {
+  readonly ready: Promise<void>
+
+  private readonly readyState = Promise.withResolvers<void>()
+  private pending = Buffer.alloc(0)
+  private published = false
+
+  constructor(
+    private readonly marker: Buffer,
+    private readonly output: PassThrough,
+  ) {
+    this.ready = this.readyState.promise
+  }
+
+  push(data: Uint8Array): void {
+    if (this.published) {
+      this.write(data)
+      return
+    }
+    const combined = Buffer.concat([this.pending, Buffer.from(data)])
+    const markerOffset = combined.indexOf(this.marker)
+    if (markerOffset < 0) {
+      const retained = Math.min(combined.length, this.marker.length - 1)
+      this.pending = Buffer.from(combined.subarray(combined.length - retained))
+      return
+    }
+    this.published = true
+    this.pending = Buffer.alloc(0)
+    this.readyState.resolve()
+    this.write(combined.subarray(markerOffset + this.marker.length))
+  }
+
+  private write(data: Uint8Array): void {
+    if (data.length > 0 && !this.output.destroyed) this.output.write(data)
+  }
+}
+
+async function waitForBootstrapOutput(
+  ready: Promise<void>,
+  completion: Promise<CommandResult>,
+  signal?: AbortSignal,
+): Promise<void> {
+  signal?.throwIfAborted()
+  await new Promise<void>((resolve, reject) => {
+    let settled = false
+    let removeAbort: (() => void) | undefined
+    const finish = (complete: () => void): void => {
+      if (settled) return
+      settled = true
+      removeAbort?.()
+      complete()
+    }
+    const onExit = (): void => {
+      finish(() => { reject(new Error('subprocess-e2b: terminal exited before publishing its output boundary')) })
+    }
+    if (signal !== undefined) {
+      const onAbort = (): void => {
+        finish(() => { reject(asError(signal.reason)) })
+      }
+      signal.addEventListener('abort', onAbort, { once: true })
+      removeAbort = () => { signal.removeEventListener('abort', onAbort) }
+    }
+    void ready.then(() => { finish(resolve) })
+    void completion.then(onExit, onExit)
+  })
+}
+
 function asError(error: unknown): Error {
  return error instanceof Error ? error : new Error(String(error))
 }
@@ -374,9 +445,12 @@ export async function spawnE2BTerminal(
    runner: posix.join(stateDir, 'runner.bash'),
    environment: posix.join(stateDir, 'environment'),
    argv: posix.join(stateDir, 'argv'),
+    outputMarker: posix.join(stateDir, 'output-marker'),
    ready: posix.join(stateDir, 'ready'),
  }
+  const outputMarker = Buffer.from(`dsh-e2b-bootstrap:${randomUUID()}`)
  const output = new PassThrough()
+  const outputFilter = new BootstrapOutputFilter(outputMarker, output)
  let handle: CommandHandle | undefined
  let completion: Promise<CommandResult> | undefined
  let stateDirectoryCreated = false
@@ -391,9 +465,10 @@ export async function spawnE2BTerminal(
      { path: paths.runner, data: TERMINAL_RUNNER_SOURCE },
      { path: paths.environment, data: environment },
      { path: paths.argv, data: argv },
+      { path: paths.outputMarker, data: outputMarker.toString('utf8') },
    ], signalOpts(spec.signal))
    await sandbox.commands.run(
-      `chmod 600 -- ${quoteE2BShellArg(paths.runner)} ${quoteE2BShellArg(paths.environment)} ${quoteE2BShellArg(paths.argv)}`,
+      `chmod 600 -- ${quoteE2BShellArg(paths.runner)} ${quoteE2BShellArg(paths.environment)} ${quoteE2BShellArg(paths.argv)} ${quoteE2BShellArg(paths.outputMarker)}`,
      signalOpts(spec.signal),
    )
    handle = await sandbox.pty.create({
@@ -403,7 +478,7 @@ export async function spawnE2BTerminal(
      envs: { TERM: 'dumb' },
      timeoutMs: 0,
      ...signalOpts(spec.signal),
-      onData: (data) => { if (!output.destroyed) output.write(Buffer.from(data)) },
+      onData: (data) => { outputFilter.push(data) },
    })
    completion = handle.wait()
    void completion.catch(() => {})
@@ -413,6 +488,7 @@ export async function spawnE2BTerminal(
    const command = `exec /bin/bash ${quoteE2BShellArg(paths.runner)} ${quoteE2BShellArg(stateDir)}\r`
    await sandbox.pty.sendInput(handle.pid, Buffer.from(command), signalOpts(spec.signal))
    await waitUntilReady(sandbox, paths, completion, spec.signal)
+    await waitForBootstrapOutput(outputFilter.ready, completion, spec.signal)
    const sessionId = await terminalSessionId(sandbox, handle.pid, spec.signal)
    return new E2BTerminalHandle(
      sandbox,