《互联网产品的客户端架构·内容、社交与实时互动》13.2 故障注入：断网、慢网与断线重连

// file: src/summary/fault-injection.ts
// tier: L2
// verify: node --test --test-concurrency=1 test/summary/fault-injection.test.ts
import { CatchUpSession, GapTracker } from "../reliability/gap.ts";
import type {
  CatchUpReport,
  GapIngestResult,
  PullSource,
  SequencedMessage,
} from "../reliability/gap.ts";

export type NetworkMode = "online" | "slow" | "offline";

export class OfflineError extends Error {
  constructor() {
    super("网络离线");
    this.name = "OfflineError";
  }
}

// 时间只以计数器形式存在，advance 负责推进；不依赖真实 setTimeout。
export class ControlledNetwork {
  #mode: NetworkMode;
  #latency: number;
  #reorder: boolean;
  #now = 0;
  #order = 0;
  #pending: Array<{
    due: number;
    order: number;
    label: string;
    value: unknown;
    resolve: (value: unknown) => void;
  }> = [];
  #delivered: string[] = [];

  constructor(options: { mode?: NetworkMode; latencyMs?: number; reorder?: boolean } = {}) {
    this.#mode = options.mode ?? "online";
    this.#latency = options.latencyMs ?? 0;
    this.#reorder = options.reorder ?? false;
    assertLatency(this.#latency);
  }

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

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

  setMode(mode: NetworkMode): void {
    this.#mode = mode;
  }

  deliveryOrder(): string[] {
    return [...this.#delivered];
  }

  request<T>(value: T, label = ""): Promise<T> {
    if (this.#mode === "offline") return Promise.reject(new OfflineError());
    if (this.#mode === "online" || this.#latency === 0) {
      this.#delivered.push(label);
      return Promise.resolve(value);
    }
    this.#order += 1;
    return new Promise<T>((resolve) => {
      this.#pending.push({
        due: this.#now + this.#latency,
        order: this.#order,
        label,
        value,
        resolve: resolve as (value: unknown) => void,
      });
    });
  }

  // 一次推进最多交付一批；需要连续交付时由调用方继续 advance。
  async advance(ms: number): Promise<void> {
    if (!Number.isInteger(ms) || ms < 0) throw new RangeError("advance 必须是非负整数");
    this.#now += ms;
    const due = this.#pending.filter((delivery) => delivery.due <= this.#now);
    this.#pending = this.#pending.filter((delivery) => delivery.due > this.#now);
    due.sort((a, b) => (this.#reorder ? b.order - a.order : a.order - b.order));
    for (const delivery of due) {
      this.#delivered.push(delivery.label);
      delivery.resolve(delivery.value);
    }
    await Promise.resolve();
    await Promise.resolve();
  }
}

export interface Mutation {
  key: string;
  kind: "like" | "comment";
  itemId: string;
  active?: boolean;
  text?: string;
}

export interface MutationState {
  likeCount: number;
  commentCount: number;
}

export interface FlushReport {
  sent: number;
  failed: number;
  deduped: number;
  pending: number;
}

// 服务端按幂等键去重；重复投递只返回当前状态，不再改计数。
export class MutationLog {
  #applied = new Set<string>();
  #liked = new Set<string>();
  #likes = new Map<string, number>();
  #comments = new Map<string, number>();
  #texts: string[] = [];

  apply(mutation: Mutation): MutationState & { applied: boolean } {
    assertMutation(mutation);
    const state = this.state(mutation.itemId);
    if (this.#applied.has(mutation.key)) return { applied: false, ...state };
    this.#applied.add(mutation.key);
    if (mutation.kind === "like") {
      const next = mutation.active ?? false;
      if (this.#liked.has(mutation.itemId) !== next) {
        this.#likes.set(mutation.itemId, Math.max(0, state.likeCount + (next ? 1 : -1)));
        if (next) this.#liked.add(mutation.itemId);
        else this.#liked.delete(mutation.itemId);
      }
    } else {
      this.#comments.set(mutation.itemId, state.commentCount + 1);
      this.#texts.push(mutation.text ?? "");
    }
    return { applied: true, ...this.state(mutation.itemId) };
  }

  state(itemId: string): MutationState {
    return {
      likeCount: this.#likes.get(itemId) ?? 0,
      commentCount: this.#comments.get(itemId) ?? 0,
    };
  }

  get appliedCount(): number {
    return this.#applied.size;
  }

  get texts(): string[] {
    return [...this.#texts];
  }
}

// 离线时写入留在队列；恢复后整批重放，服务端用幂等键兜底。
export class Outbox {
  #queue: Mutation[] = [];

  enqueue(mutation: Mutation): boolean {
    assertMutation(mutation);
    if (this.#queue.some((entry) => entry.key === mutation.key)) return false;
    this.#queue.push({ ...mutation });
    return true;
  }

  get pending(): number {
    return this.#queue.length;
  }

  async flush(network: ControlledNetwork, log: MutationLog): Promise<FlushReport> {
    const batch = [...this.#queue];
    if (batch.length === 0) return { sent: 0, failed: 0, deduped: 0, pending: 0 };
    const settled = await Promise.allSettled(
      batch.map((mutation) => network.request(mutation, mutation.key)),
    );
    let sent = 0;
    let failed = 0;
    let deduped = 0;
    const remaining: Mutation[] = [];
    settled.forEach((result, index) => {
      const mutation = batch[index];
      if (result.status === "rejected") {
        failed += 1;
        remaining.push(mutation);
        return;
      }
      sent += 1;
      if (!log.apply(result.value).applied) deduped += 1;
    });
    this.#queue = remaining;
    return { sent, failed, deduped, pending: remaining.length };
  }
}

// 包装 7.3 的 GapTracker 与 CatchUpSession：实时消息和补拉响应进入同一个 tracker。
export class RealtimeSync {
  #tracker: GapTracker;
  #running = false;

  constructor(startSeq = 0) {
    this.#tracker = new GapTracker({ startSeq });
  }

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

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

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

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

  receive(message: SequencedMessage): GapIngestResult {
    return this.#tracker.ingest(message);
  }

  async catchUp(source: PullSource): Promise<CatchUpReport> {
    if (this.#running) throw new Error("补拉已在进行中");
    this.#running = true;
    try {
      return await new CatchUpSession(this.#tracker, source, () => 0).catchUp();
    } finally {
      this.#running = false;
    }
  }
}

function assertLatency(ms: number): void {
  if (!Number.isInteger(ms) || ms < 0) throw new RangeError("latencyMs 必须是非负整数");
}

function assertMutation(mutation: Mutation): void {
  if (typeof mutation.key !== "string" || mutation.key.length === 0) {
    throw new Error("幂等键不能为空");
  }
  if (mutation.kind !== "like" && mutation.kind !== "comment") {
    throw new Error(`未知写入类型: ${mutation.kind}`);
  }
  if (typeof mutation.itemId !== "string" || mutation.itemId.length === 0) {
    throw new Error("itemId 不能为空");
  }
  if (mutation.kind === "like" && typeof mutation.active !== "boolean") {
    throw new Error("点赞必须给出目标状态");
  }
  if (mutation.kind === "comment" && (mutation.text ?? "").trim() === "") {
    throw new Error("评论内容不能为空");
  }
}

$ node --test --test-concurrency=1 test/summary/fault-injection.test.ts
✔ 离线时写入只进队列，不产生服务端副作用 (1.067ms)
✔ 同一幂等键重复入队只保留一条，重放不重复计数 (0.286334ms)
✔ 慢网下请求在假时钟推进前不落地 (0.228834ms)
✔ 乱序交付不改变幂等写入的最终状态 (0.138958ms)
✔ 断线重连后按高水位补拉缺失区间，并与实时消息去重 (0.470459ms)
✔ 补拉只返回一部分时保留剩余空洞，下一次继续补齐 (0.152ms)
✔ 补拉期间离线会失败且水位不前进，恢复后可以重试 (0.336042ms)
✔ 乱序实时消息先缓冲，缺口补齐后连续应用 (0.123584ms)
ℹ tests 8
ℹ suites 0
ℹ pass 8
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 198.857667