《互联网产品的 AI 架构·推理性能与成本工程》7.2 分片、重试与幂等

剩余预算 = min(绝对上限, floor(首次尝试数 × 比例上限)) - 已用重试

// file: src/perf/batch-shard.ts
// tier: L1
// verify: node --test test/perf/batch.test.ts

import { AiError } from "../core/errors.ts";
import type { Clock } from "../core/types.ts";

export const VIRTUAL_BUCKETS = 256;

function fail(message: string): never {
  throw new AiError("invalid-request", `batch-shard: ${message}`);
}

function fnv1a(text: string): number {
  let hash = 0x811c9dc5;
  for (let index = 0; index < text.length; index += 1) {
    hash ^= text.charCodeAt(index);
    hash = Math.imul(hash, 0x01000193) >>> 0;
  }
  return hash;
}

/**
 * 条目先落到固定的虚拟桶，桶再分配给 worker。
 * 这样条目到桶的映射不随并发数变化，扩缩容只移动桶。
 */
export function bucketOf(key: string, buckets = VIRTUAL_BUCKETS): number {
  if (key.trim() === "") fail("key is required");
  if (!Number.isInteger(buckets) || buckets <= 0) fail("buckets must be a positive integer");
  return fnv1a(key) % buckets;
}

export interface ShardPlan {
  workerCount: number;
  buckets: number;
  /** 下标是桶号，值是 worker 号。 */
  workerOfBucket: number[];
}

export function planShards(workerCount: number, buckets = VIRTUAL_BUCKETS): ShardPlan {
  if (!Number.isInteger(workerCount) || workerCount <= 0) fail("workerCount must be a positive integer");
  if (!Number.isInteger(buckets) || buckets <= 0) fail("buckets must be a positive integer");
  const workerOfBucket: number[] = [];
  for (let bucket = 0; bucket < buckets; bucket += 1) workerOfBucket.push(bucket % workerCount);
  return { workerCount, buckets, workerOfBucket };
}

export function shardOf(key: string, plan: ShardPlan): number {
  return plan.workerOfBucket[bucketOf(key, plan.buckets)];
}

export type IdempotencyState = "fresh" | "in-flight" | "done";

export interface IdempotencyLookup {
  state: IdempotencyState;
  result?: string;
}

interface Entry {
  status: "in-flight" | "done";
  expiresAtMs: number;
  result?: string;
}

/**
 * 幂等键窗口必须覆盖“作业最长重跑时间 + 重试尾巴 + 时钟偏差”。
 * 窗口短于重跑间隔时，第二次运行会把同一条目重新计费。
 */
export class IdempotencyStore {
  private readonly ttlMs: number;
  private readonly clock: Clock;
  private readonly entries = new Map<string, Entry>();

  constructor(options: { ttlMs: number; clock: Clock }) {
    if (!Number.isInteger(options.ttlMs) || options.ttlMs <= 0) fail("ttlMs must be a positive integer");
    this.ttlMs = options.ttlMs;
    this.clock = options.clock;
  }

  private prune(nowMs: number): void {
    for (const [key, entry] of this.entries) {
      if (entry.expiresAtMs <= nowMs) this.entries.delete(key);
    }
  }

  /** 只读查询，不占用键；用于断点续跑前统计还差多少。 */
  peek(key: string): IdempotencyLookup {
    this.prune(this.clock.now());
    const entry = this.entries.get(key);
    if (entry === undefined) return { state: "fresh" };
    return entry.status === "done" ? { state: "done", result: entry.result } : { state: "in-flight" };
  }

  /** 返回 fresh 才能调用模型；in-flight 表示另一个 worker 正在处理。 */
  begin(key: string): IdempotencyLookup {
    const lookup = this.peek(key);
    if (lookup.state === "fresh") {
      this.entries.set(key, { status: "in-flight", expiresAtMs: this.clock.now() + this.ttlMs });
    }
    return lookup;
  }

  complete(key: string, result: string): void {
    if (key.trim() === "") fail("key is required");
    this.entries.set(key, { status: "done", result, expiresAtMs: this.clock.now() + this.ttlMs });
  }

  size(): number {
    this.prune(this.clock.now());
    return this.entries.size;
  }
}

/** 窗口下限 = 最长作业时长 + 重试尾巴 + 时钟偏差。 */
export function requiredWindowMs(jobDurationMs: number, retryTailMs: number, clockSkewMs: number): number {
  for (const [value, field] of [
    [jobDurationMs, "jobDurationMs"],
    [retryTailMs, "retryTailMs"],
    [clockSkewMs, "clockSkewMs"],
  ] as const) {
    if (!Number.isInteger(value) || value < 0) fail(`${field} must be a non-negative integer`);
  }
  return jobDurationMs + retryTailMs + clockSkewMs;
}

export interface RetryPolicy {
  /** 整个作业允许的重试绝对上限。 */
  maxRetries: number;
  /** 重试数不得超过首次尝试数的比例，防止失败放大。 */
  maxRetryRatio: number;
}

export interface RetrySnapshot {
  firstAttempts: number;
  retries: number;
  remaining: number;
}

/**
 * 重试预算是作业级的共享池，不是每个条目各给三次。
 * 否则上游整体故障时，每个条目都重试会形成放大风暴。
 */
export class RetryLedger {
  private readonly policy: RetryPolicy;
  private firstAttempts = 0;
  private retries = 0;

  constructor(policy: RetryPolicy) {
    if (!Number.isInteger(policy.maxRetries) || policy.maxRetries < 0) fail("maxRetries must be a non-negative integer");
    if (!Number.isFinite(policy.maxRetryRatio) || policy.maxRetryRatio < 0 || policy.maxRetryRatio > 1) {
      fail("maxRetryRatio must be in [0, 1]");
    }
    this.policy = { ...policy };
  }

  recordFirstAttempt(count = 1): void {
    if (!Number.isInteger(count) || count < 0) fail("count must be a non-negative integer");
    this.firstAttempts += count;
  }

  canRetry(): boolean {
    const ratioBudget = Math.floor(this.firstAttempts * this.policy.maxRetryRatio);
    return this.retries < this.policy.maxRetries && this.retries < ratioBudget;
  }

  recordRetry(): void {
    if (!this.canRetry()) fail("retry budget exhausted");
    this.retries += 1;
  }

  snapshot(): RetrySnapshot {
    const ratioBudget = Math.floor(this.firstAttempts * this.policy.maxRetryRatio);
    return {
      firstAttempts: this.firstAttempts,
      retries: this.retries,
      remaining: Math.max(0, Math.min(this.policy.maxRetries, ratioBudget) - this.retries),
    };
  }
}

export type ItemOutcome = "ok" | "retryable" | "permanent" | "quarantined";

export interface ItemRun {
  key: string;
  outcome: ItemOutcome;
  attempts: number;
  costMicros: number;
}

export interface RunSummary {
  total: number;
  ok: number;
  retryable: number;
  permanent: number;
  quarantined: number;
  failureRate: number;
  costMicros: number;
}

export function summarizeRun(runs: readonly ItemRun[]): RunSummary {
  const summary: RunSummary = {
    total: runs.length, ok: 0, retryable: 0, permanent: 0, quarantined: 0, failureRate: 0, costMicros: 0,
  };
  for (const run of runs) {
    if (!Number.isInteger(run.attempts) || run.attempts < 0) fail("attempts must be a non-negative integer");
    if (run.costMicros < 0) fail("costMicros must be non-negative");
    summary[run.outcome] += 1;
    summary.costMicros += run.costMicros;
  }
  summary.failureRate = summary.total === 0 ? 0 : (summary.total - summary.ok) / summary.total;
  return summary;
}

export interface CommitPolicy {
  maxFailureRate: number;
  allowPartial: boolean;
}

export interface CommitDecision {
  mode: "all" | "partial" | "hold";
  reason: string;
}

/** 部分失败不是异常：要么发布成功子集并隔离失败条目，要么整批扣住。 */
export function decideCommit(summary: RunSummary, policy: CommitPolicy): CommitDecision {
  if (!Number.isFinite(policy.maxFailureRate) || policy.maxFailureRate < 0 || policy.maxFailureRate > 1) {
    fail("maxFailureRate must be in [0, 1]");
  }
  if (summary.total === 0) return { mode: "hold", reason: "空作业没有可提交内容" };
  if (summary.failureRate === 0) return { mode: "all", reason: "全部条目成功" };
  if (policy.allowPartial && summary.failureRate <= policy.maxFailureRate) {
    return { mode: "partial", reason: `失败率 ${(summary.failureRate * 100).toFixed(2)}% 在阈值内` };
  }
  return { mode: "hold", reason: `失败率 ${(summary.failureRate * 100).toFixed(2)}% 超过阈值或不允许部分发布` };
}

export interface ResumePlan<T extends { key: string }> {
  todo: T[];
  reused: number;
  inFlight: number;
}

/** 断点续跑：已完成的条目直接复用结果，不再调用模型。 */
export function planResume<T extends { key: string }>(
  items: readonly T[],
  store: IdempotencyStore,
): ResumePlan<T> {
  const plan: ResumePlan<T> = { todo: [], reused: 0, inFlight: 0 };
  for (const item of items) {
    const lookup = store.peek(item.key);
    if (lookup.state === "done") plan.reused += 1;
    else if (lookup.state === "in-flight") plan.inFlight += 1;
    else plan.todo.push(item);
  }
  return plan;
}

node --test --test-name-pattern="按键分片|重试预算耗尽|断点续跑|幂等窗口短于" test/perf/batch.test.ts

ℹ tests 4
ℹ suites 0
ℹ pass 4
ℹ fail 0
ℹ duration_ms 586.0255