Merge pull request #5975 from deepseek-harness/turtle/remote-mux-heartbeat-1689ca09e33b

fix(api-gateway): count mux traffic as heartbeat liveness

packages/api/gateway/README.i18n.yaml

@@ -15,8 +15,8 @@
   en: d9859219cdb0f367
   zh: 2a3cfa8cb786e327
 /deepseek-ai-dsh-api-gateway/host-service-typertgatewayservice-ctx-key-typertgateway:
-  en: 5179435c09004549
-  zh: c904462396531947
+  en: 2aabd26516a45771
+  zh: 5907f472a4d536df
 /deepseek-ai-dsh-api-gateway/client-service-clientremote-ctx-key-remote:
   en: c106f0ca3e547f84
   zh: 833e30d7151c41c7
@@ -27,8 +27,8 @@
   en: af5abeee32f2d93b
   zh: b320bb727a223a13
 /deepseek-ai-dsh-api-gateway/known-limitations-and-deferred-work:
-  en: 3ade2903f40a0a98
-  zh: 9af1d557fefdbd89
+  en: 5af0d863582fbc4f
+  zh: d87e4b97b579c456
 /deepseek-ai-dsh-api-gateway/known-limitations-and-deferred-work/dev-note:
   en: 10727471030aee01
   zh: e6bee82abd12127d

packages/api/gateway/README.md

@@ -32,7 +32,7 @@ The Host entry registers a trusted-host interceptor on Connection's shared `/api
 
 A cancellation-aware Remote method declares `signal: AbortSignal` as its final Host parameter. The signal is descriptor metadata rather than a wire argument: Connection supplies it to the Gateway, and the Gateway injects it after decoded business parameters. SRC recognizes the reserved final name, while strict generation additionally requires the global `AbortSignal` type.
 
-A stream Remote uses `@Remote({ mode: 'stream' })` and returns an `Iterable` or `AsyncIterable`. `ctx.typertGateway.stream()` applies the same endpoint, argument, lookup, and cancellation checks as unary invocation, then returns a cancellation-aware iterable over the business items. The Client opens the Gateway-owned `/api/remote.mux` WebSocket when its plugin activates and keeps it connected while idle. Connection owns the retry schedule; before each retry it asks the mux to cancel any candidate or active socket and make exactly one fresh physical attempt. The Host sends Ping control frames at the configured `websocketHeartbeatIntervalMs` interval (two seconds by default), and the browser answers Pong at the WebSocket protocol layer, so idle network intermediaries see traffic without any Remote stream frame. A socket that has not answered the previous Ping is terminated at the next interval. Independently cancellable logical streams share that socket; an in-process Connection carrier provides equivalent streams directly without opening it.
+A stream Remote uses `@Remote({ mode: 'stream' })` and returns an `Iterable` or `AsyncIterable`. `ctx.typertGateway.stream()` applies the same endpoint, argument, lookup, and cancellation checks as unary invocation, then returns a cancellation-aware iterable over the business items. The Client opens the Gateway-owned `/api/remote.mux` WebSocket when its plugin activates and keeps it connected while idle. Connection owns the retry schedule; before each retry it asks the mux to cancel any candidate or active socket and make exactly one fresh physical attempt. The Host sends Ping control frames at the configured `websocketHeartbeatIntervalMs` interval (two seconds by default), and the browser answers Pong at the WebSocket protocol layer, so idle network intermediaries see traffic without any Remote stream frame. A Pong, any received frame, or a downlink write that waited in the full local send buffer and then completed counts as peer activity; a socket with no activity across two Ping intervals is terminated, so a congested link that delays the Pong behind data stays connected. Independently cancellable logical streams share that socket; an in-process Connection carrier provides equivalent streams directly without opening it.
 
 When the launcher supplies `ctx.appReady`, Gateway registers the WebSocket upgrade route only after successful application startup. Earlier connection attempts remain carrier failures handled by Connection's retry policy, so a restarting Host cannot accept streams while its controllers are still initializing. Unloading cancels a pending readiness subscription and closes any registered carrier. Embedded Hosts without `appReady` register immediately and own their startup ordering; in-process invocation is unchanged.
 
@@ -86,7 +86,8 @@ No direct effect; invoked business Services own any model-visible result.
 - `$stream()` supervises carrier replacement but does not infer replay semantics; each domain owns its resume cursor or replacement-baseline validation and normal-end classification. Connection generations reopen the internal `$events` stream; one-way notifications are not replayed, while pending scoped waterfalls retain their event id across replay.
 - Lookup resolvers are configured per key; an individual Remote parameter or endpoint cannot currently select a live-only policy under the same `agent`/`session` key.
 - Forwarded events reach `$on` without business-payload projection or redaction. Ordinary notifications are not replayed after reconnect; Agent-scoped waterfalls project only the top-level Agent identity needed to select the Client Context and carry their own pending lifetime.
-- `websocketHeartbeatIntervalMs` is both the Ping cadence and the Pong deadline. The Host terminates a peer that does not answer before the next interval, so a deployment whose event loop or network can stall longer than this interval must raise it.
+- `websocketHeartbeatIntervalMs` is both the Ping cadence and the activity deadline. The Host terminates a peer that shows no activity for two intervals, so a deployment whose event loop or network can stall longer than this interval must raise it.
+- A completed queued write shows that the TCP peer acknowledged data. Behind a buffering reverse proxy that peer is the proxy, so a vanished Client during sustained downlink is detected only after the proxy stops accepting data.
 - Uplinks have no flow control beyond the bounded Host inbox: a Client that sends faster than the method reads, or that sends to a method never taking its uplink, fails its stream with `gateway/uplink-overflow`, and no uplink item is replayed across carrier generations; a domain that must resume an uplink carries its own acknowledgement cursor in the reopened request.
 
 

packages/api/gateway/README.zh.md

@@ -32,7 +32,7 @@ Connection 可用时，Host 入口会在 Connection 共享的 `/api` FetchHandle
 
 支持取消的 Remote 方法会把 `signal: AbortSignal` 声明为最后一个 Host 参数。signal 是 descriptor 元数据，而不是 wire 参数：Connection 将它提供给 Gateway，Gateway 则在已解码的业务参数之后注入它。SRC 识别这个保留的末位参数名，严格生成还要求它具有全局 `AbortSignal` 类型。
 
-流式 Remote 使用 `@Remote({ mode: 'stream' })` 并返回 `Iterable` 或 `AsyncIterable`。`ctx.typertGateway.stream()` 执行与一元调用相同的 endpoint、参数、lookup 和取消校验，再返回可取消的业务项 iterable。Client 插件激活时打开 Gateway 自有的 `/api/remote.mux` WebSocket，并让它在空闲时保持连接。Connection 拥有重试调度；每次 retry 前，它要求 mux 取消候选或活动 socket，并且只做一次全新的物理连接尝试。Host 按配置的 `websocketHeartbeatIntervalMs` 间隔（默认 2 秒）发送 Ping 控制帧，浏览器在 WebSocket 协议层自动回复 Pong，使空闲网络中间层持续看到流量，而不新增 Remote 流帧。若 socket 尚未回复上一次 Ping，Host 会在下一间隔终止它。可独立取消的逻辑流共享这条连接；进程内 Connection 载体直接提供等价的流，不打开该 WebSocket。
+流式 Remote 使用 `@Remote({ mode: 'stream' })` 并返回 `Iterable` 或 `AsyncIterable`。`ctx.typertGateway.stream()` 执行与一元调用相同的 endpoint、参数、lookup 和取消校验，再返回可取消的业务项 iterable。Client 插件激活时打开 Gateway 自有的 `/api/remote.mux` WebSocket，并让它在空闲时保持连接。Connection 拥有重试调度；每次 retry 前，它要求 mux 取消候选或活动 socket，并且只做一次全新的物理连接尝试。Host 按配置的 `websocketHeartbeatIntervalMs` 间隔（默认 2 秒）发送 Ping 控制帧，浏览器在 WebSocket 协议层自动回复 Pong，使空闲网络中间层持续看到流量，而不新增 Remote 流帧。Pong、收到的任意帧，或曾在已满的本机发送缓冲区中等待后完成的下行写入，都视为对端活动；连续两个 Ping 间隔没有活动的 socket 会被终止，因此 Pong 被数据阻塞在拥塞链路上时连接仍会保持。可独立取消的逻辑流共享这条连接；进程内 Connection 载体直接提供等价的流，不打开该 WebSocket。
 
 启动器提供 `ctx.appReady` 时，Gateway 仅在应用成功启动后注册 WebSocket 升级路由。此前的连接尝试仍属于载体故障，由 Connection 的重试策略处理，因此重启中的 Host 不会在控制器仍在初始化时接受流。卸载会取消尚未触发的就绪订阅，并关闭已注册的载体。不提供 `appReady` 的嵌入式 Host 会立即注册，并自行负责启动顺序；进程内调用不变。
 
@@ -86,7 +86,8 @@ Client waterfall 的 Context 解析保持同步。解析器可以返回借用的
 - `$stream()` 监督载体替换，但不推断回放语义；各领域自行拥有恢复 cursor 或替换 baseline 的校验，以及正常结束的分类。Connection generation 会重开内部 `$events` 流；单向通知不会重放，仍处于 pending 的 scoped waterfall 则沿用同一个 event id 重放。
 - lookup 解析器按 key 配置；当前无法让单个 Remote 参数或 endpoint 在同一 `agent`/`session` key 下选择 live-only 策略。
 - 被转发的事件到达 `$on` 时不做业务载荷投影或脱敏。普通通知在重连后不重放；Agent-scoped waterfall 只投影选择 Client Context 所需的顶层 Agent 身份，并自行携带 pending 生命周期。
-- `websocketHeartbeatIntervalMs` 同时是 Ping 周期和 Pong 截止时间。对端未在下一周期前回复时，Host 会终止连接；如果部署的事件循环或网络可能停顿超过该间隔，必须调大此配置。
+- `websocketHeartbeatIntervalMs` 同时是 Ping 周期和活动截止时间。对端连续两个周期没有活动时，Host 会终止连接；如果部署的事件循环或网络可能停顿超过该间隔，必须调大此配置。
+- 排队写入完成只说明 TCP 对端确认了数据。经过带缓冲的反向代理时，该对端是代理，因此持续下行期间 Client 消失，要等代理不再接收数据后才会被检测到。
 - 上行除了有界的 Host inbox 之外没有流控：Client 发送快于方法读取，或发给从未取用 uplink 的方法时，其流以 `gateway/uplink-overflow` 失败；上行项不会跨载体代际重放，需要恢复上行的领域在重开的请求里自带确认游标。
 
 

packages/api/gateway/src/stream-server.ts

@@ -72,12 +72,12 @@ export class RemoteStreamMuxServer {
     this.server.handleUpgrade(req, socket, head, (websocket) => {
       const release = bindPeer(websocket, peer)
       if (release === undefined) return
-      this.missedHeartbeats.set(websocket, 0)
-      websocket.on('pong', () => { this.missedHeartbeats.set(websocket, 0) })
+      const alive = (): void => { this.missedHeartbeats.set(websocket, 0) }
+      alive()
       this.startHeartbeat()
       const bound: BoundStreamOpener = (endpoint, payload, uplink, control) =>
         this.open(endpoint, payload, uplink, peer, control)
-      const connection = new RemoteStreamMuxConnection(websocket, bound, this.failure, this.streamInboxBytes)
+      const connection = new RemoteStreamMuxConnection(websocket, bound, this.failure, this.streamInboxBytes, alive)
       const done = connection.run()
       this.connections.add(done)
       void done.then(() => {
@@ -141,13 +141,23 @@ class RemoteStreamMuxConnection {
     private readonly open: BoundStreamOpener,
     private readonly failure: RemoteStreamFailureMapper,
     private readonly streamInboxBytes: number,
+    private readonly alive: () => void,
   ) {}
 
+  /**
+   * Serve frames until the socket closes. A Pong, any received frame, and a
+   * downlink write that waited in the local send buffer and then completed
+   * count as peer activity, so a congested link that delays the Pong behind
+   * data keeps its socket. A write that completes at once proves nothing
+   * about the peer and does not count.
+   */
   async run(): Promise<void> {
     const closed = new Promise<void>((resolve) => {
       this.socket.once('close', resolve)
       this.socket.once('error', () => { this.socket.terminate() })
+      this.socket.on('pong', this.alive)
       this.socket.on('message', (data, isBinary) => {
+        this.alive()
         if (isBinary) {
           this.socket.close(1003, 'text messages required')
           return
@@ -283,10 +293,18 @@ class RemoteStreamMuxConnection {
         reject(new Error('api gateway: Remote stream socket is closed'))
         return
       }
+      let queued = false
       this.socket.send(text, (error) => {
-        if (error) reject(error)
-        else resolve()
+        if (error) {
+          reject(error)
+          return
+        }
+        if (queued) this.alive()
+        resolve()
       })
+      // Read after send(): bytes still buffered mean the local send buffer was
+      // full, so their completion requires the peer to acknowledge data.
+      queued = this.socket.bufferedAmount > 0
     }))
     this.writes = delivery.catch(() => undefined)
     return delivery

packages/api/gateway/tests/stream-server.host.spec.ts

@@ -67,6 +67,54 @@ describe('Remote stream mux server carrier lifecycle', () => {
     await closed
   })
 
+  it('counts inbound frames as liveness while Pong is delayed', async () => {
+    const entry = await startMux(async (_endpoint, _payload, _uplink, _peer, control) => waitForAbort(control.signal), 20)
+    const client = await connect(entry.url, false)
+    const terminated = vi.spyOn(acceptedSocket(entry.mux), 'terminate')
+    const traffic = setInterval(() => { client.send(JSON.stringify({ type: 'cancel', streamId: 'absent' })) }, 5)
+    try {
+      await pings(client, 4)
+      expect(terminated).not.toHaveBeenCalled()
+    } finally {
+      clearInterval(traffic)
+    }
+    await vi.waitFor(() => { expect(terminated).toHaveBeenCalledOnce() })
+  })
+
+  it('counts a downlink write that waited in the send buffer as liveness', async () => {
+    const stop = Promise.withResolvers<undefined>()
+    const entry = await startMux(async () => steadyItems(stop.promise), 20)
+    const client = await connect(entry.url, false)
+    const serverSocket = acceptedSocket(entry.mux)
+    const terminated = vi.spyOn(serverSocket, 'terminate')
+    Object.defineProperty(serverSocket, 'bufferedAmount', { configurable: true, value: 1 })
+    client.send(openFrame('congested'))
+    try {
+      await pings(client, 4)
+      expect(terminated).not.toHaveBeenCalled()
+    } finally {
+      stop.resolve(undefined)
+      Reflect.deleteProperty(serverSocket, 'bufferedAmount')
+    }
+    await vi.waitFor(() => { expect(terminated).toHaveBeenCalledOnce() })
+  })
+
+  it('does not count a downlink write that completed at once as liveness', async () => {
+    const stop = Promise.withResolvers<undefined>()
+    const entry = await startMux(async () => steadyItems(stop.promise), 20)
+    const client = await connect(entry.url, false)
+    const terminated = vi.spyOn(acceptedSocket(entry.mux), 'terminate')
+    const items = vi.fn()
+    client.on('message', items)
+    client.send(openFrame('flowing'))
+    try {
+      await vi.waitFor(() => { expect(terminated).toHaveBeenCalledOnce() })
+      expect(items).toHaveBeenCalled()
+    } finally {
+      stop.resolve(undefined)
+    }
+  })
+
   it('keeps the socket when a delayed Pong arrives before the final check', async () => {
     const entry = await startMux(async (_endpoint, _payload, _uplink, _peer, control) => waitForAbort(control.signal), 20)
     const client = await connect(entry.url, false)
@@ -585,6 +633,19 @@ async function *echoUplink(uplink: AsyncIterable<unknown>): AsyncIterable<unknow
   for await (const value of uplink) yield value
 }
 
+/** Yield an item every few milliseconds until `stop` settles. */
+async function *steadyItems(stop: Promise<undefined>): AsyncIterable<string> {
+  const stopped = stop.then(() => true)
+  while (!await Promise.race([stopped, new Promise<false>((resolve) => { setTimeout(() => { resolve(false) }, 5) })])) {
+    yield 'item'
+  }
+}
+
+/** Resolve after the client has received `count` Ping frames. */
+async function pings(client: WebSocket, count: number): Promise<void> {
+  for (let received = 0; received < count; received++) await once(client, 'ping')
+}
+
 async function *waitForAbort(signal: AbortSignal): AsyncIterable<never> {
   await new Promise<void>((resolve) => {
     if (signal.aborted) resolve()