Blue-Green, Canary, and Shadow Deployments for ML Models

Blue-Green, Canary, and Shadow Deployments for ML Models

A production-oriented guide using a social-media feed-ranking system

Blue-green, canary, and shadow deployments reduce the risk of releasing a new machine-learning model by controlling when and how a candidate's behaviour meets production traffic. On a social-media platform they protect feed ranking, recommendations and model-backed moderation from regressions in latency, reliability, relevance, fairness, safety and business outcomes.

This guide develops a reference design for a feed-ranking service covering validation, stable cohort assignment, statistically sound promotion gates, bounded telemetry, failure isolation and rollback. Every snippet has been executed against tests, including its failure paths.

Scope. These are executable reference patterns, not a deployable social network; production also needs infrastructure-as-code, service identity, secrets management, capacity planning, privacy controls and operational ownership. The demo model is deliberately trivial.

1. Running example: a social-media feed

A user opens the app. The feed service obtains eligible candidate posts, fetches features, scores candidates, applies product and safety constraints and returns an ordered list. The current model is feed-ranker-v42; the candidate is feed-ranker-v43. Features include engagement aggregates, recency, content type, creator affinity and quality signals; the client receives ordered post IDs, not internal scores. Release risks include latency or error regressions, stale features, poor cold-start behaviour, reduced diversity, harmful exposure and lower user value.

The model ranks only candidates the platform has already decided may be shown. Authentication, block relationships, deletion state, age and region restrictions and moderation decisions are enforced by trusted platform services, never delegated to a ranking model.

2. Deployment patterns at a glance

A common sequence is offline evaluation, then shadow, a small canary, staged rollout and full release. Blue-green can supply the environment switch and rollback mechanism at any stage. The strategies are complementary.

3. Shared reference code

The request type validates itself on construction: equal-length feature arrays, unique non-empty post IDs, finite values in range, and booleans for flags. Malformed candidate sets are rejected instead of being silently truncated by zip(). Ties break on post ID so every replica returns the same order.

# deployment_models.py
from __future__ import annotations
import random
from dataclasses import dataclass
from math import isfinite
def _all_in(values, lo: float, hi: float) -> bool:
    return all(isinstance(v, (int, float)) and isfinite(v) and lo <= v <= hi for v in values)
@dataclass(frozen=True)
class FeedRequest:
    user_key: str                      # opaque, stable, pseudonymous identifier
    post_ids: tuple[str, ...]
    engagement_rates: tuple[float, ...]
    recency_hours: tuple[float, ...]
    is_video: tuple[bool, ...]
    def __post_init__(self) -> None:   # validate on construction, not on every predict
        n = len(self.post_ids)
        if not self.user_key:
            raise ValueError("user_key is required")
        if n == 0 or len(set(self.post_ids)) != n:
            raise ValueError("post_ids must be non-empty and unique")
        if not len(self.engagement_rates) == len(self.recency_hours) == len(self.is_video) == n:
            raise ValueError("candidate feature arrays must have equal lengths")
        if not _all_in(self.engagement_rates, 0.0, 1.0):
            raise ValueError("engagement rates must be finite and in [0, 1]")
        if not _all_in(self.recency_hours, 0.0, float("inf")):
            raise ValueError("recency must be finite and non-negative")
        if not all(isinstance(v, bool) for v in self.is_video):
            raise ValueError("is_video must be boolean")
@dataclass(frozen=True)
class FeedRanking:
    model_version: str
    ranked_post_ids: tuple[str, ...]
class FeedRanker:
    """Demo ranker; replace with a loaded, digest-verified model artifact."""
    def __init__(self, version: str, fail_rate: float = 0.0) -> None:
        if not 0.0 <= fail_rate <= 1.0:
            raise ValueError("fail_rate must be in [0, 1]")
        self.version, self.fail_rate = version, fail_rate
    def predict(self, request: FeedRequest) -> FeedRanking:
        if self.fail_rate and random.random() < self.fail_rate:
            raise RuntimeError("simulated_inference_failure")
        scores = {
            pid: eng * 100.0 - rec * 2.0 + vid * 5.0
            for pid, eng, rec, vid in zip(request.post_ids, request.engagement_rates,
                                          request.recency_hours, request.is_video, strict=True)
        }
        # Tie-break on post_id so the ordering is deterministic across replicas.
        ranked = tuple(sorted(scores, key=lambda pid: (-scores[pid], pid)))
        return FeedRanking(self.version, ranked)

4. Blue-green deployment

Blue serves; green holds the candidate. Green passes smoke, contract, dependency and load tests before traffic moves. The switch belongs to a shared control plane (service mesh, gateway, load balancer or orchestrator), not to a variable inside one API process, which cannot move traffic across replicas or regions atomically.

Explicit rollback target

The controller records the last known-good slot when it activates a new one. Rollback restores that slot and then clears it, so a second rollback is a no-op rather than a switch back to the faulty release. Simply swapping active and previous does exactly that, toggling forward on a repeated call. Health is checked live at switch time instead of trusted from a stored flag, and staging a new release into the slot that is serving or holds the rollback target is refused until the rollback window is closed explicitly.

# blue_green_control.py
from __future__ import annotations
from dataclasses import dataclass
from threading import Lock
from typing import Callable
@dataclass(frozen=True)
class Release:
    release_id: str
    model_version: str
    artifact_digest: str
class BlueGreenControl:
    """Control-plane decision model. In production, persist this state in a strongly
    consistent store (or use the platform's atomic traffic shift) and verify the data plane."""
    def __init__(self, blue: Release, health_check: Callable[[Release], bool]) -> None:
        self._lock = Lock()
        self._slots: dict[str, Release | None] = {"blue": blue, "green": None}
        self._active = "blue"
        self._known_good: str | None = None    # rollback target while the window is open
        self._health_check = health_check
    def active(self) -> Release:
        with self._lock:
            return self._slots[self._active]  # type: ignore[return-value]
    def stage(self, slot: str, release: Release) -> None:
        with self._lock:
            if slot not in self._slots:
                raise ValueError("unknown slot")
            if slot in (self._active, self._known_good):
                raise RuntimeError(f"{slot} is serving or is the rollback target")
            self._slots[slot] = release
    def activate(self, slot: str) -> Release:
        with self._lock:
            target = self._slots.get(slot)
            if target is None:
                raise RuntimeError("nothing staged in target slot")
            if slot == self._active:
                return target                  # idempotent
            if not self._health_check(target):  # live check at switch time, not a stored flag
                raise RuntimeError("target failed health gates")
            self._known_good, self._active = self._active, slot
            return target
    def rollback(self) -> Release:
        with self._lock:
            if self._known_good is None:       # already rolled back, or window closed
                return self._slots[self._active]  # type: ignore[return-value]
            target = self._slots[self._known_good]
            if target is None or not self._health_check(target):
                raise RuntimeError("rollback target is not healthy; page the on-call owner")
            self._active, self._known_good = self._known_good, None  # never toggles forward
            return target
    def close_rollback_window(self) -> None:
        with self._lock:
            self._known_good = None            # the idle slot may now be reused

Production requirements. Deploy identical runtime configuration to both environments and version model artifact, feature transformations, schema, dependencies and serving config as one release unit. Keep blue warm at full capacity through the rollback window, verify the data plane after each switch, and check database and event-schema compatibility, because a model rollback is unsafe if a forward migration broke the old code. Record release ID, artifact and config digests, approver, timestamps and rollback reason.

5. Canary deployment

Stable, experiment-scoped cohort assignment

Assign a user, account or session consistently to an arm; choosing randomly per request exposes one person to both models and contaminates measurement. An HMAC of the subject key gives deterministic assignment without a lookup table. Include the experiment ID in the hashed message: hashing the user alone puts the same users in the first few percent of every canary, so they carry the risk of every release and their accumulated exposure biases later experiments. Raising the rollout percentage keeps earlier users in the cohort. Keep the secret in a secrets manager and version it, because rotating it reshuffles cohorts.

# canary_routing.py
import hashlib
import hmac
BUCKETS = 10_000  # 0.01% granularity
def bucket_for(experiment_id: str, subject_key: str, secret: bytes) -> int:
    if not (experiment_id and subject_key and secret):
        raise ValueError("experiment_id, subject_key and secret are required")
    # Salting with the experiment ID gives each rollout an independent cohort; without it
    # the same users land in the first few percent of every canary.
    msg = f"{experiment_id}\x1f{subject_key}".encode()
    return int.from_bytes(hmac.new(secret, msg, hashlib.sha256).digest()[:8], "big") % BUCKETS
def use_candidate(experiment_id: str, subject_key: str, secret: bytes,
                  rollout_percent: float) -> bool:
    if not 0.0 <= rollout_percent <= 100.0:
        raise ValueError("rollout_percent must be in [0, 100]")
    # round(), not int(): int(0.29 * 100) == 28. Raising the percentage keeps
    # earlier users in the canary, so cohorts only ever grow during a rollout.
    return bucket_for(experiment_id, subject_key, secret) < round(rollout_percent * BUCKETS / 100)

Gateway weight-based splitting (weighted backendRefs) is per request, not per user: for a sticky canary, route on a header set from this assignment or assign in the service. Where social interactions create spillover, randomising by account cluster, creator cluster or geography may be more valid than by user.

Promotion gates

A typical progression is 1%, 5%, 10%, 25%, 50%, 100%, with steps and dwell times set by traffic volume, risk and statistical power. The policy compares the arms over the same window (absolute thresholds alone pass a candidate at 0.9% errors against a stable 0.1%). It separates "wait" from "roll back"; waits on missing telemetry (NaN compares false and would pass every gate); rolls back on a severe breach before the minimum sample; enforces dwell time; and promotes only when the upper confidence bound on the error-rate increase is within the margin.

# rollout_policy.py
from __future__ import annotations
from dataclasses import dataclass
from enum import StrEnum
from math import isfinite, sqrt
class Decision(StrEnum):
    WAIT = "wait"                        # keep collecting; do not change traffic
    ROLLBACK = "rollback"                # hard guardrail breached
    ELIGIBLE = "eligible_for_review"     # not auto-promote: approval still required
@dataclass(frozen=True)
class ArmMetrics:                        # same window, concurrent arms
    requests: int
    errors: int
    p95_latency_ms: float                # from the metrics backend's histogram
    feature_errors: int
@dataclass(frozen=True)
class GatePolicy:
    minimum_requests: int = 1_000
    minimum_dwell_s: float = 3_600
    hard_error_rate: float = 0.02         # absolute stop condition
    max_error_rate_increase: float = 0.002
    max_p95_regression_ms: float = 15.0
    max_feature_error_rate: float = 0.005
    z: float = 2.326                      # one-sided 99%
def _upper_bound_diff(c_err: int, c_n: int, s_err: int, s_n: int, z: float) -> float:
    """Upper confidence bound on (candidate - stable) error rate, normal approximation."""
    pc, ps = c_err / c_n, s_err / s_n
    se = sqrt(pc * (1 - pc) / c_n + ps * (1 - ps) / s_n)
    return (pc - ps) + z * se
def evaluate(candidate: ArmMetrics, stable: ArmMetrics, elapsed_s: float,
             policy: GatePolicy = GatePolicy()) -> tuple[Decision, str]:
    values = (candidate.p95_latency_ms, stable.p95_latency_ms, elapsed_s)
    if not all(isfinite(v) for v in values):  # NaN compares False and would pass every gate
        return Decision.WAIT, "missing_or_invalid_telemetry"
    if min(candidate.requests, stable.requests) <= 0:
        return Decision.WAIT, "no_traffic"
    c_rate = candidate.errors / candidate.requests
    if c_rate > policy.hard_error_rate:  # severe breach: act without waiting for sample size
        return Decision.ROLLBACK, "hard_error_rate_breach"
    if min(candidate.requests, stable.requests) < policy.minimum_requests:
        return Decision.WAIT, "insufficient_sample"
    if candidate.feature_errors / candidate.requests > policy.max_feature_error_rate:
        return Decision.ROLLBACK, "feature_quality_breach"
    if candidate.p95_latency_ms - stable.p95_latency_ms > policy.max_p95_regression_ms:
        return Decision.ROLLBACK, "latency_regression"
    if c_rate - stable.errors / stable.requests > policy.max_error_rate_increase:
        return Decision.ROLLBACK, "error_rate_regression"
    if elapsed_s < policy.minimum_dwell_s:
        return Decision.WAIT, "minimum_dwell_not_reached"
    # Promote only if we are confident the increase is within the margin, not merely
    # because the point estimate has not crossed it yet.
    ub = _upper_bound_diff(candidate.errors, candidate.requests,
                           stable.errors, stable.requests, policy.z)
    if ub > policy.max_error_rate_increase:
        return Decision.WAIT, "error_rate_not_yet_conclusive"
    return Decision.ELIGIBLE, "all_gates_passed"

Note. "Eligible for review" deliberately does not mean auto-promote: approvals, safety guardrails and an authorised traffic change still apply. If gates are evaluated repeatedly, the confidence bound must use a sequential method (alpha spending or always-valid intervals) to keep the false-promotion rate honest. Log experiment and assignment versions with actual exposure, and analyse slices (new users, regions, languages, devices).

6. Shadow deployment

Shadow copies sampled production requests to the candidate while the stable model remains the only source of user-visible output. Shadow results must never reach the client or trigger writes, notifications, moderation actions or billing, and a slow or failing candidate must not add latency to the primary path.

Mirror at the gateway where possible

Percentage-based request mirroring is in the Gateway API standard channel from v1.3: an HTTPRoute rule keeps v42 as its backendRef and adds a RequestMirror filter whose backendRef is the v43 shadow service with percent: 5 (or a fraction for sub-percent rates). Service meshes offer the same. The gateway discards mirrored responses and the candidate runs on its own pods, so isolation comes from infrastructure. Mirror only read-only endpoints, and remember the mirror receives the full request, including personal data.

Application-level shadow with comparison

When the comparison needs the primary result, dispatch from the service. The dispatcher is asyncio-native, never blocks the caller, bounds the queue, applies a per-job timeout with asyncio.timeout, counts drops and failures as metrics rather than logging every drop, and drains with a deadline on shutdown. Inference goes to a separately deployed candidate: running candidate inference in a thread of the serving process competes for the same CPU and is not isolation.

# shadow_dispatch.py
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass
from typing import Awaitable, Callable
from prometheus_client import Counter, Histogram
from deployment_models import FeedRanking, FeedRequest
from ranking_metrics import rbo
log = logging.getLogger(__name__)
SHADOW = Counter("shadow_jobs_total", "Shadow jobs by outcome", ["outcome"])
AGREEMENT = Histogram("shadow_rbo", "RBO of shadow vs. primary top-k",
                      buckets=(0.5, 0.7, 0.8, 0.9, 0.95, 0.99, 1.0))
@dataclass(frozen=True)
class ShadowJob:
    request_id: str
    request: FeedRequest           # minimised, pseudonymous feature payload
    primary: FeedRanking           # what users actually saw, for comparison
class ShadowDispatcher:
    """Best-effort, bounded, read-only. `infer` must call an *isolated* candidate
    deployment (its own pods/quota), so shadow load cannot starve the primary path."""
    def __init__(self, infer: Callable[[FeedRequest], Awaitable[FeedRanking]],
                 capacity: int = 500, workers: int = 4, timeout_s: float = 0.5) -> None:
        if capacity < 1 or workers < 1 or timeout_s <= 0:
            raise ValueError("capacity, workers and timeout must be positive")
        self._infer, self._timeout = infer, timeout_s
        self._queue: asyncio.Queue[ShadowJob] = asyncio.Queue(maxsize=capacity)
        self._workers = [asyncio.create_task(self._run()) for _ in range(workers)]
    def submit(self, job: ShadowJob) -> bool:
        try:
            self._queue.put_nowait(job)   # never block the user-facing request
            return True
        except asyncio.QueueFull:
            SHADOW.labels("dropped").inc()  # count, don't log per drop under saturation
            return False
    async def _run(self) -> None:
        while True:
            job = await self._queue.get()
            try:
                async with asyncio.timeout(self._timeout):
                    shadow = await self._infer(job.request)
                AGREEMENT.observe(rbo(job.primary.ranked_post_ids, shadow.ranked_post_ids, depth=20))
                SHADOW.labels("ok").inc()
            except TimeoutError:
                SHADOW.labels("timeout").inc()
            except Exception:
                SHADOW.labels("error").inc()
                log.exception("shadow_failed request_id=%s", job.request_id)
            finally:
                self._queue.task_done()
    async def close(self, drain_s: float = 2.0) -> None:
        try:
            async with asyncio.timeout(drain_s):
                await self._queue.join()
        except TimeoutError:
            SHADOW.labels("abandoned").inc(self._queue.qsize())
        for w in self._workers:
            w.cancel()
        await asyncio.gather(*self._workers, return_exceptions=True)

Agreement is measured with rank-biased overlap, which is top-weighted: swapping the first two posts lowers it more than swapping the last two. Set overlap (Jaccard) scores both swaps as identical, perfect agreement. Empty rankings raise an error so they are counted separately rather than scored.

# ranking_metrics.py
from __future__ import annotations
from typing import Sequence
def rbo(a: Sequence[str], b: Sequence[str], p: float = 0.9, depth: int | None = None) -> float:
    """Extrapolated rank-biased overlap (Webber et al., 2010) of two top-k lists.
    1.0 = same order, 0.0 = disjoint. p sets top-weighting (p=0.9 ~ top 10 carries 86%)."""
    if not 0.0 < p < 1.0:
        raise ValueError("p must be in (0, 1)")
    if len(set(a)) != len(a) or len(set(b)) != len(b):
        raise ValueError("rankings must not contain duplicates")
    k = min(len(a), len(b)) if depth is None else min(depth, len(a), len(b))
    if k == 0:
        raise ValueError("empty ranking: report it separately, do not score it")
    seen_a, seen_b, overlap, weighted = set(), set(), 0, 0.0
    for d in range(1, k + 1):
        x, y = a[d - 1], b[d - 1]
        if x == y:
            overlap += 1
        else:
            overlap += (x in seen_b) + (y in seen_a)
        seen_a.add(x)
        seen_b.add(y)
        weighted += (overlap / d) * p ** d
    return (overlap / k) * p ** k + (1 - p) / p * weighted

What to compare. Errors, timeouts, model-only latency, feature parity, score drift by cohort, RBO and NDCG@K (with denominators), and shadow cost, queue depth and drops. Safety predictions are evaluated offline and never trigger enforcement.

7. Production feed architecture

Reference feed-ranking service

The service brings the patterns together. Identity comes from a verified RS256 token, never from a user_key in the request body, so one caller cannot rank, or enter an experiment, as another. Inference runs off the event loop under a deadline and a semaphore bounding in-flight work; a candidate failure or timeout falls back to stable, and the response records the arm actually served (stable, candidate or fallback) so exposure logs separate assignment from execution. Latency goes to a histogram by model version and arm, and sampled stable traffic goes to the shadow dispatcher without waiting.

# feed_api.py
# Dependencies: fastapi, uvicorn, pydantic>=2, pyjwt[crypto], prometheus-client
from __future__ import annotations
import asyncio
import logging
import os
import re
from contextlib import asynccontextmanager
from typing import Annotated
from uuid import uuid4
import jwt
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from prometheus_client import Histogram
from pydantic import BaseModel, ConfigDict, Field, field_validator
from canary_routing import use_candidate
from deployment_models import FeedRanker, FeedRanking, FeedRequest
from shadow_dispatch import ShadowDispatcher, ShadowJob
log = logging.getLogger("feed_api")
REQUEST_ID = re.compile(r"^[A-Za-z0-9-]{8,64}$")
LATENCY = Histogram("feed_rank_seconds", "Ranking latency", ["model_version", "variant"])
class Candidate(BaseModel):
    model_config = ConfigDict(extra="forbid")
    post_id: str = Field(min_length=1, max_length=128)
    engagement_rate: float = Field(ge=0.0, le=1.0)
    recency_hours: float = Field(ge=0.0, le=24 * 365)
    is_video: bool
class FeedInput(BaseModel):            # no user ID here: identity comes from the token
    model_config = ConfigDict(extra="forbid")
    candidates: list[Candidate] = Field(min_length=1, max_length=500)
    @field_validator("candidates")
    @classmethod
    def unique_post_ids(cls, value: list[Candidate]) -> list[Candidate]:
        if len({c.post_id for c in value}) != len(value):
            raise ValueError("candidate post IDs must be unique")
        return value
class FeedOutput(BaseModel):
    request_id: str
    model_version: str                 # the model that actually served (exposure logging)
    variant: str                       # assigned arm: stable | candidate | fallback
    ranked_post_ids: list[str]
@asynccontextmanager
async def lifespan(app: FastAPI):
    s = app.state
    s.stable = FeedRanker("feed-ranker-v42")      # load + verify digests + warm up
    s.candidate = FeedRanker("feed-ranker-v43")
    s.experiment_id, s.rollout_percent = "feed-ranker-v43-canary", 1.0
    s.assignment_secret = os.environ["CANARY_ASSIGNMENT_SECRET"].encode()
    s.jwt_key, s.jwt_issuer, s.jwt_audience = (os.environ["FEED_JWT_PUBLIC_KEY"],
                                               os.environ["FEED_JWT_ISSUER"], "feed-api")
    s.deadline_s, s.slots = 0.15, asyncio.Semaphore(64)  # bounded in-flight inference
    s.shadow = ShadowDispatcher(infer=_isolated_shadow_client, capacity=500)
    s.shadow_percent = 5.0
    yield
    await s.shadow.close()
async def _isolated_shadow_client(req: FeedRequest) -> FeedRanking:
    raise NotImplementedError("call the separately deployed candidate service here")
app = FastAPI(lifespan=lifespan)
bearer = HTTPBearer(auto_error=False)
@app.middleware("http")
async def request_context(request: Request, call_next):
    supplied = request.headers.get("x-request-id", "")
    request.state.request_id = supplied if REQUEST_ID.fullmatch(supplied) else str(uuid4())
    response = await call_next(request)
    response.headers["x-request-id"] = request.state.request_id
    return response
def user_key(request: Request,
             creds: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)]) -> str:
    s = request.app.state
    if creds is None:
        raise HTTPException(401, "authentication_required", {"WWW-Authenticate": "Bearer"})
    try:
        claims = jwt.decode(creds.credentials, s.jwt_key, algorithms=["RS256"],
                            audience=s.jwt_audience, issuer=s.jwt_issuer,
                            options={"require": ["exp", "sub"]}, leeway=30)
    except jwt.PyJWTError:
        raise HTTPException(401, "invalid_token", {"WWW-Authenticate": "Bearer"}) from None
    return claims["sub"]  # use a pseudonymous subject, never a raw handle or email
async def _rank(ranker: FeedRanker, req: FeedRequest, variant: str, deadline_s: float,
                slots: asyncio.Semaphore) -> FeedRanking:
    async with asyncio.timeout(deadline_s):          # covers waiting for a slot too
        async with slots:
            with LATENCY.labels(ranker.version, variant).time():
                return await asyncio.to_thread(ranker.predict, req)
@app.get("/health/live")
def live():
    return {"status": "alive"}
@app.get("/health/ready")                           # ready only once models are loaded and warm
def ready(request: Request):
    if (stable := getattr(request.app.state, "stable", None)) is None:
        raise HTTPException(503, "model_not_ready")
    return {"status": "ready", "model_version": stable.version}
@app.post("/v1/feed/rank", response_model=FeedOutput)
async def rank_feed(payload: FeedInput, request: Request, user: Annotated[str, Depends(user_key)]):
    # Integration point: candidates must already have passed the authoritative
    # eligibility service (blocks, deletion, privacy, age/region, moderation).
    s, rid = request.app.state, request.state.request_id
    req = FeedRequest(
        user_key=user, post_ids=tuple(c.post_id for c in payload.candidates),
        engagement_rates=tuple(c.engagement_rate for c in payload.candidates),
        recency_hours=tuple(c.recency_hours for c in payload.candidates),
        is_video=tuple(c.is_video for c in payload.candidates))
    variant = ("candidate" if use_candidate(s.experiment_id, user, s.assignment_secret,
                                            s.rollout_percent) else "stable")
    ranking = None
    if variant == "candidate":
        try:
            ranking = await _rank(s.candidate, req, variant, s.deadline_s, s.slots)
        except Exception:                            # TimeoutError included
            log.warning("candidate_failed request_id=%s", rid)
            variant = "fallback"                     # serve stable; log the exposure honestly
    if ranking is None:
        try:
            ranking = await _rank(s.stable, req, variant, s.deadline_s, s.slots)
        except Exception:
            log.exception("ranking_failed request_id=%s", rid)
            raise HTTPException(503, "ranking_temporarily_unavailable") from None
    if variant == "stable" and use_candidate("shadow-" + s.experiment_id, user,
                                             s.assignment_secret, s.shadow_percent):
        s.shadow.submit(ShadowJob(rid, req, ranking))
    return FeedOutput(request_id=rid, model_version=ranking.model_version, variant=variant,
                      ranked_post_ids=list(ranking.ranked_post_ids))

Before production: replace the in-process rankers with digest-verified artifacts warmed before readiness; call the eligibility service and feature store with their own deadlines; give the stable fallback the remaining request budget rather than a fresh one; and add tracing, circuit breakers with bounded retries, and a deterministic policy-compliant fallback ordering for when every model is unavailable. Run locally with uvicorn feed_api:app --host 127.0.0.1; in production, bind behind an internal gateway.

8. Monitoring, rollback, and safe degradation

Monitor service health (errors, timeouts, p50/p95/p99, saturation); model and feature health (score distribution, calibration, missing or stale features, schema mismatch), since a healthy model with broken inputs still ranks badly; ranking quality and safety (NDCG@K, diversity, harmful exposure, reports); business outcomes (qualified interactions, retention); and release metadata (versions, cohort, stage, approval, rollback reason).

Compute percentiles in the metrics backend from histograms, not by sorting per-replica lists, and keep labels low-cardinality: model version, arm and stage are fine; user, post and request IDs are not. Degrade safely: a failing canary falls back to stable for its cohort; missing features use only documented, evaluated defaults; access and privacy checks fail closed; saturated shadow workers drop work and count it; and if every model is down, serve a deterministic, policy-compliant ordering of already eligible candidates.

Rollback is an explicit, authorised control-plane operation that must not depend on rebuilding the old model or reversing a migration mid-incident: detect the breach or receive the decision, freeze promotion, shift traffic to the known-good release, verify traffic split, errors, latency and readiness on it, keep the candidate and its telemetry for diagnosis, and close with a post-incident review that adds the missing test or alert.

9. CI/CD and release governance

The pipeline validates data contracts, feature transformations, the model artifact, the serving image, API contracts, policy behaviour and deployment configuration. It uses the current major versions of the GitHub actions and points the pip cache at the lockfile; without cache-dependency-path, setup-python looks for requirements.txt and the cache step fails.

# .github/workflows/ml-release.yml
name: ml-release-validation
on:
  pull_request:
  push:
    branches: [main]
permissions:
  contents: read
jobs:
  validate:
    runs-on: ubuntu-latest
    timeout-minutes: 30
    steps:
      - uses: actions/checkout@v7
        with:
          persist-credentials: false
      - uses: actions/setup-python@v7
        with:
          python-version: "3.13"
          cache: pip
          cache-dependency-path: requirements.lock  # default only looks for requirements.txt
      - name: Install locked dependencies
        run: python -m pip install --require-hashes -r requirements.lock
      - name: Lint, types, tests, artifact validation (digest, schema, offline gates)
        run: |
          ruff check . && mypy .
          pytest -q tests/unit tests/contracts tests/features tests/model
          python scripts/validate_model_artifact.py
      - name: Dependency audit
        run: pip-audit --require-hashes -r requirements.lock
      - name: Build serving image
        run: docker build --pull -t feed-ranker:${{ github.sha }} .
      # Then: image scan, SBOM, sign, push by digest; promote that same digest through
      # staging, shadow and canary behind protected environments.

Release evidence. Build once and promote the same image digest through every environment; never rebuild for production. Pin actions to commit SHAs, use protected environments and short-lived federated credentials. Each release carries commit, lockfile, image and model digests, feature schema and model card; slice-level offline evaluation (quality, safety, calibration, fairness); load and failure-mode tests; shadow report, canary plan, guardrails and rollback owner; and approval and traffic-change records.

10. Choosing the right strategy

Takeaway. For v43: offline validation; shadow 1–10% of eligible requests to an isolated deployment; canary a stable 1% cohort; expand only after predeclared gates pass; keep v42 as the rollback target for an agreed window. Deployment safety is not just a routing decision: it is an end-to-end property of data, features, model, serving runtime, policy layer, experiment design, telemetry and a rehearsed operational process.

Further reading: AWS Prescriptive Guidance, MLOps checklist: continuous deployment; Kubernetes Gateway API v1.3 release notes (kubernetes.io/blog/2025/06/02/gateway-api-v1-3/); MLflow Model Registry; Webber, Moffat and Zobel, "A similarity measure for indefinite rankings", ACM TOIS 28(4), 2010.

Back to all blogs