《互联网产品的 AI 架构·推理性能与成本工程》12.1 供应商限流与分流

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

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

/** 一个供应商在当前窗口内的硬容量，rpm/tpm 都是每分钟口径。 */
export interface ProviderCapacity {
  providerId: string; rpm: number; tpm: number; maxInflight: number;
  /** 健康时的分流权重；权重不是配额，只是概率。 */
  weight: number; healthy: boolean;
}

/** 单请求上限与全局重试预算必须同时存在。 */
export interface RetryPolicy {
  maxAttempts: number; maxRetryRatio: number; crossProvider: boolean;
  baseCooldownMs: number; maxCooldownMs: number;
}

export const DEFAULT_RETRY: RetryPolicy = {
  maxAttempts: 3, maxRetryRatio: 0.1, crossProvider: true,
  baseCooldownMs: 1_000, maxCooldownMs: 60_000,
};

export interface FleetUsage { requests: number; tokens: number; inflight: number; cooldownUntilMs: number; }
export interface Fleet { capacities: Map<string, ProviderCapacity>; usage: Map<string, FleetUsage>; }
export type AdmissionReason = "cooldown" | "rpm" | "tpm" | "inflight" | "unhealthy";
export type Admission = { ok: true } | { ok: false; reason: AdmissionReason; retryAfterMs: number };

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

function integer(value: number, field: string, min = 0): number {
  if (!Number.isInteger(value) || value < min) fail(`${field} must be an integer >= ${min}`);
  return value;
}

function finite(value: number, field: string, min = 0): number {
  if (!Number.isFinite(value) || value < min) fail(`${field} must be a finite number >= ${min}`);
  return value;
}

export function createFleet(providers: readonly ProviderCapacity[]): Fleet {
  if (providers.length === 0) fail("providers must not be empty");
  const capacities = new Map<string, ProviderCapacity>();
  const usage = new Map<string, FleetUsage>();
  for (const provider of providers) {
    if (provider.providerId.trim() === "") fail("providerId must not be empty");
    if (capacities.has(provider.providerId)) fail(`duplicate provider ${provider.providerId}`);
    integer(provider.rpm, `${provider.providerId}.rpm`, 1);
    integer(provider.tpm, `${provider.providerId}.tpm`, 1);
    integer(provider.maxInflight, `${provider.providerId}.maxInflight`, 1);
    finite(provider.weight, `${provider.providerId}.weight`);
    if (provider.weight <= 0) fail(`${provider.providerId}.weight must be > 0`);
    capacities.set(provider.providerId, { ...provider });
    usage.set(provider.providerId, { requests: 0, tokens: 0, inflight: 0, cooldownUntilMs: 0 });
  }
  return { capacities, usage };
}

function usageOf(fleet: Fleet, providerId: string): FleetUsage {
  const usage = fleet.usage.get(providerId);
  if (!usage) fail(`unknown provider ${providerId}`);
  return usage;
}

/** 检查不修改状态；调用方可以先判断，再决定是否消耗一次重试预算。 */
export function checkAdmission(fleet: Fleet, providerId: string, tokens: number, nowMs: number): Admission {
  const capacity = fleet.capacities.get(providerId);
  const usage = usageOf(fleet, providerId);
  if (!capacity.healthy) return { ok: false, reason: "unhealthy", retryAfterMs: 0 };
  if (nowMs < usage.cooldownUntilMs) {
    return { ok: false, reason: "cooldown", retryAfterMs: usage.cooldownUntilMs - nowMs };
  }
  if (usage.inflight >= capacity.maxInflight) return { ok: false, reason: "inflight", retryAfterMs: 0 };
  if (usage.requests + 1 > capacity.rpm) return { ok: false, reason: "rpm", retryAfterMs: 60_000 };
  if (usage.tokens + tokens > capacity.tpm) return { ok: false, reason: "tpm", retryAfterMs: 60_000 };
  return { ok: true };
}

export function commitAdmission(fleet: Fleet, providerId: string, tokens: number): void {
  const usage = usageOf(fleet, providerId);
  usage.requests += 1; usage.tokens += tokens; usage.inflight += 1;
}

export function release(fleet: Fleet, providerId: string): void {
  const usage = usageOf(fleet, providerId);
  if (usage.inflight > 0) usage.inflight -= 1;
}

/** 429 之后让供应商休息；Retry-After 优先，缺失时用基准值，二者都被上限截断。 */
export function markRateLimited(
  fleet: Fleet, providerId: string, nowMs: number,
  retryAfterMs: number | undefined, policy: RetryPolicy = DEFAULT_RETRY,
): number {
  const delay = Math.max(0, Math.min(retryAfterMs ?? policy.baseCooldownMs, policy.maxCooldownMs));
  const usage = usageOf(fleet, providerId);
  usage.cooldownUntilMs = Math.max(usage.cooldownUntilMs, nowMs + delay);
  return delay;
}

/** 解析 Retry-After：秒数与 HTTP 日期都接受，非法值回退到基准冷却。 */
export function parseRetryAfter(value: string | null, nowMs: number, policy: RetryPolicy = DEFAULT_RETRY): number {
  if (value === null || value.trim() === "") return policy.baseCooldownMs;
  const seconds = Number(value.trim());
  if (Number.isFinite(seconds) && seconds >= 0) return Math.min(seconds * 1000, policy.maxCooldownMs);
  const at = Date.parse(value);
  if (Number.isFinite(at)) return Math.max(0, Math.min(at - nowMs, policy.maxCooldownMs));
  return policy.baseCooldownMs;
}

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;
}

/** 确定性哈希分流：同一个 requestId 重放多少次，首选顺序都不变。 */
export function orderProviders(fleet: Fleet, requestId: string): string[] {
  return [...fleet.capacities.values()]
    .filter((provider) => provider.healthy)
    .map((provider) => ({
      providerId: provider.providerId,
      score: (fnv1a(`${requestId}:${provider.providerId}`) % 10_000) / provider.weight,
    }))
    .sort((left, right) => left.score - right.score || left.providerId.localeCompare(right.providerId))
    .map((item) => item.providerId);
}

export interface RetryLedger { originalRequests: number; retryLimit: number; retryUsed: number; }

/** 全局重试预算按原始请求数计算，而不是按失败数计算。 */
export function createRetryLedger(originalRequests: number, policy: RetryPolicy = DEFAULT_RETRY): RetryLedger {
  integer(originalRequests, "originalRequests");
  finite(policy.maxRetryRatio, "maxRetryRatio");
  if (policy.maxRetryRatio < 0 || policy.maxRetryRatio > 1) fail("maxRetryRatio must be in [0, 1]");
  integer(policy.maxAttempts, "maxAttempts", 1);
  return { originalRequests, retryLimit: Math.floor(originalRequests * policy.maxRetryRatio), retryUsed: 0 };
}

export function takeRetry(ledger: RetryLedger): boolean {
  if (ledger.retryUsed >= ledger.retryLimit) return false;
  ledger.retryUsed += 1;
  return true;
}

/** 配额池的剩余量：把每个供应商这一分钟还能接的请求数相加。 */
export function poolHeadroom(fleet: Fleet, nowMs: number): number {
  let total = 0;
  for (const capacity of fleet.capacities.values()) {
    const usage = usageOf(fleet, capacity.providerId);
    if (capacity.healthy && nowMs >= usage.cooldownUntilMs) {
      total += Math.max(0, capacity.rpm - usage.requests);
    }
  }
  return total;
}

export interface StormRequest { requestId: string; tokens: number; }
export type StormReplyKind = "ok" | "rate-limit" | "unavailable";
export interface StormReplyInput { providerId: string; request: StormRequest; index: number; attempt: number; }
export interface StormProviderStats {
  attempts: number; ok: number; rateLimited: number; unavailable: number; admissionSkips: number;
}
export interface StormReport {
  original: number; attempts: number; served: number; shed: number;
  rateLimited: number; unavailable: number; admissionSkips: number;
  retryAttempts: number; retryLimit: number; retryBudgetExhausted: number;
  amplification: number; perProvider: Record<string, StormProviderStats>;
}
export interface StormInput {
  fleet: Fleet; requests: readonly StormRequest[]; policy: RetryPolicy;
  /** 必须是确定性纯函数：同一个输入返回同一个结果。 */
  reply(input: StormReplyInput): StormReplyKind;
  nowMs?: number; cooldownMs?: number;
}

/**
 * 跑一批请求，同时观察单请求上限与全局重试预算。
 * 准入失败不算一次尝试，也不消耗重试预算；只有真正发出去的请求才计数。
 */
export function simulateStorm(input: StormInput): StormReport {
  const nowMs = input.nowMs ?? 0;
  const cooldownMs = input.cooldownMs ?? 0;
  const ledger = createRetryLedger(input.requests.length, input.policy);
  const perProvider: Record<string, StormProviderStats> = {};
  for (const providerId of input.fleet.capacities.keys()) {
    perProvider[providerId] = { attempts: 0, ok: 0, rateLimited: 0, unavailable: 0, admissionSkips: 0 };
  }
  let attempts = 0; let served = 0; let shed = 0; let rateLimited = 0;
  let unavailable = 0; let admissionSkips = 0; let retryBudgetExhausted = 0;

  input.requests.forEach((request, index) => {
    let used = 0; let done = false;
    for (const providerId of orderProviders(input.fleet, request.requestId)) {
      if (used >= input.policy.maxAttempts) break;
      if (used > 0 && !input.policy.crossProvider) break;
      if (!checkAdmission(input.fleet, providerId, request.tokens, nowMs).ok) {
        admissionSkips += 1; perProvider[providerId].admissionSkips += 1; continue;
      }
      if (used > 0 && !takeRetry(ledger)) { retryBudgetExhausted += 1; break; }
      commitAdmission(input.fleet, providerId, request.tokens);
      used += 1; attempts += 1; perProvider[providerId].attempts += 1;
      const reply = input.reply({ providerId, request, index, attempt: used });
      if (reply === "ok") {
        served += 1; perProvider[providerId].ok += 1; release(input.fleet, providerId); done = true; break;
      }
      if (reply === "rate-limit") {
        rateLimited += 1; perProvider[providerId].rateLimited += 1;
        markRateLimited(input.fleet, providerId, nowMs, cooldownMs, input.policy);
      } else {
        unavailable += 1; perProvider[providerId].unavailable += 1;
      }
      release(input.fleet, providerId);
    }
    if (!done) shed += 1;
  });

  return {
    original: input.requests.length, attempts, served, shed, rateLimited, unavailable, admissionSkips,
    retryAttempts: ledger.retryUsed, retryLimit: ledger.retryLimit, retryBudgetExhausted,
    amplification: input.requests.length === 0 ? 0 : attempts / input.requests.length, perProvider,
  };
}