时间线：
────────●─────────────✂─────────────●────────▶
      t0 断线          （30s 黑洞）      t1 重连成功

子问题 1（知道补哪段）：客户端断线前收到的最后一条数据的时间戳 lastTs —— "游标"
子问题 2（历史从哪来）：服务器必须留存历史数据，按游标查出缺口区间
子问题 3（如何衔接）  ：补传数据 + 实时数据可能重叠/乱序 → 去重 + 排序

/**
 * 协议消息全集（本周最终版）
 */
export type WsMessage =
  // 实时数据（服务器 → 客户端，每秒）
  | { type: "snapshot"; payload: PanelSnapshot; ts: number }
  // 服务器在 snapshot 里附带的最新序号（客户端据此更新游标）
  | { type: "snapshot"; seq: number; payload: PanelSnapshot }
  // 补传请求（客户端 → 服务器，重连成功后发）
  | { type: "backfill"; fromSeq: number }
  // 补传响应（服务器 → 客户端，一批历史数据 + 是否还有更多）
  | { type: "backfill-data"; fromSeq: number; snapshots: PanelSnapshot[]; hasMore: boolean }
  // 心跳
  | { type: "ping"; seq: number }
  | { type: "pong"; seq: number };

时间戳游标的问题：
- 服务器和客户端时钟不同步 → 边界判断误差
- 同一毫秒多条消息 → 边界重复或遗漏

序号游标：
- 服务器单调整数，天然严格全序，无歧义
- 时间戳仍保留在 payload 里（显示用），游标职责分离

环形缓冲（定长数组，写满覆盖最旧）：
容量 600 条（10 分钟 @ 1条/秒）

push 序列： n-3, n-2, n-1, n, [覆盖最旧] ...
读指针逻辑：backfill(fromSeq) → 从缓冲里找 seq >= fromSeq 的连续段

/**
 * 定长环形缓冲：服务器留存最近 N 条快照
 * 写满覆盖最旧——内存恒定，历史窗口有限（10 分钟内可补）
 */
export class RingBuffer<T extends { seq: number }> {
  private items: T[] = [];
  constructor(private capacity: number) {}

  /** 追加（自动覆盖最旧，同时维护 seq 单调） */
  push(item: T): void {
    this.items.push(item);
    if (this.items.length > this.capacity) {
      this.items.shift();   // 容量小（百级）时 shift 足够；万级以上用首尾指针法
    }
  }

  /**
   * 查询 seq 大于 fromSeq 的连续段（补传核心）
   * @returns 数据 + 起始游标是否还在缓冲内（太旧的缺口无法补全）
   */
  queryAfter(fromSeq: number, limit: number): { items: T[]; hasMore: boolean; gapLost: boolean } {
    const idx = this.items.findIndex((it) => it.seq > fromSeq);
    if (idx === -1) {
      // 两种情况合并处理：请求的游标太新（没有数据要补）或太旧（缓冲头之后）
      return { items: [], hasMore: false, gapLost: fromSeq < (this.items[0]?.seq ?? 0) - 1 };
    }
    const items = this.items.slice(idx, idx + limit);
    const hasMore = idx + limit < this.items.length;
    return { items, hasMore, gapLost: false };
  }

  /** 缓冲内最旧序号（客户端可据此判断缺口是否可补） */
  get oldestSeq(): number {
    return this.items[0]?.seq ?? 0;
  }
}

import { WebSocketServer, WebSocket } from "ws";
import { RingBuffer } from "./ring-buffer";

const wss = new WebSocketServer({ port: 8080, path: "/sensor-feed" });
const history = new RingBuffer<{ seq: number; payload: PanelSnapshot }>(600);
let seq = 0;

wss.on("connection", (ws: WebSocket) => {
  // 实时推送（每秒）—— 同时写入历史缓冲
  const timer = setInterval(() => {
    if (ws.readyState !== WebSocket.OPEN) return;
    const payload = generateSnapshot();          // 模拟数据生成
    const item = { seq: ++seq, payload };
    history.push(item);
    ws.send(JSON.stringify({ type: "snapshot", seq: item.seq, payload }));
  }, 1000);

  ws.on("message", (raw) => {
    const msg = JSON.parse(raw.toString());

    if (msg.type === "backfill") {
      // 补传：按游标回放历史（分批，每批最多 120 条防大帧卡连接）
      const { items, hasMore, gapLost } = history.queryAfter(msg.fromSeq, 120);
      ws.send(JSON.stringify({
        type: "backfill-data",
        fromSeq: msg.fromSeq,
        snapshots: items,
        hasMore,
        gapLost,     // true 时客户端应提示"部分历史已不可恢复"
      }));
    }
  });

  ws.on("close", () => clearInterval(timer));
});

        ┌──────────────────────────────────────────┐
        │ open ──正常收 snapshot──▶ 更新游标 lastSeq  │
        │   │                                       │
        │   │ 断线（心跳判死/1006）                    │
        │   ▼                                       │
        │ reconnecting ──重连成功──▶ 恢复中           │
        │                              │            │
        │                    发 backfill(lastSeq)    │
        │                              │            │
        │                    收 backfill-data        │
        │                      ├── hasMore → 继续请求 │
        │                      └── 完成 ──▶ open      │
        └──────────────────────────────────────────┘

/**
 * 会话恢复管理器：叠在 ReconnectingWebSocket 之上
 * 职责：游标维护、补传请求、补传数据投递
 */
export class SessionManager {
  private lastSeq = 0;                    // 游标：最后收到的快照序号
  private restoring = false;              // 恢复中标记（防补传/实时交错乱序）

  constructor(private conn: ReconnectingWebSocket) {
    // Day 58 预留的钩子今天派上用场
    conn.onSessionRestored = () => this.requestBackfill();

    conn.onMessage((data) => {
      const msg = JSON.parse(data);
      switch (msg.type) {
        case "snapshot":
          if (!this.restoring) {
            // 正常态：直接投递 + 推进游标
            this.lastSeq = msg.seq;
            this.onSnapshot?.(msg.payload);
          } else {
            // 恢复中收到的实时数据：暂存，等补传完成后统一排序投递（见 4.3）
            this.pendingDuringRestore.push(msg);
          }
          break;

        case "backfill-data":
          this.handleBackfill(msg);
          break;
      }
    });
  }

  /** 恢复入口：请求从游标之后的数据 */
  private requestBackfill(): void {
    if (this.lastSeq === 0) return;   // 从未收到过数据（首连）——无需补传
    this.restoring = true;
    this.pendingDuringRestore = [];
    this.conn.send(JSON.stringify({ type: "backfill", fromSeq: this.lastSeq }));
  }

  /** 处理补传批次 */
  private handleBackfill(msg: {
    snapshots: { seq: number; payload: PanelSnapshot }[];
    hasMore: boolean;
    gapLost: boolean;
  }): void {
    if (msg.gapLost) {
      console.warn("部分历史已超出服务器缓冲窗口，无法完整补传");
    }
    // 补传数据按序投递 + 推进游标
    for (const s of msg.snapshots) {
      if (s.seq <= this.lastSeq) continue;   // 防重复（见第五节）
      this.lastSeq = s.seq;
      this.onSnapshot?.(s.payload);
    }
    if (msg.hasMore) {
      // 还有更多：继续翻页请求
      this.conn.send(JSON.stringify({ type: "backfill", fromSeq: this.lastSeq }));
    } else {
      // 补传完成：把恢复期间暂存的实时数据排序补投
      this.pendingDuringRestore
        .sort((a, b) => a.seq - b.seq)
        .forEach((m) => {
          if (m.seq > this.lastSeq) {
            this.lastSeq = m.seq;
            this.onSnapshot?.(m.payload);
          }
        });
      this.restoring = false;
    }
  }

  /** 快照回调（面板层订阅这个） */
  onSnapshot: ((s: PanelSnapshot) => void) | null = null;

  private pendingDuringRestore: { seq: number; payload: PanelSnapshot }[] = [];
}

1. 补传边界：backfill(fromSeq) 的语义是"严格大于 fromSeq"——若语义含糊（>=），边界重复
2. 恢复期间暂存的实时消息，可能与补传批次尾部重叠
3. （MQTT 场景）QoS 1 重发

// 核心一行：只接受比游标新的数据
if (s.seq <= this.lastSeq) continue;

// ❌ 错误思路：用 Set<seq> 记录"见过的序号"
//    内存无限增长 + 恢复后集合膨胀——去重要利用"单调递增"这个结构性质

1. 启动服务器 + 面板，正常收数（游标推进）
2. DevTools → Network → Offline（或杀服务器进程）
3. 观察面板：状态灯变黄（reconnecting）、曲线停在断线时刻
4. 等待 30 秒（期间服务器数据继续进环形缓冲）
5. 恢复网络（或重启服务器）
6. 验收点：
   a. 状态灯变绿
   b. 曲线上断线缺口被数据填平（无空白、无打结）
   c. 控制台无重复 seq 投递
7. 极限测试：断线 15 分钟（超过缓冲窗口 10 分钟）
   → 面板提示"部分历史不可恢复"，曲线留一小段空白但实时流正常继续