《移动游戏产品的架构·网络、资源与热更新》4.2 操作队列的幂等与去重

// file: src/net/offline-queue.ts
// tier: L1
// verify: node --test test/net/offline-queue.test.ts

export interface QueuedOperation {
  readonly opId: string;
  readonly seq: number;
  readonly kind: string;
  readonly payload: string;
  readonly createdAt: number;
  readonly attempts: number;
  readonly nextRetryAt: number;
}

export interface QueueState {
  readonly items: readonly QueuedOperation[];
  readonly acked: ReadonlyMap<string, string>;
  readonly nextSeq: number;
}

export interface EnqueueInput {
  readonly opId: string;
  readonly kind: string;
  readonly payload: string;
  readonly createdAt: number;
}

export type EnqueueCode = "appended" | "duplicate-pending" | "duplicate-acked" | "payload-mismatch";

export interface EnqueueResult {
  readonly state: QueueState;
  readonly code: EnqueueCode;
}

export interface RetryPolicy {
  readonly baseDelayMs: number;
  readonly maxDelayMs: number;
  readonly maxAttempts: number;
}

const DEFAULT_RETRY_POLICY: RetryPolicy = {
  baseDelayMs: 1_000,
  maxDelayMs: 60_000,
  maxAttempts: 5,
};

function signature(kind: string, payload: string): string {
  return `${kind}\u0000${payload}`;
}

export function createQueue(): QueueState {
  return { items: [], acked: new Map(), nextSeq: 1 };
}

export function enqueue(state: QueueState, input: EnqueueInput): EnqueueResult {
  const pending = state.items.find((item) => item.opId === input.opId);
  if (pending !== undefined) {
    const same = pending.kind === input.kind && pending.payload === input.payload;
    return { state, code: same ? "duplicate-pending" : "payload-mismatch" };
  }

  const ackedSignature = state.acked.get(input.opId);
  if (ackedSignature !== undefined) {
    const same = ackedSignature === signature(input.kind, input.payload);
    return { state, code: same ? "duplicate-acked" : "payload-mismatch" };
  }

  const item: QueuedOperation = {
    opId: input.opId,
    seq: state.nextSeq,
    kind: input.kind,
    payload: input.payload,
    createdAt: input.createdAt,
    attempts: 0,
    nextRetryAt: input.createdAt,
  };
  return {
    state: { items: [...state.items, item], acked: state.acked, nextSeq: state.nextSeq + 1 },
    code: "appended",
  };
}

export interface BatchPlan {
  readonly batch: readonly QueuedOperation[];
  readonly deferred: number;
}

export function nextBatch(state: QueueState, now: number, limit: number): BatchPlan {
  if (!Number.isInteger(limit) || limit <= 0) {
    throw new RangeError(`batch limit must be a positive integer: ${limit}`);
  }
  const ordered = [...state.items].sort((a, b) => a.seq - b.seq);
  const batch = ordered.filter((item) => item.nextRetryAt <= now).slice(0, limit);
  return { batch, deferred: ordered.length - batch.length };
}

export interface AckResult {
  readonly state: QueueState;
  readonly acked: readonly string[];
  readonly alreadyAcked: readonly string[];
  readonly unknown: readonly string[];
}

export function acknowledge(state: QueueState, opIds: readonly string[]): AckResult {
  const pendingIds = new Set(opIds);
  const ackedNow: string[] = [];
  const remaining: QueuedOperation[] = [];
  const ackedMap = new Map(state.acked);

  for (const item of state.items) {
    if (pendingIds.has(item.opId)) {
      ackedMap.set(item.opId, signature(item.kind, item.payload));
      ackedNow.push(item.opId);
      pendingIds.delete(item.opId);
    } else {
      remaining.push(item);
    }
  }

  const alreadyAcked: string[] = [];
  const unknown: string[] = [];
  for (const opId of pendingIds) {
    if (ackedMap.has(opId)) {
      alreadyAcked.push(opId);
    } else {
      unknown.push(opId);
    }
  }

  return {
    state: { items: remaining, acked: ackedMap, nextSeq: state.nextSeq },
    acked: ackedNow,
    alreadyAcked,
    unknown,
  };
}

export type FailureCode = "retry-scheduled" | "dropped" | "unknown";

export interface FailureResult {
  readonly state: QueueState;
  readonly code: FailureCode;
  readonly retryAt?: number;
}

export function markFailed(
  state: QueueState,
  opId: string,
  now: number,
  policy: RetryPolicy = DEFAULT_RETRY_POLICY,
): FailureResult {
  const target = state.items.find((item) => item.opId === opId);
  if (target === undefined) {
    return { state, code: "unknown" };
  }

  const attempts = target.attempts + 1;
  if (attempts >= policy.maxAttempts) {
    return {
      state: { items: state.items.filter((item) => item.opId !== opId), acked: state.acked, nextSeq: state.nextSeq },
      code: "dropped",
    };
  }

  const delay = Math.min(policy.baseDelayMs * 2 ** target.attempts, policy.maxDelayMs);
  const retryAt = now + delay;
  const items = state.items.map((item) =>
    item.opId === opId ? { ...item, attempts, nextRetryAt: retryAt } : item,
  );
  return {
    state: { items, acked: state.acked, nextSeq: state.nextSeq },
    code: "retry-scheduled",
    retryAt,
  };
}

export function describeQueue(state: QueueState, now: number): string {
  const ready = state.items.filter((item) => item.nextRetryAt <= now).length;
  const waiting = state.items.length - ready;
  const next = state.items.length === 0 ? "无" : String(state.items[0].seq);
  return `队列 ${state.items.length} 条（就绪 ${ready}，等待重试 ${waiting}），已确认 ${state.acked.size} 条，下一条 seq=${next}`;
}