《互联网产品的客户端架构·内容、社交与实时互动》7.2 消息去重与有序：端上能保证什么

// file: src/reliability/ordering.ts
// tier: L1
// verify: node --test test/reliability/ordering.test.ts

export interface SequencedMessage {
  id: string;
  seq: number;
}

export type IngestStatus = "applied" | "buffered" | "duplicate" | "late" | "conflict" | "gap";

export interface IngestResult {
  status: IngestStatus;
  applied: SequencedMessage[];
  buffered: number;
  expected: number;
}

export interface OrderingOptions {
  initialSeq?: number;
  window?: number;
  seenLimit?: number;
}

export class OrderingGate {
  #lastSeq: number;
  #window: number;
  #seenLimit: number;
  #seen = new Set<string>();
  #seenOrder: string[] = [];
  #pending = new Map<number, SequencedMessage>();
  #appliedOrder: number[] = [];

  constructor(options: OrderingOptions = {}) {
    const initialSeq = options.initialSeq ?? 0;
    const window = options.window ?? 64;
    const seenLimit = options.seenLimit ?? 4096;
    if (!Number.isInteger(initialSeq) || initialSeq < 0) throw new Error("initialSeq 必须是非负整数");
    if (!Number.isInteger(window) || window < 1) throw new Error("window 必须是正整数");
    if (!Number.isInteger(seenLimit) || seenLimit < 1) throw new Error("seenLimit 必须是正整数");
    this.#lastSeq = initialSeq;
    this.#window = window;
    this.#seenLimit = seenLimit;
  }

  get lastSeq(): number {
    return this.#lastSeq;
  }

  get pendingCount(): number {
    return this.#pending.size;
  }

  appliedSeqs(): number[] {
    return [...this.#appliedOrder];
  }

  ingest(message: SequencedMessage): IngestResult {
    assertMessage(message);
    const expected = this.#lastSeq + 1;
    if (this.#seen.has(message.id)) {
      return this.#result("duplicate", [], expected);
    }
    if (message.seq <= this.#lastSeq) {
      // 序列号已经越过它：即使 id 没见过，也不能插回已应用的历史位置。
      this.#remember(message.id);
      return this.#result("late", [], expected);
    }
    const maxAhead = this.#lastSeq + this.#window;
    if (message.seq > maxAhead) {
      // 超出窗口说明中间缺得太多，交给补拉逻辑，不在本地无限缓存。
      return this.#result("gap", [], expected);
    }
    if (message.seq > expected) {
      const existing = this.#pending.get(message.seq);
      if (existing) {
        return this.#result(existing.id === message.id ? "duplicate" : "conflict", [], expected);
      }
      this.#pending.set(message.seq, message);
      this.#remember(message.id);
      return this.#result("buffered", [], expected);
    }
    this.#remember(message.id);
    const applied: SequencedMessage[] = [message];
    this.#lastSeq = message.seq;
    this.#appliedOrder.push(message.seq);
    this.#trimHistory();
    while (this.#pending.has(this.#lastSeq + 1)) {
      const next = this.#pending.get(this.#lastSeq + 1) as SequencedMessage;
      this.#pending.delete(next.seq);
      applied.push(next);
      this.#lastSeq = next.seq;
      this.#appliedOrder.push(next.seq);
      this.#trimHistory();
    }
    return this.#result("applied", applied, expected);
  }

  #remember(id: string): void {
    if (this.#seen.has(id)) return;
    this.#seen.add(id);
    this.#seenOrder.push(id);
    while (this.#seenOrder.length > this.#seenLimit) {
      const oldest = this.#seenOrder.shift() as string;
      this.#seen.delete(oldest);
    }
  }

  #trimHistory(): void {
    if (this.#appliedOrder.length > 1024) this.#appliedOrder.shift();
  }

  #result(status: IngestStatus, applied: SequencedMessage[], expected: number): IngestResult {
    return { status, applied, buffered: this.#pending.size, expected };
  }
}

function assertMessage(message: SequencedMessage): void {
  if (typeof message.id !== "string" || message.id.length === 0) throw new Error("id 不能为空");
  if (!Number.isInteger(message.seq) || message.seq < 1) throw new Error("seq 必须是正整数");
}

$ node --test test/reliability/ordering.test.ts
✔ 顺序到达的消息立即应用 (0.943167ms)
✔ 重复 id 不推进序列号 (0.081792ms)
✔ 窗口内乱序先缓冲，缺口补齐后按序应用 (0.107167ms)
✔ 序列号已越过的陌生消息标记为迟到 (0.071791ms)
✔ 超出缓冲窗口的消息标记为空洞 (0.552375ms)
✔ 窗口边界内缓冲，超出一条即判空洞 (0.12575ms)
✔ 同一序列号被不同 id 占用时判定冲突 (0.095084ms)
✔ 去重集合有上限，淘汰后的旧消息仍被判迟到 (0.072417ms)
ℹ tests 8
ℹ suites 0
ℹ pass 8
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 157.427125