Design Deployment Verification

1
public final class VerificationGateEvaluator {
2
  public GateVerdict evaluate(SliSnapshot candidate, SliSnapshot baseline, GatePolicy policy) {
3
    if (candidate.sampleCount() < policy.minSamples()) {
4
      return GateVerdict.failClosed("insufficient_samples");
5
    }
6
    if (candidate.p99LatencyMs() > baseline.p99LatencyMs() * policy.maxLatencyFactor()) {
7
      return GateVerdict.rollback("latency_regression");
8
    }
9
    if (candidate.errorRate() - baseline.errorRate() > policy.maxErrorDelta()) {
10
      return GateVerdict.rollback("error_budget_burn");
11
    }
12
    return GateVerdict.pass();
13
  }
14
}

1
async def run_smoke_suite(deployment_id: str, base_url: str, cases: list[dict]) -> dict:
2
    results = []
3
    for case in cases:
4
        resp = await http_request(
5
            method=case["method"],
6
            url=f"{base_url}{case['path']}",
7
            headers={"X-Deployment-Id": deployment_id},
8
            timeout_s=case.get("timeout", 5),
9
        )
10
        ok = resp.status < 400 and case["assert"](resp.body)
11
        results.append({"name": case["name"], "ok": ok, "status": resp.status})
12
    failed = [r for r in results if not r["ok"]]
13
    return {"passed": len(failed) == 0, "failed": failed, "results": results}

1
export type VerificationPhase = "pre-traffic" | "post-traffic" | "soak";
2
 
3
export interface VerificationRunRequest {
4
  deploymentId: string;
5
  artifactDigest: string;
6
  environment: string;
7
  phases: VerificationPhase[];
8
}
9
 
10
export async function startVerificationRun(
11
  req: VerificationRunRequest,
12
): Promise<{ runId: string; status: "queued" }> {
13
  return post("/api/v1/verification/runs", req, {
14
    headers: { "Idempotency-Key": req.deploymentId },
15
  });
16
}