《移动游戏产品的架构·网络、资源与热更新》3.3 重连后的状态校准

// file: src/net/resync.ts
// tier: L1
// verify: node --test test/net/resync.test.ts
/**
 * 重连后的状态校准模型（mga-net-lab）。
 * 增量窗口与提交结构是实验设定，不是案例事实。
 */

export const DEFAULT_DELTA_WINDOW = 128;

export type OpKind = "set" | "add";

export interface ServerOp {
  readonly seq: number;
  readonly key: string;
  readonly op: OpKind;
  readonly value: number;
}

export interface ClientState {
  readonly epoch: string;
  readonly seq: number;
  readonly values: Readonly<Record<string, number>>;
}

export interface PendingSubmit {
  readonly submitId: string;
  readonly roundId: number;
  readonly score: number;
}

export interface ResyncRequest {
  readonly clientEpoch: string;
  readonly serverEpoch: string;
  readonly lastAckedSeq: number;
  readonly serverSeq: number;
  readonly deltaWindow?: number;
}

export type ResyncKind = "noop" | "delta" | "snapshot" | "reset";

export interface ResyncPlan {
  readonly kind: ResyncKind;
  readonly fromSeq: number;
  readonly toSeq: number;
  readonly missing: number;
  readonly reason: string;
}

export interface ServerResyncResponse {
  readonly epoch: string;
  readonly seq: number;
  readonly ops: readonly ServerOp[];
  readonly snapshot?: Readonly<Record<string, number>>;
  readonly ackedSubmitIds: readonly string[];
}

export interface ReconcileResult {
  readonly plan: ResyncPlan;
  readonly state: ClientState;
  readonly replay: readonly PendingSubmit[];
  readonly dropped: readonly PendingSubmit[];
  readonly applied: number;
  readonly skipped: number;
}

function assertSeq(value: number, label: string): void {
  if (!Number.isInteger(value) || value < 0) {
    throw new RangeError(`${label} 必须是非负整数，收到 ${value}`);
  }
}

export function planResync(request: ResyncRequest): ResyncPlan {
  assertSeq(request.lastAckedSeq, "lastAckedSeq");
  assertSeq(request.serverSeq, "serverSeq");
  const window = request.deltaWindow ?? DEFAULT_DELTA_WINDOW;
  assertSeq(window, "deltaWindow");
  const { lastAckedSeq: from, serverSeq: to } = request;

  if (request.clientEpoch !== request.serverEpoch) {
    return {
      kind: "snapshot",
      fromSeq: from,
      toSeq: to,
      missing: Math.max(0, to - from),
      reason: "会话纪元不一致，本地增量不可复用",
    };
  }
  if (from > to) {
    return {
      kind: "reset",
      fromSeq: from,
      toSeq: to,
      missing: 0,
      reason: "客户端序号超前，服务端可能已回滚",
    };
  }
  if (from === to) {
    return {
      kind: "noop",
      fromSeq: from,
      toSeq: to,
      missing: 0,
      reason: "序号一致，无需校准",
    };
  }
  const missing = to - from;
  if (missing <= window) {
    return {
      kind: "delta",
      fromSeq: from,
      toSeq: to,
      missing,
      reason: `缺失 ${missing} 条，在增量窗口 ${window} 内`,
    };
  }
  return {
    kind: "snapshot",
    fromSeq: from,
    toSeq: to,
    missing,
    reason: `缺失 ${missing} 条，超过增量窗口 ${window}`,
  };
}

export function applyDelta(
  state: ClientState,
  ops: readonly ServerOp[],
  toSeq: number,
): { state: ClientState; applied: number; skipped: number } {
  assertSeq(toSeq, "toSeq");
  const values: Record<string, number> = { ...state.values };
  let cursor = state.seq;
  let applied = 0;
  let skipped = 0;
  for (const op of ops) {
    if (op.seq <= cursor) {
      skipped += 1;
      continue;
    }
    if (op.seq !== cursor + 1) {
      throw new Error(`增量不连续：期望 ${cursor + 1}，收到 ${op.seq}`);
    }
    if (op.op === "set") {
      values[op.key] = op.value;
    } else {
      values[op.key] = (values[op.key] ?? 0) + op.value;
    }
    cursor = op.seq;
    applied += 1;
  }
  if (cursor !== toSeq) {
    throw new Error(`增量未到达目标序号：游标 ${cursor}，目标 ${toSeq}`);
  }
  return { state: { ...state, seq: cursor, values }, applied, skipped };
}

export function classifyPending(
  pending: readonly PendingSubmit[],
  ackedSubmitIds: readonly string[],
): { replay: readonly PendingSubmit[]; dropped: readonly PendingSubmit[] } {
  const acked = new Set(ackedSubmitIds);
  const replay: PendingSubmit[] = [];
  const dropped: PendingSubmit[] = [];
  for (const submit of pending) {
    if (acked.has(submit.submitId)) {
      dropped.push(submit);
    } else {
      replay.push(submit);
    }
  }
  return { replay, dropped };
}

export function reconcile(
  local: ClientState,
  pending: readonly PendingSubmit[],
  response: ServerResyncResponse,
  deltaWindow = DEFAULT_DELTA_WINDOW,
): ReconcileResult {
  const plan = planResync({
    clientEpoch: local.epoch,
    serverEpoch: response.epoch,
    lastAckedSeq: local.seq,
    serverSeq: response.seq,
    deltaWindow,
  });

  let state: ClientState;
  let applied = 0;
  let skipped = 0;

  if (plan.kind === "noop") {
    state = local;
  } else if (plan.kind === "delta") {
    const result = applyDelta(local, response.ops, response.seq);
    state = result.state;
    applied = result.applied;
    skipped = result.skipped;
  } else if (plan.kind === "snapshot") {
    if (response.snapshot === undefined) {
      throw new Error("快照校准缺少 snapshot 数据");
    }
    state = { epoch: response.epoch, seq: response.seq, values: { ...response.snapshot } };
  } else {
    state = {
      epoch: response.epoch,
      seq: response.seq,
      values: response.snapshot === undefined ? {} : { ...response.snapshot },
    };
  }

  const classified = classifyPending(pending, response.ackedSubmitIds);
  return {
    plan,
    state,
    replay: classified.replay,
    dropped: classified.dropped,
    applied,
    skipped,
  };
}

export function formatResync(result: ReconcileResult): string {
  const labels: Record<ResyncKind, string> = {
    noop: "无需校准",
    delta: "增量校准",
    snapshot: "快照替换",
    reset: "重置校准",
  };
  return `${labels[result.plan.kind]}：seq ${result.plan.fromSeq} → ${result.plan.toSeq}，缺失 ${result.plan.missing} 条，应用 ${result.applied} 条，跳过 ${result.skipped} 条，重放 ${result.replay.length} 条提交，丢弃 ${result.dropped.length} 条重复提交`;
}

增量校准：seq 10 → 22，缺失 12 条，应用 12 条，跳过 0 条，重放 1 条提交，丢弃 1 条重复提交