File size: 1,890 Bytes
dfb775d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 | """Atomic hot-swap with canary — promote canary -> live, rollback to previous.
Pure registry-state mutation; the actual vLLM router slot rotation happens
in the inference layer (out of scope for the hackathon MVP).
"""
from __future__ import annotations
from datetime import UTC, datetime
from pathlib import Path
from mindxtrain.deploy.registry import DeployRegistry, Slot
def promote_canary_to_live(registry_path: Path) -> Slot:
"""Promote `canary` -> `live` atomically; retain previous live for rollback.
Returns the new live Slot. Raises if no canary is staged.
"""
reg = DeployRegistry(registry_path)
canary = reg.get("canary")
if canary is None:
msg = "no canary slot to promote"
raise RuntimeError(msg)
current_live = reg.get("live")
if current_live is not None:
reg.set(
Slot(
name="previous_live",
run_id=current_live.run_id,
blake3=current_live.blake3,
activated_at=current_live.activated_at,
)
)
new_live = Slot(
name="live",
run_id=canary.run_id,
blake3=canary.blake3,
activated_at=datetime.now(tz=UTC),
)
reg.set(new_live)
reg.clear("canary")
return new_live
def rollback_live(registry_path: Path) -> Slot:
"""Roll `live` back to `previous_live`. Raises if no rollback target exists."""
reg = DeployRegistry(registry_path)
prev = reg.get("previous_live")
if prev is None:
msg = "no previous_live slot — nothing to roll back to"
raise RuntimeError(msg)
new_live = Slot(
name="live",
run_id=prev.run_id,
blake3=prev.blake3,
activated_at=datetime.now(tz=UTC),
)
reg.set(new_live)
reg.clear("previous_live")
return new_live
__all__ = ["promote_canary_to_live", "rollback_live"]
|