《互联网产品的客户端架构·内容、社交与实时互动》7.3 断线期间的消息补拉与空洞检测

// file: src/reliability/gap.ts
// tier: L2
// verify: node --test --test-concurrency=1 test/reliability/gap.test.ts

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

export interface SeqRange {
  from: number;
  to: number;
}

export type GapIngestStatus = "applied" | "buffered" | "duplicate" | "late";

export interface GapIngestResult {
  status: GapIngestStatus;
  applied: SequencedMessage[];
  gap: SeqRange | null;
}

export interface GapTrackerOptions {
  startSeq?: number;
  seenLimit?: number;
  historyLimit?: number;
}

export class GapTracker {
  #highWater: number;
  #seenLimit: number;
  #historyLimit: number;
  #seen = new Set<string>();
  #seenOrder: string[] = [];
  #buffered = new Map<number, SequencedMessage>();
  #appliedOrder: number[] = [];

  constructor(options: GapTrackerOptions = {}) {
    const startSeq = options.startSeq ?? 0;
    const seenLimit = options.seenLimit ?? 4096;
    const historyLimit = options.historyLimit ?? 1024;
    if (!Number.isInteger(startSeq) || startSeq < 0) throw new Error("startSeq 必须是非负整数");
    if (!Number.isInteger(seenLimit) || seenLimit < 1) throw new Error("seenLimit 必须是正整数");
    if (!Number.isInteger(historyLimit) || historyLimit < 1) throw new Error("historyLimit 必须是正整数");
    this.#highWater = startSeq;
    this.#seenLimit = seenLimit;
    this.#historyLimit = historyLimit;
  }

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

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

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

  ingest(message: SequencedMessage): GapIngestResult {
    assertMessage(message);
    if (this.#seen.has(message.id)) return { status: "duplicate", applied: [], gap: null };
    if (message.seq <= this.#highWater) {
      this.#remember(message.id);
      return { status: "late", applied: [], gap: null };
    }
    if (message.seq > this.#highWater + 1) {
      const existing = this.#buffered.get(message.seq);
      if (existing) {
        // 同一个序列号被两条不同 id 占用，按重复处理并保留先到者。
        return { status: "duplicate", applied: [], gap: this.#gapBefore(message.seq) };
      }
      this.#buffered.set(message.seq, message);
      this.#remember(message.id);
      return { status: "buffered", applied: [], gap: this.#gapBefore(message.seq) };
    }
    this.#remember(message.id);
    const applied: SequencedMessage[] = [message];
    this.#highWater = message.seq;
    this.#appliedOrder.push(message.seq);
    while (this.#buffered.has(this.#highWater + 1)) {
      const next = this.#buffered.get(this.#highWater + 1) as SequencedMessage;
      this.#buffered.delete(next.seq);
      applied.push(next);
      this.#highWater = next.seq;
      this.#appliedOrder.push(next.seq);
    }
    this.#trimHistory();
    return { status: "applied", applied, gap: null };
  }

  // 只报告已缓存消息之间能确认的空洞；没有上界时无法推断尾部缺失。
  gaps(): SeqRange[] {
    return this.#missing(this.#bufferedHigh());
  }

  // 以服务端高水位为上界，生成补拉区间，并按 batchSize 切分。
  planPull(serverHighWater: number, batchSize = 50): SeqRange[] {
    if (!Number.isInteger(serverHighWater) || serverHighWater < 0) {
      throw new Error("serverHighWater 必须是非负整数");
    }
    if (!Number.isInteger(batchSize) || batchSize < 1) throw new Error("batchSize 必须是正整数");
    const upTo = Math.max(serverHighWater, this.#bufferedHigh());
    return this.#missing(upTo).flatMap((range) => splitRange(range, batchSize));
  }

  #bufferedHigh(): number {
    let high = this.#highWater;
    for (const seq of this.#buffered.keys()) {
      if (seq > high) high = seq;
    }
    return high;
  }

  #gapBefore(seq: number): SeqRange | null {
    const ranges = this.#missing(seq - 1);
    return ranges.length > 0 ? ranges[0] : null;
  }

  #missing(upTo: number): SeqRange[] {
    const ranges: SeqRange[] = [];
    let start = -1;
    for (let seq = this.#highWater + 1; seq <= upTo; seq += 1) {
      if (this.#buffered.has(seq)) {
        if (start !== -1) {
          ranges.push({ from: start, to: seq - 1 });
          start = -1;
        }
        continue;
      }
      if (start === -1) start = seq;
    }
    if (start !== -1) ranges.push({ from: start, to: upTo });
    return ranges;
  }

  #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 > this.#historyLimit) this.#appliedOrder.shift();
  }
}

export interface PullSource {
  serverHighWater(): number;
  fetchRange(range: SeqRange): Promise<SequencedMessage[]>;
}

export interface CatchUpReport {
  requests: number;
  received: number;
  applied: number;
  duplicates: number;
  remainingGaps: SeqRange[];
  highWater: number;
  finishedAt: number;
}

export type Clock = () => number;

export class CatchUpSession {
  #tracker: GapTracker;
  #source: PullSource;
  #now: Clock;
  #running = false;

  constructor(tracker: GapTracker, source: PullSource, now: Clock = () => Date.now()) {
    this.#tracker = tracker;
    this.#source = source;
    this.#now = now;
  }

  get running(): boolean {
    return this.#running;
  }

  // 补拉期间实时通道仍可能推送，直接进入同一个 tracker，由 id 去重合并。
  ingestLive(message: SequencedMessage): GapIngestResult {
    return this.#tracker.ingest(message);
  }

  async catchUp(): Promise<CatchUpReport> {
    if (this.#running) throw new Error("补拉已在进行中");
    this.#running = true;
    try {
      const serverHigh = this.#source.serverHighWater();
      const ranges = this.#tracker.planPull(serverHigh);
      let requests = 0;
      let received = 0;
      let applied = 0;
      let duplicates = 0;
      for (const range of ranges) {
        const page = await this.#source.fetchRange(range);
        requests += 1;
        received += page.length;
        for (const message of page) {
          const result = this.#tracker.ingest(message);
          applied += result.applied.length;
          if (result.status === "duplicate") duplicates += 1;
        }
      }
      return {
        requests,
        received,
        applied,
        duplicates,
        remainingGaps: this.#tracker.planPull(Math.max(serverHigh, this.#tracker.highWater)),
        highWater: this.#tracker.highWater,
        finishedAt: this.#now(),
      };
    } finally {
      this.#running = false;
    }
  }
}

function splitRange(range: SeqRange, batchSize: number): SeqRange[] {
  const chunks: SeqRange[] = [];
  for (let from = range.from; from <= range.to; from += batchSize) {
    chunks.push({ from, to: Math.min(range.to, from + batchSize - 1) });
  }
  return chunks;
}

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-concurrency=1 test/reliability/gap.test.ts
✔ 按已收序列识别缺口 (0.996042ms)
✔ 补拉区间按批大小切分 (0.260917ms)
✔ 补拉结果与已缓存的实时消息合并 (0.267209ms)
✔ 补拉期间到达的实时消息不重复应用 (0.186583ms)
✔ 只补最新一条会留下永远补不上的空洞 (0.10275ms)
✔ 已经连续时补拉不发请求 (0.0895ms)
✔ 服务端返回不完整时保留剩余空洞 (0.092709ms)
ℹ tests 7
ℹ suites 0
ℹ pass 7
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 175.77425