Reconnect stale periodic market streams

.changeset/periodic-stream-reconnect.md

@@ -0,0 +1,5 @@
+---
+"@imbingox/acex": patch
+---
+
+修复周期性行情流在业务 payload 停止推送但 WebSocket 仍保持 open 时的恢复行为。Binance funding / mark-price stream 超过配置的 stale 阈值后会主动重连并重放订阅；同一连接上的其它订阅仍会收到正常断线状态，不会被该业务 stale 恢复路径误吞。

.trellis/spec/sdk/adapters.md

@@ -323,12 +323,15 @@ adapter 不得自己构造 `AcexError` 或其他业务错误码——错误码
 
 - **职责划分**：
   - 通用核（`SubscriptionMultiplexer`）：按 `connectionKey` 池化物理连接、引用计数、重连后重放订阅、控制帧批量+限速、per-subscription ready / freshness fan-out、连接级 stale watchdog、最后一个订阅关闭时拆连接、单连接订阅数达上限时同 `connectionKey` 开新连接（连接池）。
-  - venue 策略（`VenueStreamProtocol`）：`subscriptionKey` / `connectionKey` / `connectionUrl` / `parseMessage` / `encodeSubscribe` / `encodeUnsubscribe` / `routeMessage(→ data|status|ack|ignore)`。所有交易所特定的 base URL、帧格式、消息路由判据都在这里，**不得**泄漏进通用核。
+  - venue 策略（`VenueStreamProtocol`）：`subscriptionKey` / `connectionKey` / `connectionUrl` / `parseMessage` / `encodeSubscribe` / `encodeUnsubscribe` / `routeMessage(→ data|status|ack|ignore)` / 可选 `livenessPolicy(descriptor)`。所有交易所特定的 base URL、帧格式、消息路由判据和 logical stream 活性语义都在这里，**不得**泄漏进通用核。
 - **logical stream 契约**：
   - `ready`：该订阅的 subscribe ACK / 等价订阅接受信号到达时 resolve；超 `initialMessageTimeoutMs` 未到则 reject 并清理该订阅。若首条 `data` 在 ACK 前到达，且 `routeMessage` 能给出匹配的 `subscriptionKey`，该 data 同时作为等价订阅接受信号、payload 和 freshness 来源。
   - `status` route：只分发给 `callbacks.onStatus`，不清首包 timer、不 resolve `ready`、不把订阅 freshness 切成 `fresh`。只用于真正非数据型 stream 状态；Deribit quote 的 bid-only / ask-only / empty 必须使用 `data` route。
   - `close()`：发出该订阅的 `UNSUBSCRIBE`；物理连接保持，直到其上最后一个订阅 close 才断开。幂等。
-  - stale：不做 per-subscription 独立 stale。只要同一物理连接仍有其他订阅收到消息，静默订阅保持 `fresh`；整条连接超过 `staleAfterMs` 无任何有效消息时，由连接级 watchdog 对该连接所有订阅触发 `onFreshnessChange("stale","heartbeat_timeout")`。
+  - stale 默认语义：未声明 business liveness policy 的订阅不做 per-subscription 独立 stale。只要同一物理连接仍有其他订阅收到消息，静默订阅保持 `fresh`；整条连接超过 `staleAfterMs` 无任何有效消息时，由连接级 watchdog 对该连接所有订阅触发 `onFreshnessChange("stale","heartbeat_timeout")`。
+  - business liveness policy：周期性业务帧 stream（例如 Binance funding/mark price）可以由 `VenueStreamProtocol.livenessPolicy()` 返回 `{ kind: "periodic" | "low_volume"; staleAfterMs: number; onStale: "mark_stale" | "reconnect" }`。`staleAfterMs` 必须由 venue policy 显式给出，不能在复用器里回落到连接级 watchdog 阈值；timer 只能在订阅已接受后启动（subscribe ACK 成功、无 ACK 协议的 subscribe frame 已发送、或首条 `data` 作为等价接受信号），并且只由 routed `data` payload 刷新；ACK、status、ignore、pong 和其它连接活性都不算该 logical subscription 的业务 payload。
+  - business stale 行为：policy timer 到期时只标记对应 logical subscription 的 subscriber stale，reason 固定为 `"heartbeat_timeout"`。`onStale:"reconnect"` 必须对当前 raw socket 触发 `ManagedWebSocketSession.restart()`，复用 close → reconnect → replay 路径；`restart()` 返回是否实际发起 raw close，只有返回 true 时才抑制触发订阅本身的 `onDisconnected()`，避免刚写入的 business `heartbeat_timeout` 立即被上层覆盖成 transport `ws_disconnected`。同一 socket 上的其它订阅仍应收到 `onDisconnected()`，因为它们经历的是底层 transport close。
+  - Binance policy：`fundingRate` descriptor 使用 `periodic + reconnect`，且把 adapter stream options 里的 `staleAfterMs` 显式写入 policy；`l1book` 和未声明 policy 的其它 stream 保持默认 stale 语义。private user-data stream 不使用行情级 business liveness policy，继续走 private adapter/listenKey/reconcile 的长周期恢复契约。
   - 断线：对该连接所有订阅 `onDisconnected()`（上层据此置 `ws_disconnected`），并**静默**置内部 freshness=stale（不再额外发 `heartbeat_timeout` freshness 事件）；重连 `open` 后自动重放全部活跃订阅，各订阅在各自首条消息回到 `fresh`。
 - **限额（保守口径，依据 Binance 官方）**：
   - 单连接 stream 上限：Binance 现货与 USDⓈ-M 合约**均为 1024**（Options 才是 200，不适用）。本仓库 `maxSubscriptionsPerConnection` 取保守值 **200**，到上限即开新连接。200 不是交易所硬限制，而是运营上限：平衡控制帧开销、per-subscription heartbeat/control 消息、路由与 fan-out 的 CPU/内存成本、高订阅数下的延迟/丢包观测、连接稳定性与重连限额，以及突发订阅的安全余量。后续可依据 `tests/soak/`（尤其 market L1 continuity）与线上 telemetry/packet-loss/latency 指标重新调优。
@@ -378,6 +381,10 @@ interface VenueHeartbeat {
 | `close()` 被外部调用两次 | 第二次 no-op | 抛 `ClosedError` 或重复清 timer |
 | 稳态时 WS 被对端关闭 | 触发 `onDisconnected` + 内部触发 ManagedWebSocket 重连 | 让 adapter 自行 `setTimeout` 重连 |
 | 整条连接长时间不收消息 | 连接级 watchdog 对该连接所有订阅触发 `onFreshnessChange("stale", "heartbeat_timeout")` | 单个订阅静默时自行标 stale 或主动 close 重连 |
+| periodic business stream 超过 `staleAfterMs` 没有 routed `data` | 只标记该 logical subscription `heartbeat_timeout`，按 policy 可 `restart()` raw socket 并 replay | 把 ACK/status/pong 当业务 payload，或用 `session.close()` 禁掉重连 |
+| business stale 触发主动 restart | 对触发订阅保留 reason 为 `heartbeat_timeout`；同 socket 其它订阅仍收到 `onDisconnected()` | 用 connection 级 suppression 吞掉无关订阅的 transport disconnect |
+| `ManagedWebSocketSession.restart()` 因 socket closing/closed no-op | 返回 false，复用器清掉本次 suppression 标记 | no-op 后留下 suppression，导致后续真实 close 被吞掉 |
+| 未声明 liveness policy 的订阅静默但同连接仍活跃 | 不启动 per-sub timer、不主动重连；继续依赖连接级 watchdog | 默认给所有 logical subscription 加短周期 business timeout |
 | 交易所要求应用层客户端 ping | 在 `VenueStreamProtocol.heartbeat` 声明文本帧、pong 判定、调度模式；由 ManagedWebSocket 统一调度 | adapter/multiplexer 外层自行 `setInterval` 发 ping |
 | 收到应用层 pong raw 帧 | 在 `parseMessage()` 前消费，清 pending pong、刷新连接活性，不路由给订阅 | 让 pong 进入 `routeMessage()` 或误判为 data/ack |
 | 应用层 ping 超过 `pongTimeoutMs` 未收到 pong | 对底层 raw socket 调 `close()`，复用 close→reconnect→replay 路径 | 调 session `close()` 禁掉重连，或直接在 adapter 手写重连 |
@@ -496,6 +503,7 @@ bun run test
 - 修改或新增 `fetchServerTime()` 时必须覆盖正常解析、HTTP 失败、无自动重试、缺失 / 非 number `serverTime`、限流顺序、单调时钟 RTT、manager 错误包装与不支持 venue。
 - 修改或新增 public market REST query（如 raw trades / funding history）时必须覆盖：endpoint/参数映射、`receivedAt` 采样、raw clone、manager canonical 化、unsupported venue/market、adapter failure error wrapping、rate-limit semantic plan。
 - 修改 Binance derivatives catalog 归一化时必须覆盖 `TRADIFI_PERPETUAL` fixture，断言其 symbol 为永续格式（如 `AAPL/USDT:USDT`）、`type:"swap"`、无 `expiry`，并能走 L1 book 与 funding/mark price 订阅。
+- 修改 `SubscriptionMultiplexer` business liveness policy 时，`tests/unit/subscription-multiplexer.test.ts` 必须覆盖：periodic subscription 业务帧 stale 后标记 `heartbeat_timeout` 并 raw socket restart/replay；未声明 policy 的订阅不因单订阅静默主动重连；无 ACK 协议在 subscribe frame 发送后启动 timer；业务 stale restart 不触发普通 `onDisconnected()` 覆盖上层 reason。`tests/unit/binance-stream-protocol.test.ts` 必须覆盖 Binance funding descriptor 返回 `periodic + reconnect`，L1 descriptor 不返回 policy；`tests/integration/market.test.ts` 必须覆盖 funding stale 后旧 socket 关闭、新 socket replay、收到新 mark price 后 snapshot 恢复 fresh。
 - `tests/integration/market.test.ts`、`tests/integration/account.test.ts`、`tests/integration/order.test.ts` 针对各 manager 的集成测试仍然过——这些测试通过 fake REST / fake WebSocket 间接验证 adapter contract。
 - 修改 `refreshAccount()` 或 account polling 时，必须覆盖 REST refresh 后 `risk.updated` 与 `position.updated` 都通过 public event/getter 可见。
 - 新增随行情变化的账户字段时，测试必须覆盖“没有 WS 消息、只有 refresh/polling”时字段仍会更新；同时覆盖 WS 断线期间 refresh 成功不会把 stream status 改成 healthy。

src/adapters/binance/adapter.ts

@@ -254,18 +254,24 @@ export class BinanceMarketAdapter implements MarketAdapter {
         BinanceStreamMessage,
         BinanceStreamDescriptor,
         BinanceStreamPayload
-      >(new BinanceStreamProtocol(), {
-        initialMessageTimeoutMs: config.initialMessageTimeoutMs,
-        staleAfterMs: config.staleAfterMs,
-        reconnectDelayMs: config.reconnectDelayMs,
-        reconnectMaxDelayMs: config.reconnectMaxDelayMs,
-        controlFrameMaxPerSec: BINANCE_CONTROL_FRAME_MAX_PER_SEC,
-        maxSubscriptionsPerConnection: BINANCE_MAX_SUBSCRIPTIONS_PER_CONNECTION,
-        now: config.now,
-        onReconnect: ({ descriptors }) => {
-          this.emitReconnectMetric(descriptors);
+      >(
+        new BinanceStreamProtocol({
+          fundingRateStaleAfterMs: config.staleAfterMs,
+        }),
+        {
+          initialMessageTimeoutMs: config.initialMessageTimeoutMs,
+          staleAfterMs: config.staleAfterMs,
+          reconnectDelayMs: config.reconnectDelayMs,
+          reconnectMaxDelayMs: config.reconnectMaxDelayMs,
+          controlFrameMaxPerSec: BINANCE_CONTROL_FRAME_MAX_PER_SEC,
+          maxSubscriptionsPerConnection:
+            BINANCE_MAX_SUBSCRIPTIONS_PER_CONNECTION,
+          now: config.now,
+          onReconnect: ({ descriptors }) => {
+            this.emitReconnectMetric(descriptors);
+          },
         },
-      });
+      );
       this.multiplexerConfig = config;
       return this.multiplexer;
     }

src/adapters/binance/stream-protocol.ts

@@ -1,5 +1,6 @@
 import type {
   EncodedVenueControlFrame,
+  StreamLivenessPolicy,
   VenueControlAck,
   VenueStreamProtocol,
 } from "../../internal/subscription-multiplexer.ts";
@@ -48,6 +49,10 @@ export type BinanceStreamPayload =
       readonly exchangeTs?: number;
     };
 
+interface BinanceStreamProtocolOptions {
+  readonly fundingRateStaleAfterMs: number;
+}
+
 const BINANCE_SPOT_WS_BASE_URL = "wss://stream.binance.com:9443/ws";
 const BINANCE_USDM_WS_BASE_URL = "wss://fstream.binance.com/ws";
 const BINANCE_USDM_MARKET_WS_BASE_URL = "wss://fstream.binance.com/market/ws";
@@ -94,6 +99,8 @@ export class BinanceStreamProtocol
 {
   private nextControlFrameId = 1;
 
+  constructor(private readonly options: BinanceStreamProtocolOptions) {}
+
   subscriptionKey(descriptor: BinanceStreamDescriptor): string {
     return `${descriptor.channel}:${descriptor.market.id}`;
   }
@@ -131,6 +138,20 @@ export class BinanceStreamProtocol
     return this.encodeControlFrame("UNSUBSCRIBE", descriptors);
   }
 
+  livenessPolicy(
+    descriptor: BinanceStreamDescriptor,
+  ): StreamLivenessPolicy | undefined {
+    if (descriptor.channel !== "fundingRate") {
+      return undefined;
+    }
+
+    return {
+      kind: "periodic",
+      staleAfterMs: this.options.fundingRateStaleAfterMs,
+      onStale: "reconnect",
+    };
+  }
+
   routeMessage(message: BinanceStreamMessage):
     | {
         kind: "data";

src/internal/managed-websocket.ts

@@ -62,6 +62,7 @@ export interface ManagedWebSocketOptions<TMessage> {
 export interface ManagedWebSocketSession {
   readonly ready: Promise<void>;
   send(data: string): void;
+  restart(reason?: string): boolean;
   close(): void;
 }
 
@@ -476,6 +477,19 @@ export function createManagedWebSocket<TMessage>(
 
       activeSocket.send(data);
     },
+    restart(reason = "websocket restart requested"): boolean {
+      if (
+        closed ||
+        !activeSocket ||
+        activeSocket.readyState === WebSocket.CLOSING ||
+        activeSocket.readyState === WebSocket.CLOSED
+      ) {
+        return false;
+      }
+
+      activeSocket.close(1000, reason);
+      return true;
+    },
     close() {
       if (closed) {
         return;