《互联网产品的 AI 架构·推理性能与成本工程》11.3 灰度发布与自动回滚

bucket = floor(hash(rolloutId, tenantId) / 2^32 × 10000)
arm    = bucket < percent × 100 ? treatment : control

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

import { createHash } from "node:crypto";
import { AiError } from "../core/errors.ts";

export interface ArmMetrics {
  requests: number; errors: number; p95LatencyMs: number;
  avgCostMicros: number; qualityPasses: number; qualitySamples: number;
}

export interface RolloutWindow {
  /** 当前灰度比例，1 到 100 的整数。 */
  stagePercent: number; startedAtMs: number;
  control: ArmMetrics; treatment: ArmMetrics;
}

export interface RollbackPolicy {
  /** treatment 错误率允许比 control 高出的绝对值。 */
  maxErrorRateDelta: number; maxErrorRate: number;
  maxLatencyRatio: number; maxCostRatio: number; maxQualityDrop: number;
  minRequestsPerArm: number; minQualitySamples: number;
  /** 软指标连续多少个窗口越界才回滚；硬指标立即回滚。 */
  consecutiveBreaches: number;
}

export const DEFAULT_ROLLBACK: RollbackPolicy = {
  maxErrorRateDelta: 0.005, maxErrorRate: 0.02,
  maxLatencyRatio: 1.15, maxCostRatio: 1.1, maxQualityDrop: 0.02,
  minRequestsPerArm: 1000, minQualitySamples: 200, consecutiveBreaches: 2,
};

export type Arm = "control" | "treatment";

/**
 * 同一个租户在同一个 rollout 内始终落在同一臂：分桶只依赖 rolloutId 与 tenantId，
 * 不依赖当前比例，所以 1% 升到 5% 时已经进入 treatment 的租户不会被换回去。
 */
export function assignArm(tenantId: string, rolloutId: string, percent: number): Arm {
  if (typeof tenantId !== "string" || tenantId.trim() === "") fail("tenantId is required");
  if (typeof rolloutId !== "string" || rolloutId.trim() === "") fail("rolloutId is required");
  if (!Number.isFinite(percent) || percent < 0 || percent > 100) fail("percent must be between 0 and 100");
  if (percent === 0) return "control";
  if (percent === 100) return "treatment";
  const digest = createHash("sha256").update(`${rolloutId}\u0000${tenantId}`, "utf8").digest();
  const bucket = Math.floor((digest.readUInt32BE(0) / 4294967296) * 10000);
  return bucket < Math.round(percent * 100) ? "treatment" : "control";
}

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

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

function validateMetrics(metrics: ArmMetrics, label: string): void {
  integer(metrics.requests, `${label}.requests`, 0);
  integer(metrics.errors, `${label}.errors`, 0);
  if (metrics.errors > metrics.requests) fail(`${label}.errors cannot exceed requests`);
  if (!Number.isFinite(metrics.p95LatencyMs) || metrics.p95LatencyMs < 0) fail(`${label}.p95LatencyMs must be >= 0`);
  integer(metrics.avgCostMicros, `${label}.avgCostMicros`, 0);
  integer(metrics.qualitySamples, `${label}.qualitySamples`, 0);
  integer(metrics.qualityPasses, `${label}.qualityPasses`, 0);
  if (metrics.qualityPasses > metrics.qualitySamples) fail(`${label}.qualityPasses cannot exceed qualitySamples`);
}

function rate(numerator: number, denominator: number): number {
  return denominator === 0 ? 0 : numerator / denominator;
}

function ratio(candidate: number, baseline: number): number {
  if (baseline === 0) return candidate === 0 ? 1 : Number.POSITIVE_INFINITY;
  return candidate / baseline;
}

export type CheckSeverity = "hard" | "soft";

export interface RolloutCheck {
  id: string;
  passed: boolean;
  severity: CheckSeverity;
  detail: string;
}

export interface WindowEvaluation {
  stagePercent: number;
  checks: RolloutCheck[];
  hardBreach: boolean;
  softBreach: boolean;
  sufficient: boolean;
  action: "pass" | "hold" | "breach";
}

/**
 * 单个观测窗口的判定。错误率是硬指标，延迟、成本与质量是软指标：
 * 硬指标立即回滚，软指标要连续多个窗口越界才回滚，避免一次抖动打回全量。
 */
export function evaluateWindow(window: RolloutWindow, policy: RollbackPolicy = DEFAULT_ROLLBACK): WindowEvaluation {
  integer(window.stagePercent, "stagePercent", 1);
  if (window.stagePercent > 100) fail("stagePercent must be <= 100");
  integer(window.startedAtMs, "startedAtMs", 0);
  validateMetrics(window.control, "control");
  validateMetrics(window.treatment, "treatment");

  const controlError = rate(window.control.errors, window.control.requests);
  const treatmentError = rate(window.treatment.errors, window.treatment.requests);
  const controlQuality = rate(window.control.qualityPasses, window.control.qualitySamples);
  const treatmentQuality = rate(window.treatment.qualityPasses, window.treatment.qualitySamples);
  const checks: RolloutCheck[] = [];

  checks.push({ id: "error-absolute", passed: treatmentError <= policy.maxErrorRate, severity: "hard", detail: `treatment error ${(treatmentError * 100).toFixed(2)}%, limit ${(policy.maxErrorRate * 100).toFixed(2)}%` });
  checks.push({ id: "error-delta", passed: treatmentError - controlError <= policy.maxErrorRateDelta, severity: "hard", detail: `delta ${((treatmentError - controlError) * 100).toFixed(2)}pt, limit ${(policy.maxErrorRateDelta * 100).toFixed(2)}pt` });
  checks.push({ id: "latency", passed: ratio(window.treatment.p95LatencyMs, window.control.p95LatencyMs) <= policy.maxLatencyRatio, severity: "soft", detail: `p95 ratio ${ratio(window.treatment.p95LatencyMs, window.control.p95LatencyMs).toFixed(3)} <= ${policy.maxLatencyRatio}` });
  checks.push({ id: "cost", passed: ratio(window.treatment.avgCostMicros, window.control.avgCostMicros) <= policy.maxCostRatio, severity: "soft", detail: `cost ratio ${ratio(window.treatment.avgCostMicros, window.control.avgCostMicros).toFixed(3)} <= ${policy.maxCostRatio}` });

  const qualitySufficient =
    window.control.qualitySamples >= policy.minQualitySamples &&
    window.treatment.qualitySamples >= policy.minQualitySamples;
  if (qualitySufficient) {
    checks.push({ id: "quality", passed: treatmentQuality >= controlQuality - policy.maxQualityDrop, severity: "soft", detail: `quality delta ${((treatmentQuality - controlQuality) * 100).toFixed(2)}pt >= -${(policy.maxQualityDrop * 100).toFixed(2)}pt` });
  }

  const requestsSufficient =
    window.control.requests >= policy.minRequestsPerArm &&
    window.treatment.requests >= policy.minRequestsPerArm;
  const sufficient = requestsSufficient && qualitySufficient;
  const hardBreach = checks.some((check) => check.severity === "hard" && !check.passed);
  const softBreach = checks.some((check) => check.severity === "soft" && !check.passed);
  return {
    stagePercent: window.stagePercent, checks, hardBreach, softBreach, sufficient,
    action: hardBreach || softBreach ? "breach" : sufficient ? "pass" : "hold",
  };
}

export interface RolloutDecision {
  action: "continue" | "hold" | "rollback";
  stoppedAtStage: number | null;
  breachStreak: number;
  windows: WindowEvaluation[];
  reasons: string[];
}

/**
 * 按时间顺序回放观测窗口。硬指标一次越界就回滚；软指标连续越界达到阈值才回滚。
 * 样本不足的窗口只返回 hold，不用“看起来没问题”代替证据。
 */
export function evaluateRollout(
  windows: readonly RolloutWindow[],
  policy: RollbackPolicy = DEFAULT_ROLLBACK,
): RolloutDecision {
  if (windows.length === 0) fail("at least one rollout window is required");
  integer(policy.minRequestsPerArm, "minRequestsPerArm", 1);
  integer(policy.minQualitySamples, "minQualitySamples", 1);
  integer(policy.consecutiveBreaches, "consecutiveBreaches", 1);

  const evaluations: WindowEvaluation[] = [];
  const reasons: string[] = [];
  let streak = 0;
  let action: RolloutDecision["action"] = "continue";
  let stoppedAtStage: number | null = null;
  let previousStart = -1;

  for (const window of windows) {
    if (window.startedAtMs < previousStart) fail("windows must be ordered by startedAtMs");
    previousStart = window.startedAtMs;
    const evaluation = evaluateWindow(window, policy);
    evaluations.push(evaluation);
    for (const check of evaluation.checks.filter((item) => !item.passed)) {
      reasons.push(`stage ${window.stagePercent}%: ${check.id} (${check.detail})`);
    }
    if (evaluation.hardBreach) {
      action = "rollback";
      stoppedAtStage = window.stagePercent;
      break;
    }
    if (evaluation.softBreach) {
      streak += 1;
      if (streak >= policy.consecutiveBreaches) {
        action = "rollback";
        stoppedAtStage = window.stagePercent;
        break;
      }
      action = "hold";
      continue;
    }
    if (evaluation.sufficient) {
      streak = 0;
      action = "continue";
    } else {
      action = "hold";
    }
  }
  return { action, stoppedAtStage, breachStreak: streak, windows: evaluations, reasons };
}

export interface RollbackContext {
  rolloutId: string; treatmentPercent: number; rollbackAtMs: number;
  inFlightRequests: number; drainPerSecond: number; graceMs: number;
  cacheKeyPrefixes: readonly string[]; downstreamTargets: readonly string[];
  idempotencyWindowMs: number;
}

export interface RollbackPlan {
  routeTreatmentPercent: 0; drainDeadlineMs: number; cacheAction: "bump-epoch";
  cacheEpoch: string; keepIdempotencyRecords: true; idempotencyWindowMs: number;
  reconcileTargets: string[]; steps: string[];
}

/**
 * 回滚计划先切流、再排空、最后对账。缓存用 epoch 提升而不是暴力删除；
 * 幂等记录保留，避免同一个键被重算成第二次副作用。
 */
export function planRollback(context: RollbackContext): RollbackPlan {
  if (typeof context.rolloutId !== "string" || context.rolloutId.trim() === "") fail("rolloutId is required");
  if (!Number.isFinite(context.treatmentPercent) || context.treatmentPercent < 0 || context.treatmentPercent > 100) {
    fail("treatmentPercent must be between 0 and 100");
  }
  integer(context.rollbackAtMs, "rollbackAtMs", 0);
  integer(context.inFlightRequests, "inFlightRequests", 0);
  if (!Number.isFinite(context.drainPerSecond) || context.drainPerSecond <= 0) fail("drainPerSecond must be > 0");
  integer(context.graceMs, "graceMs", 0);
  integer(context.idempotencyWindowMs, "idempotencyWindowMs", 0);

  const drainMs = context.inFlightRequests === 0 ? 0 : Math.ceil((context.inFlightRequests / context.drainPerSecond) * 1000);
  const drainDeadlineMs = drainMs + context.graceMs;
  const cacheEpoch = `${context.rolloutId}:${context.rollbackAtMs}`;
  const prefixes = [...new Set(context.cacheKeyPrefixes)].sort();
  const targets = [...new Set(context.downstreamTargets)].sort();
  const steps = [
    "route 100% of new traffic back to control and stop treatment assignment",
    `drain ${context.inFlightRequests} in-flight requests within ${drainDeadlineMs}ms`,
    prefixes.length > 0 ? `bump cache epoch to ${cacheEpoch} for [${prefixes.join(", ")}]` : `bump cache epoch to ${cacheEpoch}`,
    `keep idempotency records for at least ${context.idempotencyWindowMs}ms; never recompute a served key`,
    targets.length > 0 ? `reconcile downstream targets [${targets.join(", ")}] by rolloutId and arm` : "reconcile downstream targets by rolloutId and arm",
  ];
  return {
    routeTreatmentPercent: 0, drainDeadlineMs, cacheAction: "bump-epoch", cacheEpoch,
    keepIdempotencyRecords: true, idempotencyWindowMs: context.idempotencyWindowMs,
    reconcileTargets: targets, steps,
  };
}

node examples/perf/rollout.ts

decision=rollback stopped=5 streak=0
window stage=1 action=hold sufficient=false hard=false soft=false
window stage=5 action=breach sufficient=true hard=true soft=false
reason stage 5%: error-absolute (treatment error 2.12%, limit 2.00%)
reason stage 5%: error-delta (delta 1.12pt, limit 0.50pt)
assignment 5% treated=517/10000
rollback route=0 drain=7500ms epoch=exp-42:1762300900000
step route 100% of new traffic back to control and stop treatment assignment
step drain 1800 in-flight requests within 7500ms
step bump cache epoch to exp-42:1762300900000 for [chat:reply, chat:reply:v2]
step keep idempotency records for at least 86400000ms; never recompute a served key
step reconcile downstream targets [event-log, ledger] by rolloutId and arm