fix(web): quiesce websocket teardown

2026-08-04 16:42:50 +08:00
parent c6d0cbd8de
commit 7f3a2dae91
17 changed files with 124 additions and 43 deletions
--- a/packages/client/connection/README.i18n.yaml
+++ b/packages/client/connection/README.i18n.yaml
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/client/connection/README.md
-README.md: 11bfc950b7d4f09f1a3075e0f444966de841be70
-README.zh.md: 4b346de9e0dbc468d6b94546aa963a5b57b62127
+README.md: faf093964a740092983e13bf88f2cccd853c3e36
+README.zh.md: b06ab245dedbde13957aa416be044ef107b2753c
--- a/packages/client/connection/README.md
+++ b/packages/client/connection/README.md
@@ -10,7 +10,7 @@ The node half guards every entry under `/api` before bridging or upgrading (`src

 ## `/api` WebSocket downlinks

-`/api/events.mux` and `/api/events.host` each accept a WebSocket upgrade and send only the corresponding `ServerRequest` text messages to the browser; the client sends no application data over these sockets. If either socket ends, the current connection generation fails and rebuilds both streams; readiness still requires both sockets to be open and the `host.describe` HTTP call to succeed. Ordinary network GETs to these paths return 426 with no SSE fallback; `toFetchHandler`'s SSE codec serves only the isomorphic in-process carrier.
+`/api/events.mux` and `/api/events.host` each accept a WebSocket upgrade and send only the corresponding `ServerRequest` text messages to the browser; the client sends no application data over these sockets. If either socket ends, the current connection generation fails and rebuilds both streams; readiness still requires both sockets to be open and the `host.describe` HTTP call to succeed. Host teardown terminates both sockets, aborts their sources, and waits for source cleanup before returning. Ordinary network GETs to these paths return 426 with no SSE fallback; `toFetchHandler`'s SSE codec serves only the isomorphic in-process carrier.

 ## Keyless fixture

--- a/packages/client/connection/README.zh.md
+++ b/packages/client/connection/README.zh.md
@@ -10,7 +10,7 @@ node 半侧在桥接或 upgrade 前守卫 `/api` 下的每个入口（`src/api-r

 ## `/api` WebSocket 下行

-`/api/events.mux` 与 `/api/events.host` 各接受一条 WebSocket upgrade，并只向浏览器发送对应的 `ServerRequest` text message；客户端不会在这些 socket 上发送业务数据。任一 socket 结束都会使当前 connection generation 失败并重建两条流，连接就绪仍要求两条 socket open 且 `host.describe` HTTP 调用成功。普通网络 GET 这些路径会返回 426，不保留 SSE 回退；`toFetchHandler` 的 SSE 编解码只服务进程内同构载体。
+`/api/events.mux` 与 `/api/events.host` 各接受一条 WebSocket upgrade，并只向浏览器发送对应的 `ServerRequest` text message；客户端不会在这些 socket 上发送业务数据。任一 socket 结束都会使当前 connection generation 失败并重建两条流，连接就绪仍要求两条 socket open 且 `host.describe` HTTP 调用成功。Host teardown 会终止两条 socket、中止各自的 source，并等待 source 清理完成后再返回。普通网络 GET 这些路径会返回 426，不保留 SSE 回退；`toFetchHandler` 的 SSE 编解码只服务进程内同构载体。

 ## 无密钥 fixture

--- a/packages/client/connection/src/websocket-downlink.ts
+++ b/packages/client/connection/src/websocket-downlink.ts
@@ -50,6 +50,7 @@ function failureFrame(error: unknown): RpcRequest<Frame> {
 */
 export class WebSocketDownlinks {
  private readonly server = new WebSocketServer({ noServer: true })
+  private readonly pumps = new Set<Promise<void>>()

  /** @param api - host API supplying the typed event streams. */
  constructor(private readonly api: ApiProxy) {}
@@ -81,17 +82,18 @@ export class WebSocketDownlinks {
  }

  /**
-   * Terminate owned sockets and await the no-server acceptor's close.
-   * @returns A promise resolving after every accepted socket has closed.
+   * Terminate owned sockets and await the no-server acceptor plus frame pumps.
+   * @returns A promise resolving after every socket and source iterator stops.
   */
-  close(): Promise<void> {
+  async close(): Promise<void> {
    for (const socket of this.server.clients) socket.terminate()
-    return new Promise((resolve, reject) => {
+    await new Promise<void>((resolve, reject) => {
      this.server.close((error) => {
        if (error === undefined) resolve()
        else reject(error)
      })
    })
+    await Promise.all(this.pumps)
  }

  private upgrade<F extends Frame>(
@@ -107,7 +109,9 @@ export class WebSocketDownlinks {
      websocket.once('message', () => {
        websocket.close(1008, 'downlink only')
      })
-      void this.pump(websocket, open(abort.signal), abort)
+      const pump = this.pump(websocket, open(abort.signal), abort)
+      this.pumps.add(pump)
+      void pump.then(() => { this.pumps.delete(pump) })
    })
  }

--- a/packages/client/connection/tests/websocket-downlink.spec.ts
+++ b/packages/client/connection/tests/websocket-downlink.spec.ts
@@ -21,7 +21,9 @@ afterEach(async () => {

 function untilAbort(signal: AbortSignal): Promise<void> {
  if (signal.aborted) return Promise.resolve()
-  return new Promise(resolve => signal.addEventListener('abort', () => { resolve() }, { once: true }))
+  return new Promise((resolve) => {
+    signal.addEventListener('abort', () => { resolve() }, { once: true })
+  })
 }

 async function * idle<F>(signal: AbortSignal): AsyncGenerator<RpcRequest<F>> {
@@ -147,7 +149,7 @@ describe('WebSocket downlinks', () => {
    await once(socket, 'open')
    const closed = once(socket, 'close')
    socket.send('upstream payload')
-    const [code, reason] = await closed
+    const [code, reason] = await closed as [number, Buffer]
    expect(code).toBe(1008)
    expect(String(reason)).toBe('downlink only')
    await vi.waitFor(() => { expect(aborted).toBe(true) })
@@ -197,9 +199,9 @@ describe('WebSocket downlinks', () => {

  it('drops a source frame that races after the client has closed', async () => {
    let release!: () => void
-    const gate = new Promise<void>(resolve => { release = resolve })
+    const gate = new Promise<void>((resolve) => { release = resolve })
    let finish!: () => void
-    const finished = new Promise<void>(resolve => { finish = resolve })
+    const finished = new Promise<void>((resolve) => { finish = resolve })
    let sourceSignal: AbortSignal | undefined
    const downlinks = new WebSocketDownlinks(api(
      async function * (signal) {
@@ -227,7 +229,7 @@ describe('WebSocket downlinks', () => {

  it('contains socket send callback failures and closes the downlink', async () => {
    let release!: () => void
-    const gate = new Promise<void>(resolve => { release = resolve })
+    const gate = new Promise<void>((resolve) => { release = resolve })
    const downlinks = new WebSocketDownlinks(api(
      async function * () {
        await gate
@@ -262,4 +264,39 @@ describe('WebSocket downlinks', () => {
    await downlinks.close()
    await expect(downlinks.close()).rejects.toThrow('The server is not running')
  })
+
+  it('waits for source cleanup before teardown resolves', async () => {
+    let cleanupStarted!: () => void
+    const started = new Promise<void>((resolve) => { cleanupStarted = resolve })
+    let releaseCleanup!: () => void
+    const cleanupGate = new Promise<void>((resolve) => { releaseCleanup = resolve })
+    let cleaned = false
+    const downlinks = new WebSocketDownlinks(api(
+      async function * (signal) {
+        try {
+          await untilAbort(signal)
+        } finally {
+          cleanupStarted()
+          await cleanupGate
+          cleaned = true
+        }
+      },
+      idle,
+    ))
+    const host = await serve(downlinks)
+    const socket = new WebSocket(`${host.origin}${MUX_EVENTS_PATH}`)
+    await once(socket, 'open')
+    let closed = false
+    const closing = host.close().then(() => { closed = true })
+    try {
+      await started
+      expect(closed).toBe(false)
+      releaseCleanup()
+      await closing
+      expect(cleaned).toBe(true)
+    } finally {
+      releaseCleanup()
+      await closing
+    }
+  })
 })
--- a/packages/host/webserver/README.i18n.yaml
+++ b/packages/host/webserver/README.i18n.yaml
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/host/webserver/README.md
-README.md: f01c1a66b19e4f49b9cad31a6d41555aecb28168
-README.zh.md: 980bc3dbac4dac2e758e0043ead292fb5a42e674
+README.md: 196f350d87c5322cd3e9cda6e40587d35acd08c4
+README.zh.md: 0ae0470eab0aae2f6b539404621c611d95827977
--- a/packages/host/webserver/README.md
+++ b/packages/host/webserver/README.md
@@ -6,7 +6,7 @@ Web HTTP and upgrade-route registration plugin (default-exported `HttpServerServ

 The package knows no harness concepts: the `/api` HTTP bridge and downlink WebSockets are routes owned by the connection plugin, while plugin bundles and the HMR event stream are routes owned by the modules/hmr plugins. The upgrade handler owns the protocol handshake and connection contents; the webserver only delivers the raw socket and request. `host` accepts only `127.0.0.1` (default posture) and `0.0.0.0` (deliberate network exposure); `distIndex` is an assembly fact the composing app resolves and injects, never self-resolved (dist location is workspace knowledge of the app). Web (browser) shape only — Electron loads dist over `file://` and carries fetch over an IPC bridge, not this server. This package never prints; the URL line belongs to the shell.

-A listen failure (EADDRINUSE…) throws out of activation and rejects Loader composition with the bind diagnostic; the failed candidate fiber is disposed. An HTTP request whose handling throws (a malformed %-escape hitting `decodeURIComponent`, a client dropping mid-body) is answered 400 — or the socket destroyed when headers are already out — and logged as a warning; it never exits the process. An upgrade-handler exception is logged as a warning and destroys its socket. Disposal first calls `close()` and `closeAllConnections()`, then destroys upgraded sockets the webserver still tracks so they cannot hold teardown open.
+A listen failure (EADDRINUSE…) throws out of activation and rejects Loader composition with the bind diagnostic; the failed candidate fiber is disposed. An HTTP request whose handling throws (a malformed %-escape hitting `decodeURIComponent`, a client dropping mid-body) is answered 400 — or the socket destroyed when headers are already out — and logged as a warning; it never exits the process. An upgrade-handler exception or upgraded-socket transport error is logged as a warning and destroys its socket. Disposal starts `close()` and `closeAllConnections()`, destroys every tracked upgraded socket, and returns only after the HTTP server and those sockets have closed.

 In development, the client-plugin registry synchronously captures each built bundle's stat baseline before it returns, then polls those baselines and re-hashes changed content. Each rescan stages its candidate table, graph, and watch map before publishing them, so a baseline failure preserves the prior graph. An immediate rebuild therefore cannot disappear into an asynchronously established watch baseline; a rename window marks the path dirty, retains the last successful baseline, and forces a re-hash when the bundle reappears even with identical metadata.

--- a/packages/host/webserver/README.zh.md
+++ b/packages/host/webserver/README.zh.md
@@ -6,7 +6,7 @@ Web HTTP 与 upgrade route 注册插件（默认导出 `HttpServerService`，配

 该包不了解任何 harness 概念：`/api` HTTP 桥接与下行 WebSocket 是 connection 插件的 route，插件 bundle 与 HMR（热模块替换）事件流则是 modules／hmr 插件的 route。upgrade handler 拥有协议握手与连接内容；webserver 只交付原始 socket 与 request。`host` 只接受 `127.0.0.1`（默认姿态）和 `0.0.0.0`（有意向网络开放）；`distIndex` 是由组合应用解析并注入的组装事实，绝不会自行解析，因为 dist 位置属于应用的工作区知识。该服务器只服务 Web（浏览器）形态；Electron 通过 `file://` 加载 dist，并经 IPC 桥接承载 fetch，而不使用本服务器。该包从不打印内容；URL 行属于 shell。

-监听失败（EADDRINUSE……）会从激活过程抛出，以 bind 诊断使 Loader 组合 reject；失败的候选 fiber 会被 dispose（资源释放）。处理 HTTP 请求时抛错（例如格式错误的百分号转义传入 `decodeURIComponent`，或客户端在请求体传输中途断开）时，服务器会响应 400；若响应头已经发出，则销毁 socket，并记录 warning，但绝不会退出进程。upgrade handler 抛错会记录 warning 并销毁其 socket。资源释放会先调用 `close()` 与 `closeAllConnections()`，再销毁 webserver 仍跟踪的升级 socket，确保升级连接不会悬住 teardown。
+监听失败（EADDRINUSE……）会从激活过程抛出，以 bind 诊断使 Loader 组合 reject；失败的候选 fiber 会被 dispose（资源释放）。处理 HTTP 请求时抛错（例如格式错误的百分号转义传入 `decodeURIComponent`，或客户端在请求体传输中途断开）时，服务器会响应 400；若响应头已经发出，则销毁 socket，并记录 warning，但绝不会退出进程。upgrade handler 抛错或升级 socket 出现传输错误时，会记录 warning 并销毁对应 socket。资源释放会启动 `close()` 与 `closeAllConnections()`，销毁所有受跟踪的升级 socket，并仅在 HTTP server 与这些 socket 均已关闭后返回。

 在开发环境中，客户端插件注册表会在返回前同步捕获每个已构建 bundle 的 stat 基线，随后轮询这些基线，并在内容变化后重新计算哈希。每次重新扫描都会先暂存候选表、图和监听 map，再统一发布，因此基线失败会保留先前的图。这样，即时重建不会消失在异步建立的监听基线中；重命名窗口会把路径标记为脏，保留最近一次成功基线，并在 bundle 重新出现时强制重新计算哈希，即使其元数据完全相同也不例外。

--- a/packages/host/webserver/src/index.ts
+++ b/packages/host/webserver/src/index.ts
@@ -172,6 +172,15 @@ export class HttpServerService extends Service {
      })
    })
    this.server.on('upgrade', (req, socket, head) => {
+      const onError = (error: Error): void => {
+        this.ctx.logger.warn(error)
+        socket.destroy()
+      }
+      socket.on('error', onError)
+      socket.once('close', () => {
+        socket.off('error', onError)
+        this.upgradedSockets.delete(socket)
+      })
      let route: WebUpgradeRoute | undefined
      try {
        /* v8 ignore next -- node:http always sets url on server requests. */
@@ -186,7 +195,6 @@ export class HttpServerService extends Service {
        return
      }
      this.upgradedSockets.add(socket)
-      socket.once('close', () => { this.upgradedSockets.delete(socket) })
      try {
        Promise.resolve(route.handler(req, socket, head)).catch((error: unknown) => {
          this.ctx.logger.warn(error instanceof Error ? error : new Error(String(error)))
@@ -210,11 +218,17 @@ export class HttpServerService extends Service {

    // Node does not include upgraded sockets in closeAllConnections(), so the
    // service tracks and destroys them as part of the same ownership boundary.
-    this.ctx.effect(() => () => new Promise<void>((resolve) => {
-      this.server.close(() => { resolve() })
+    this.ctx.effect(() => async () => {
+      const serverClosed = new Promise<void>((resolve) => {
+        this.server.close(() => { resolve() })
+      })
      this.server.closeAllConnections()
-      for (const socket of this.upgradedSockets) socket.destroy()
-    }), 'httpServer.listen')
+      const upgradedClosed = [...this.upgradedSockets].map(socket => new Promise<void>((resolve) => {
+        socket.once('close', () => { resolve() })
+        socket.destroy()
+      }))
+      await Promise.all([serverClosed, ...upgradedClosed])
+    }, 'httpServer.listen')
  }

  /** Longest-prefix-wins over the prefix table after an exact-table miss. */
--- a/packages/host/webserver/tests/webserver.spec.ts
+++ b/packages/host/webserver/tests/webserver.spec.ts
@@ -87,7 +87,7 @@ async function upgrade(port: number, path: string): Promise<ReturnType<typeof co
    '',
    '',
  ].join('\r\n'))
-  const [data] = await response
+  const [data] = await response as [Buffer]
  expect(String(data)).toContain('101 Switching Protocols')
  return socket
 }
@@ -154,9 +154,11 @@ describe('real Loader composition', () => {
    // Upgrade routes match exact pathnames, reject duplicate ownership, and
    // become registrable again after disposal. The accepted socket stays open
    // so the teardown assertion also covers upgraded-connection ownership.
+    let upgradedServerClosed = false
    const disposeUpgrade = server.registerUpgrade({
      path: '/events',
      handler: (_req, socket) => {
+        socket.once('close', () => { upgradedServerClosed = true })
        socket.write('HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: dsh-test\r\n\r\n')
      },
    })
@@ -166,10 +168,34 @@ describe('real Loader composition', () => {
    disposeUpgrade()
    expect(() => server.registerUpgrade({ path: '/events', handler: () => {} })).not.toThrow()

+    // The webserver contains raw-socket errors even before an upgrade handler
+    // has installed its protocol implementation.
+    server.registerUpgrade({
+      path: '/upgrade-error',
+      handler: async (_req, socket) => {
+        await Promise.resolve()
+        socket.destroy(new Error('test upgrade transport failure'))
+      },
+    })
+    const failedUpgrade = connect(port, '127.0.0.1')
+    failedUpgrade.on('error', () => { /* The server-side reset is the fixture outcome. */ })
+    await once(failedUpgrade, 'connect')
+    const failedUpgradeClosed = once(failedUpgrade, 'close')
+    failedUpgrade.write([
+      'GET /upgrade-error HTTP/1.1',
+      `Host: 127.0.0.1:${String(port)}`,
+      'Connection: Upgrade',
+      'Upgrade: dsh-test',
+      '',
+      '',
+    ].join('\r\n'))
+    await failedUpgradeClosed
+    expect(await request(port, '/probe')).toMatchObject({ status: 200, body: 'EXACT' })
+
    // Teardown closes both ordinary and upgraded sockets before it resolves.
-    const upgradedClosed = once(upgraded, 'close')
    await loaded.fiber.dispose()
-    await upgradedClosed
+    expect(upgradedServerClosed).toBe(true)
+    upgraded.destroy()
    await expect(request(port, '/probe')).rejects.toThrow()
  })