《互联网产品的 AI 架构·多能力编排与任务闭环》12.2 编排版本与灰度

// file: src/orch/rollout.ts
// tier: L2
// verify: node --test test/orch/evolution.test.ts

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

export type WorkflowVersion = "old" | "new";

/** 灰度策略：三个维度（租户、流量、任务类型）共同决定一次请求走哪版。 */
export interface RolloutPolicy {
  /** 新版本标识，用于分桶与缓存键，不是语义版本。 */
  version: string;
  /** 流量档位，必须严格递增且最后一个是 1。 */
  trafficStages: readonly number[];
  /** 当前生效档位下标。 */
  activeStage: number;
  tenantAllowlist: readonly string[];
  tenantDenylist: readonly string[];
  /** 允许进入新版本的任务类型；其余任务留在旧版本。 */
  taskTypes: readonly string[];
  cacheNamespace: string;
  previousCacheNamespace: string;
  /** 新旧版本共享的状态字段。 */
  stateFields: readonly string[];
  /** 旧版本读取状态时必须存在的字段。 */
  legacyRequiredFields: readonly string[];
  /** 新版本额外需要的字段。 */
  newRequiredFields: readonly string[];
  /** 任务最长执行时间，用于判断状态 TTL 是否足够。 */
  maxTaskMinutes: number;
  stateTtlMinutes: number;
}

export interface PolicyIssue {
  code: string;
  message: string;
}

export interface RolloutRequest {
  requestId: string;
  tenantId: string;
  taskType: string;
}

export interface RoutingDecision {
  version: WorkflowVersion;
  reason:
    | "tenant-pinned-old"
    | "tenant-pinned-new"
    | "task-type-excluded"
    | "traffic-in-stage"
    | "traffic-below-stage";
  /** 请求落在 0–9999 的哪个桶；同一版本与 requestId 永远得到同一个值。 */
  bucket: number;
  activeShare: number;
}

export interface RolloutReadiness {
  ready: boolean;
  issues: PolicyIssue[];
  activeShare: number;
}

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

function ratio(value: number, field: string): number {
  if (!Number.isFinite(value) || value <= 0 || value > 1) fail(`${field} 必须落在 (0, 1]`);
  return value;
}

function nonNegative(value: number, field: string): number {
  if (!Number.isFinite(value) || value < 0) fail(`${field} 必须是 >= 0 的有限数`);
  return value;
}

/** FNV-1a 32 位；只用于稳定分桶，不用于安全场景。 */
function hash32(input: string): number {
  let hash = 0x811c9dc5;
  for (let index = 0; index < input.length; index += 1) {
    hash ^= input.charCodeAt(index);
    hash = Math.imul(hash, 0x01000193) >>> 0;
  }
  return hash >>> 0;
}

/** 同一 version 与 requestId 的分桶结果稳定，重放不会换版本。 */
export function trafficBucket(version: string, requestId: string): number {
  if (version.trim() === "" || requestId.trim() === "") fail("version 与 requestId 不能为空");
  return hash32(`${version}:${requestId}`) % 10_000;
}

/** 结构性问题：档位、名单与任务类型。它们不涉及状态兼容，但会让灰度失真。 */
export function policyIssues(policy: RolloutPolicy): PolicyIssue[] {
  const issues: PolicyIssue[] = [];
  if (policy.trafficStages.length < 3) {
    issues.push({ code: "stage-count", message: "至少需要 3 个流量档位，否则无法观察爬坡过程" });
  }
  policy.trafficStages.forEach((stage, index) => {
    if (!Number.isFinite(stage) || stage <= 0 || stage > 1) {
      issues.push({ code: "stage-range", message: `第 ${index} 档 ${stage} 必须落在 (0, 1]` });
    }
    if (index > 0 && stage <= policy.trafficStages[index - 1]) {
      issues.push({ code: "stage-order", message: `第 ${index} 档 ${stage} 必须大于前一档 ${policy.trafficStages[index - 1]}` });
    }
  });
  if (policy.trafficStages.at(-1) !== 1) {
    issues.push({ code: "stage-final", message: "最后一档必须是 1，否则新版本永远无法全量" });
  }
  if (!Number.isInteger(policy.activeStage) || policy.activeStage < 0 || policy.activeStage >= policy.trafficStages.length) {
    issues.push({ code: "active-stage", message: `activeStage ${policy.activeStage} 越界` });
  }
  const allowed = new Set(policy.tenantAllowlist);
  for (const tenant of policy.tenantDenylist) {
    if (allowed.has(tenant)) issues.push({ code: "tenant-overlap", message: `租户 ${tenant} 同时在允许与排除名单` });
  }
  if (policy.taskTypes.length === 0) {
    issues.push({ code: "no-task-types", message: "至少指定一种允许进入新版本的任务类型" });
  }
  if (new Set(policy.taskTypes).size !== policy.taskTypes.length) {
    issues.push({ code: "duplicate-task-type", message: "任务类型不能重复" });
  }
  if (policy.cacheNamespace.trim() === "" || policy.previousCacheNamespace.trim() === "") {
    issues.push({ code: "cache-namespace-empty", message: "新旧版本都必须有缓存命名空间" });
  }
  return issues;
}

/**
 * 兼容性只回答一件事：新旧版本并行时，状态与缓存会不会互相读错。
 * 两个方向都要检查：旧版本读新状态，新版本读旧状态。
 */
export function compatibilityIssues(policy: RolloutPolicy): PolicyIssue[] {
  const issues: PolicyIssue[] = [];
  const state = new Set(policy.stateFields);
  const duplicates = policy.stateFields.filter((field, index) => policy.stateFields.indexOf(field) !== index);
  for (const field of new Set(duplicates)) {
    issues.push({ code: "duplicate-state-field", message: `状态字段 ${field} 重复声明` });
  }
  for (const field of policy.newRequiredFields) {
    if (!state.has(field)) {
      issues.push({ code: "new-state-missing", message: `新版本需要 ${field}，但共享状态没有该字段` });
    }
  }
  for (const field of policy.legacyRequiredFields) {
    if (!state.has(field)) {
      issues.push({ code: "legacy-state-missing", message: `旧版本需要 ${field}，新版本不能停止写入` });
    }
  }
  if (policy.cacheNamespace === policy.previousCacheNamespace) {
    issues.push({ code: "cache-collision", message: "新旧版本共用缓存命名空间，语义变化会命中旧结果" });
  }
  nonNegative(policy.maxTaskMinutes, "maxTaskMinutes");
  if (policy.stateTtlMinutes < policy.maxTaskMinutes) {
    issues.push({ code: "state-ttl-short", message: `状态 TTL ${policy.stateTtlMinutes} 分钟短于任务上限 ${policy.maxTaskMinutes} 分钟` });
  }
  return issues;
}

export function rolloutReadiness(policy: RolloutPolicy): RolloutReadiness {
  const issues = [...policyIssues(policy), ...compatibilityIssues(policy)];
  const activeShare = policy.trafficStages[policy.activeStage] ?? 0;
  return { ready: issues.length === 0, issues, activeShare };
}

function assertActiveStage(policy: RolloutPolicy): number {
  ratio(policy.trafficStages[policy.activeStage] ?? Number.NaN, "activeShare");
  if (policy.tenantAllowlist.some((tenant) => policy.tenantDenylist.includes(tenant))) {
    fail("同一租户不能同时在允许与排除名单");
  }
  return policy.trafficStages[policy.activeStage];
}

/**
 * 三维分流：租户名单优先于任务类型，任务类型优先于流量。
 * 排除名单永远赢：被钉在旧版本的租户不会因为分桶落进新版本。
 */
export function decideVersion(policy: RolloutPolicy, request: RolloutRequest): RoutingDecision {
  const activeShare = assertActiveStage(policy);
  const bucket = trafficBucket(policy.version, request.requestId);
  const base = { bucket, activeShare };
  if (policy.tenantDenylist.includes(request.tenantId)) {
    return { ...base, version: "old", reason: "tenant-pinned-old" };
  }
  if (policy.tenantAllowlist.includes(request.tenantId)) {
    return { ...base, version: "new", reason: "tenant-pinned-new" };
  }
  if (!policy.taskTypes.includes(request.taskType)) {
    return { ...base, version: "old", reason: "task-type-excluded" };
  }
  const threshold = Math.round(activeShare * 10_000);
  return bucket < threshold
    ? { ...base, version: "new", reason: "traffic-in-stage" }
    : { ...base, version: "old", reason: "traffic-below-stage" };
}

/** 缓存键把版本写进命名空间与键本身，双重隔离，避免语义变化后命中旧结果。 */
export function cacheKeyFor(policy: RolloutPolicy, tenantId: string, taskType: string): string {
  if (tenantId.trim() === "" || taskType.trim() === "") fail("tenantId 与 taskType 不能为空");
  return `${policy.cacheNamespace}:v${policy.version}:${tenantId}:${taskType}`;
}

/** 下一档流量比例；已经全量时返回 null。 */
export function nextTrafficStage(policy: RolloutPolicy): number | null {
  const next = policy.activeStage + 1;
  return next < policy.trafficStages.length ? policy.trafficStages[next] : null;
}