src/kube_runtime.rs
//! KUBERNETES REFRESH — [`KubeRuntime::refresh`], the K8s parity of the systemd
//! [`crate::install::refresh_product`] / [`crate::service::Lifecycle::refresh`].
//!
//! **One verb, two substrates.** A product promotes a new software version
//! IDENTICALLY on Metal (systemd, `current`-flip + D-Bus restart) and on
//! Kubernetes (Deployment rolling-update over the apiserver) — the *same* stamped,
//! ordered migration model drives both. This module is the K8s half; it reuses the
//! product-owns-its-migration doctrine and the pure chain math of
//! [`crate::migrate`] wholesale, changing only the SUBSTRATE the five refresh steps
//! run on.
//!
//! # The mapping — each systemd step to its K8s equivalent
//!
//! | systemd ([`refresh_product`]) | Kubernetes ([`kube_refresh`](crate::kube_runtime::kube_refresh)) |
//! |---|---|
//! | **deliver** — unfold the new bundle into `software/<newver>` | SSA-apply the Deployment carrying the NEW container image ([`rolling_deployment`](crate::kube_runtime::rolling_deployment)) |
//! | **migrate** — replay `stamped→head` via the NEW binary's `--migrate-config` | a pre-upgrade **migration Job** per hop, image = the NEW image, args = `--migrate-config from=X to=Y` ([`migration_job`](crate::kube_runtime::migration_job)); the rollout is gated on Job success |
//! | **swap** — atomic `current` symlink flip | the Deployment **RollingUpdate** (a new ReplicaSet supersedes the old) |
//! | **restart + health** — D-Bus restart, then the TCP/HTTP probe | the readiness/liveness gates: the rollout only *completes* when the new pods are Ready ([`KubeRefreshOps::wait_rollout`](crate::kube_runtime::KubeRefreshOps::wait_rollout)) |
//! | **rollback** — flip `current` back to the prior generation + restart | re-apply the prior pod spec: k8s brings the prior **ReplicaSet/revision** back (the `rollout undo` equivalent), waited on again |
//!
//! # The migration model is UNCHANGED — only the stamp's home moves
//!
//! [`crate::migrate::migration_steps`] (the pure `stamped → head` chain diff) is
//! reused verbatim: `stamped == head` ⇒ zero hops (idempotency), a stamp outside
//! the chain is a hard error (no silent downgrade), each hop is stamped forward on
//! success so a failed run resumes precisely. The only difference from the systemd
//! side is *where the stamp lives*: on Metal it is `<data_dir>/.migration-stamp.json`
//! beside the data; on Kube it is a small **ConfigMap** the engine reads before and
//! writes after each good hop (the migration Job mounts the product's PVC so
//! `--migrate-config` reads/writes the SAME data root the running pods do).
//!
//! # Testable offline, exactly like the systemd side
//!
//! The engine ([`kube_refresh`](crate::kube_runtime::kube_refresh)) is a **pure orchestration** over the
//! [`KubeRefreshOps`](crate::kube_runtime::KubeRefreshOps) seam — the K8s analog of the systemd
//! [`crate::install::RefreshControl`] + [`crate::migrate::MigrateRunner`]
//! injections. A scripted mock drives the whole sequence — including the
//! unready-rollout rollback — with no cluster, no tokio, and no `kube` in the
//! build. The ONE production impl ([`KubeRuntime`], behind the `kubernetes`
//! feature) bridges each sync seam call to async `kube-rs` over a tokio runtime,
//! reusing the existing [`crate::k8s_apply`] engine (`apply_json_manifests` +
//! `run_job`) — **zero-shell**: no `kubectl`, no `helm`.
//!
//! [`refresh_product`]: crate::install::refresh_product
//! [`KubeRuntime`]: live::KubeRuntime
use std::time::Duration;
use anyhow::{Result, anyhow};
use serde_json::{Value, json};
use crate::migrate::migration_steps;
// ───────────────────────────── the target ─────────────────────────────
/// Which Deployment, in which namespace, which container carries the image, and
/// which product owns the `--migrate-config` verb + its schema chain. The
/// [`ServiceSpec`](crate::service::ServiceSpec)-level facts a kube refresh needs,
/// kept off the (host-oriented) spec exactly like the systemd
/// [`RefreshRequest`](crate::install::RefreshRequest) keeps its heavier inputs off
/// the `ServiceSpec`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RefreshTarget {
pub namespace: String,
/// The Deployment name (also the `app.kubernetes.io/name` selector value).
pub workload: String,
/// The container within the pod template that carries the image.
pub container: String,
/// The product whose `--migrate-config` verb the migration Job runs.
pub product: String,
/// The port the container listens on (referenced by an HTTP/TCP probe and the
/// containerPort). `None` ⇒ no port is declared.
pub port: Option<u16>,
/// The readiness/liveness gate the rollout completes against.
pub health: RolloutProbe,
/// The PVC the product keeps its data (+ stamp) in, mounted into the
/// pre-upgrade migration Job so `--migrate-config` reads/writes the SAME data
/// root the running pods do. `None` ⇒ a stateless product (the Job runs with no
/// data volume).
pub data: Option<DataVolume>,
}
impl RefreshTarget {
pub fn new(namespace: &str, workload: &str, container: &str, product: &str) -> Self {
RefreshTarget {
namespace: namespace.to_string(),
workload: workload.to_string(),
container: container.to_string(),
product: product.to_string(),
port: None,
health: RolloutProbe::None,
data: None,
}
}
pub fn port(mut self, port: u16) -> Self {
self.port = Some(port);
self
}
pub fn health(mut self, h: RolloutProbe) -> Self {
self.health = h;
self
}
pub fn data(mut self, claim: &str, mount_path: &str) -> Self {
self.data = Some(DataVolume {
claim: claim.to_string(),
mount_path: mount_path.to_string(),
});
self
}
/// The ConfigMap the engine keeps the migration stamp in: `<product>-migration-stamp`.
pub fn stamp_configmap(&self) -> String {
format!("{}-migration-stamp", self.product)
}
}
/// The product's data PVC, mounted into the migration Job (and, by the same claim,
/// into the running pods) so the migration reads/writes the live data root.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DataVolume {
pub claim: String,
pub mount_path: String,
}
/// The rollout's readiness/liveness gate — the K8s projection of the neutral
/// [`HealthProbe`](crate::service::HealthProbe) the host `Lifecycle::health` uses.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RolloutProbe {
/// No probe (the workload's own restart handles it).
None,
/// `tcpSocket: { port }`.
Tcp(u16),
/// `httpGet: { path, port }`.
Http { path: String, port: u16 },
}
// ───────────────────────── the pure manifest renderers ─────────────────────────
/// The shared pod labels (`app.kubernetes.io/name`) that tie a workload, its
/// migration Job, and its Service together — the same key the sync
/// [`KubernetesRenderer`](crate::k8s::KubernetesRenderer) uses.
pub(crate) fn name_label(workload: &str) -> Value {
json!({ "app.kubernetes.io/name": workload })
}
/// Render the readiness+liveness probe action block, or `None` for
/// [`RolloutProbe::None`] (keeps a no-probe container minimal).
pub(crate) fn probe_action(p: &RolloutProbe) -> Option<Value> {
match p {
RolloutProbe::None => None,
RolloutProbe::Tcp(port) => Some(json!({ "tcpSocket": { "port": port } })),
RolloutProbe::Http { path, port } => {
let path = if path.starts_with('/') {
path.clone()
} else {
format!("/{path}")
};
Some(json!({ "httpGet": { "path": path, "port": port } }))
}
}
}
/// **deliver + swap** — the refreshed Deployment: the NEW `image`, an explicit
/// **RollingUpdate** strategy (a new ReplicaSet supersedes the old — the analog of
/// the systemd `current` flip), and the readiness/liveness probes the rollout
/// gates completion on. SSA-applying this is the deliver step; k8s performing the
/// rolling update is the swap.
///
/// **Pure** — no I/O, no cluster — so it is golden-tested without a cluster, and it
/// is the SAME object the rollback step re-applies with the prior image to bring
/// the prior revision back.
pub fn rolling_deployment(target: &RefreshTarget, image: &str) -> Value {
let mut container = json!({
"name": target.container,
"image": image,
});
if let Some(port) = target.port {
container["ports"] = json!([{ "containerPort": port }]);
}
// Both readiness and liveness gate on the one declared endpoint — readiness
// holds the rolling update back until the new pod is serving; liveness restarts
// a hung one. Same single-home-two-views shape as the sync renderer.
if let Some(action) = probe_action(&target.health) {
container["readinessProbe"] = action.clone();
container["livenessProbe"] = action;
}
json!({
"apiVersion": "apps/v1",
"kind": "Deployment",
"metadata": {
"name": target.workload,
"namespace": target.namespace,
"labels": name_label(&target.workload),
},
"spec": {
"selector": { "matchLabels": name_label(&target.workload) },
// The explicit RollingUpdate strategy: surge one new pod, keep every old
// one until a new one is Ready (maxUnavailable 0 ⇒ zero-downtime swap).
"strategy": {
"type": "RollingUpdate",
"rollingUpdate": { "maxUnavailable": 0, "maxSurge": 1 },
},
"template": {
"metadata": { "labels": name_label(&target.workload) },
"spec": { "containers": [container] },
},
},
})
}
/// **migrate** — a pre-upgrade migration Job for ONE chain hop: the product's own
/// `--migrate-config from=X to=Y`, run from the **NEW** image (migrations are a
/// property of the incoming code, exactly like the systemd
/// [`ProductBinRunner`](crate::migrate::ProductBinRunner) execs the NEW
/// generation's binary). The rollout is gated on this Job succeeding.
///
/// `restartPolicy: Never` + `backoffLimit: 0`: a failed migration is a hard, single
/// verdict (no silent retry masking a bad hop). When the target declares a data
/// PVC it is mounted so `--migrate-config` reads/writes the SAME data root the
/// running pods do — the product forward-migrates its own data, the ship only
/// orders the hops. **Pure** — golden-tested without a cluster.
pub fn migration_job(target: &RefreshTarget, image: &str, from: &str, to: &str) -> Value {
let job_name = format!("{}-migrate-{}-{}", target.product, from, to).to_lowercase();
let mut container = json!({
"name": "migrate",
"image": image,
// Direct argv (no shell): the product's own migration verb.
"args": ["--migrate-config", format!("from={from}"), format!("to={to}")],
});
let mut pod_spec = json!({
"restartPolicy": "Never",
"containers": [Value::Null], // filled below once the container is final
});
if let Some(vol) = &target.data {
container["volumeMounts"] = json!([{ "name": "data", "mountPath": vol.mount_path }]);
pod_spec["volumes"] = json!([{
"name": "data",
"persistentVolumeClaim": { "claimName": vol.claim },
}]);
}
pod_spec["containers"] = json!([container]);
json!({
"apiVersion": "batch/v1",
"kind": "Job",
"metadata": {
"name": job_name,
"namespace": target.namespace,
"labels": {
"app.kubernetes.io/name": target.workload,
"app.kubernetes.io/component": "migrate",
},
},
"spec": {
"backoffLimit": 0,
"template": {
"metadata": {
"labels": {
"app.kubernetes.io/name": target.workload,
"app.kubernetes.io/component": "migrate",
},
},
"spec": pod_spec,
},
},
})
}
// ───────────────────────────── outcomes ─────────────────────────────
/// The terminal state of a migration Job — the K8s analog of a migration hop's
/// `Ok`/`Err` on the systemd side. Mirrors [`crate::k8s_apply::JobOutcome`] but
/// lives in the lean core so the engine + its tests compile with no `kube`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MigrationJobOutcome {
Succeeded,
Failed,
TimedOut,
}
impl MigrationJobOutcome {
pub fn ok(self) -> bool {
matches!(self, MigrationJobOutcome::Succeeded)
}
}
/// What the migration phase did — the K8s twin of [`crate::migrate::MigrateReport`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MigrationSummary {
/// The schema the data spoke before this refresh (`None` = never stamped).
pub from_stamp: Option<String>,
/// The head schema the chain targets.
pub head: String,
/// The hops actually replayed (each a migration Job), in order.
pub replayed: Vec<(String, String)>,
/// `true` when the data was already at head and nothing ran (idempotent no-op).
pub was_noop: bool,
}
/// How a [`kube_refresh`] ended. **Only [`Rolled`](KubeRefreshOutcome::Rolled) is a
/// completed promotion.** The failure shapes are kept apart because they mean
/// different things to the person on call — the same honesty the systemd
/// [`RefreshOutcome`](crate::install::RefreshOutcome) and the promote
/// [`RolloutOutcome`](crate::promote::RolloutOutcome) keep.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum KubeRefreshOutcome {
/// The new image is live and its pods are Ready.
Rolled {
from: Option<String>,
to: String,
migrated: MigrationSummary,
},
/// A pre-upgrade migration Job failed. **Nothing was delivered** — the old
/// Deployment is untouched and still serving; the stamp is left at the last
/// good schema so a retry resumes precisely. No rollback is needed because no
/// swap happened. The safe analog of a systemd migration that fails before the
/// `current` flip.
MigrationFailed {
hop: (String, String),
outcome: MigrationJobOutcome,
stamped: String,
},
/// The new image failed to become Ready and the previous revision is back. The
/// environment is serving; the deploy is not done.
RolledBack {
to_previous: String,
failure: String,
migrated: MigrationSummary,
},
/// The new image failed AND the rollback failed too. **The environment may be
/// down.** Never collapsed into "rollout failed".
RollbackFailed {
failure: String,
rollback_failure: String,
},
/// The new image failed and there was no prior image to restore. Distinct from a
/// successful rollback: nothing was brought back.
NoAnchor { failure: String },
}
impl KubeRefreshOutcome {
/// Did the refresh complete (new image live)?
pub fn is_success(&self) -> bool {
matches!(self, KubeRefreshOutcome::Rolled { .. })
}
/// One line for an operator.
pub fn summary(&self) -> String {
match self {
KubeRefreshOutcome::Rolled { from, to, migrated } => format!(
"refresh → {to} OK (from {}{})",
from.as_deref().unwrap_or("unknown"),
if migrated.was_noop {
String::new()
} else {
format!(", {} migration hop(s)", migrated.replayed.len())
}
),
KubeRefreshOutcome::MigrationFailed {
hop,
outcome,
stamped,
} => format!(
"refresh ABORTED at migration hop {}→{} ({outcome:?}) — nothing delivered, \
old Deployment still serving, stamped at `{stamped}`",
hop.0, hop.1
),
KubeRefreshOutcome::RolledBack {
to_previous,
failure,
..
} => {
format!("refresh FAILED ({failure}) — ROLLED BACK to {to_previous}")
}
KubeRefreshOutcome::RollbackFailed {
failure,
rollback_failure,
} => format!(
"refresh FAILED ({failure}) AND the ROLLBACK FAILED ({rollback_failure}) — may be DOWN"
),
KubeRefreshOutcome::NoAnchor { failure } => {
format!("refresh FAILED ({failure}) — no prior image, nothing restored")
}
}
}
/// Collapse to a `Result` for a caller that only wants "did it work". A
/// rolled-back refresh is a SAFE outcome (serving the prior image) but not a
/// success, so it is an `Err` with an honest message — same posture as the
/// systemd side.
pub fn into_result(self) -> Result<Self> {
match &self {
KubeRefreshOutcome::Rolled { .. } => Ok(self),
other => Err(anyhow!("{}", other.summary())),
}
}
}
// ───────────────────────────── the seam ─────────────────────────────
/// The K8s side effects [`kube_refresh`] cannot perform purely — read/advance the
/// stamp, read the rollback anchor, run a migration Job, apply the Deployment, and
/// wait for the rollout. Injected so the whole orchestration (migrate → deliver →
/// swap → health → rollback) runs offline against a scripted mock, exactly like the
/// systemd [`RefreshControl`](crate::install::RefreshControl) +
/// [`MigrateRunner`](crate::migrate::MigrateRunner) injections.
///
/// Sync by design — the same shape as the systemd seams — so the engine and its
/// tests pull no async runtime. The production [`KubeRuntime`](live::KubeRuntime)
/// bridges each call to async `kube-rs` over its own tokio runtime.
pub trait KubeRefreshOps {
/// The product's stamped schema version (from its migration-stamp ConfigMap).
/// `None` = never stamped (a brand-new deployment). The analog of
/// [`crate::migrate::read_stamp`].
fn read_stamp(&self) -> Result<Option<String>>;
/// Advance the stamp to `schema` (after every SUCCESSFUL hop, so a failed run
/// resumes precisely). The analog of [`crate::migrate::write_stamp`].
fn write_stamp(&self, schema: &str) -> Result<()>;
/// The image the Deployment's container runs RIGHT NOW — the rollback anchor.
/// `None` when the workload/container is absent. **Never a guess.**
fn current_image(&self) -> Result<Option<String>>;
/// Run ONE pre-upgrade migration Job to completion and report its terminal
/// state (replacing any prior Job of the same name first).
fn run_migration_job(&self, job: &Value) -> Result<MigrationJobOutcome>;
/// SSA-apply the Deployment manifest (deliver + begin the rolling update).
fn apply_deployment(&self, deployment: &Value) -> Result<()>;
/// Wait until the rollout is complete — every pod updated, ready and available
/// at the desired count — or time out. The readiness/liveness gate.
fn wait_rollout(&self, timeout: Duration) -> Result<()>;
}
// ───────────────────────────── the engine ─────────────────────────────
/// What a [`kube_refresh`] should do — the K8s twin of the systemd
/// [`RefreshRequest`](crate::install::RefreshRequest).
#[derive(Debug, Clone)]
pub struct RefreshPlan<'a> {
pub target: &'a RefreshTarget,
/// The image to promote to (the K8s [`UpgradeSource`] — an image ref, where the
/// systemd side takes a signed bundle path).
pub new_image: &'a str,
/// The product's ordered schema-migration chain (oldest → head). Its LAST entry
/// is the schema the incoming image speaks.
pub chain: &'a [String],
/// How long to wait for the new pods to become Ready.
pub rollout_wait: Duration,
/// …and for the prior revision to come back if it has to roll back.
pub rollback_wait: Duration,
}
impl<'a> RefreshPlan<'a> {
pub fn new(target: &'a RefreshTarget, new_image: &'a str, chain: &'a [String]) -> Self {
RefreshPlan {
target,
new_image,
chain,
rollout_wait: Duration::from_secs(240),
rollback_wait: Duration::from_secs(180),
}
}
pub fn rollout_wait(mut self, d: Duration) -> Self {
self.rollout_wait = d;
self
}
pub fn rollback_wait(mut self, d: Duration) -> Self {
self.rollback_wait = d;
self
}
/// The dry run: what this would do, as lines. **Pure** — no cluster call, no
/// subprocess. A preview cannot touch the world because there is no code path
/// here that could (the same shape as the systemd/promote describes).
pub fn describe(&self) -> Vec<String> {
vec![
format!(
"read the current image of {}/{} [{}] — the rollback anchor",
self.target.namespace, self.target.workload, self.target.container
),
"replay the migration chain (stamped → head) as pre-upgrade Jobs on the NEW image, \
stamping each good hop"
.to_string(),
format!(
"SSA-apply the Deployment with image {} (RollingUpdate)",
self.new_image
),
format!(
"wait for the rollout to become Ready (≤{}s)",
self.rollout_wait.as_secs()
),
format!(
"on failure: re-apply the prior revision and wait ≤{}s; if there was no anchor, \
say so rather than pretend a rollback happened",
self.rollback_wait.as_secs()
),
]
}
}
/// **REFRESH over Kubernetes** — migrate (pre-upgrade Jobs) → deliver+swap
/// (RollingUpdate) → health (readiness gate) → roll back (prior revision) on an
/// unready rollout. The K8s engine behind [`KubeRuntime::refresh`](live::KubeRuntime::refresh),
/// EXACTLY parallel to [`crate::install::refresh_product`]:
///
/// 1. **Anchor.** Read the current image FIRST (before any mutation) — a rollback
/// needs to know where it is going back to, and the moment after the apply is too
/// late to ask.
/// 2. **Migrate.** Read the stamp; compute [`migration_steps`] (`stamped → head`).
/// An absent stamp is a brand-new deployment (stamp head, run nothing). Each hop
/// runs a [`migration_job`] on the NEW image; on success the stamp advances (so a
/// failed run resumes precisely), and a **failed hop aborts BEFORE any delivery**
/// — the old Deployment is untouched and still serving ([`MigrationFailed`]).
/// 3. **Deliver + swap.** SSA-apply [`rolling_deployment`] with the new image; k8s
/// runs the rolling update.
/// 4. **Health.** Wait for the rollout to become Ready.
/// 5. **Roll back** on an unready rollout: re-apply the prior image's spec — k8s
/// brings the prior ReplicaSet/revision back — and wait again. No anchor ⇒
/// [`NoAnchor`]; a failed rollback ⇒ [`RollbackFailed`] (the environment may be
/// down — never collapsed into "rollout failed").
///
/// **Idempotent:** re-refreshing the same image re-applies the same Deployment (a
/// no-op rolling update), the migration is a `stamped == head` no-op, and the
/// readiness gate passes again.
///
/// [`MigrationFailed`]: KubeRefreshOutcome::MigrationFailed
/// [`NoAnchor`]: KubeRefreshOutcome::NoAnchor
/// [`RollbackFailed`]: KubeRefreshOutcome::RollbackFailed
pub fn kube_refresh(
ops: &dyn KubeRefreshOps,
plan: &RefreshPlan<'_>,
) -> Result<KubeRefreshOutcome> {
let target = plan.target;
let head = plan
.chain
.last()
.ok_or_else(|| anyhow!("migration chain for `{}` is empty", target.product))?
.clone();
// 1. Anchor — read the rollback target BEFORE anything is patched.
let anchor = ops.current_image().ok().flatten();
// 2. Migrate: replay stamped→head as pre-upgrade Jobs, stamping each good hop.
let migrated = match ops.read_stamp()? {
// Brand-new deployment: stamp head, migrate nothing (a fresh install has
// nothing to forward-migrate). Mirrors replay_migrations' absent-stamp path.
None => {
ops.write_stamp(&head)?;
MigrationSummary {
from_stamp: None,
head: head.clone(),
replayed: Vec::new(),
was_noop: true,
}
}
Some(stamped) => {
let hops = migration_steps(&stamped, plan.chain)?;
let mut replayed = Vec::new();
let mut good = stamped.clone();
for (from, to) in &hops {
let job = migration_job(target, plan.new_image, from, to);
let outcome = ops.run_migration_job(&job)?;
if !outcome.ok() {
// A failed hop aborts here — nothing has been delivered yet, so
// the old Deployment is still live and healthy. The stamp is at
// the last good schema, so a retry resumes precisely.
let out = KubeRefreshOutcome::MigrationFailed {
hop: (from.clone(), to.clone()),
outcome,
stamped: good,
};
crate::surface_status!(
"skidbladnir/kube_runtime",
"kube_refresh",
false,
"{}",
out.summary()
);
return Ok(out);
}
replayed.push((from.clone(), to.clone()));
good = to.clone();
ops.write_stamp(&good)?;
}
MigrationSummary {
from_stamp: Some(stamped),
head: head.clone(),
replayed,
was_noop: hops.is_empty(),
}
}
};
// 3 + 4. Deliver + swap, then the readiness gate.
let desired = rolling_deployment(target, plan.new_image);
ops.apply_deployment(&desired)?;
let failure = match ops.wait_rollout(plan.rollout_wait) {
Ok(()) => {
let out = KubeRefreshOutcome::Rolled {
from: anchor,
to: plan.new_image.to_string(),
migrated,
};
crate::surface_status!(
"skidbladnir/kube_runtime",
"kube_refresh",
true,
"{}",
out.summary()
);
return Ok(out);
}
Err(e) => format!("{e:#}"),
};
// 5. Roll back to the prior revision by re-applying its pod spec.
let Some(previous) = anchor else {
let out = KubeRefreshOutcome::NoAnchor { failure };
crate::surface_status!(
"skidbladnir/kube_runtime",
"kube_refresh",
false,
"{}",
out.summary()
);
return Ok(out);
};
let prior = rolling_deployment(target, &previous);
if let Err(e) = ops.apply_deployment(&prior) {
let out = KubeRefreshOutcome::RollbackFailed {
failure,
rollback_failure: format!("{e:#}"),
};
crate::surface_status!(
"skidbladnir/kube_runtime",
"kube_refresh",
false,
"{}",
out.summary()
);
return Ok(out);
}
if let Err(e) = ops.wait_rollout(plan.rollback_wait) {
let out = KubeRefreshOutcome::RollbackFailed {
failure,
rollback_failure: format!("{e:#}"),
};
crate::surface_status!(
"skidbladnir/kube_runtime",
"kube_refresh",
false,
"{}",
out.summary()
);
return Ok(out);
}
let out = KubeRefreshOutcome::RolledBack {
to_previous: previous,
failure,
migrated,
};
crate::surface_status!(
"skidbladnir/kube_runtime",
"kube_refresh",
false,
"{}",
out.summary()
);
Ok(out)
}
// ───────────────────────── the live implementation ─────────────────────────
/// The real thing, over `kube-rs`. Feature `kubernetes`.
#[cfg(feature = "kubernetes")]
pub mod live {
use std::time::Duration;
use anyhow::{Context, Result, anyhow};
use k8s_openapi::api::apps::v1::Deployment;
use k8s_openapi::api::core::v1::ConfigMap;
use kube::Client;
use kube::api::{Api, Patch, PatchParams};
use serde_json::{Value, json};
use super::{
KubeRefreshOps, KubeRefreshOutcome, MigrationJobOutcome, RefreshPlan, RefreshTarget,
kube_refresh,
};
use crate::k8s_apply::{JobOutcome, K8sApplier, run_job};
/// A connected kube refresh runtime — the K8s peer of the systemd
/// [`SystemdRefreshControl`](crate::install::SystemdRefreshControl). Owns a
/// tokio runtime so its sync [`KubeRefreshOps`](crate::kube_runtime::KubeRefreshOps) methods can each `block_on` an
/// async `kube-rs` call — the same sync→async bridge `unfold`'s Talos target
/// uses. Zero-shell: every op is an apiserver call, never `kubectl`/`helm`.
pub struct KubeRuntime {
rt: tokio::runtime::Runtime,
client: Client,
applier: K8sApplier,
target: RefreshTarget,
/// Field manager server-side apply attributes changes to.
manager: String,
}
impl KubeRuntime {
/// Connect (kubeconfig on a laptop, in-cluster ServiceAccount as a Job) and
/// run API discovery.
pub fn connect(target: RefreshTarget) -> Result<Self> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.context("build tokio runtime for KubeRuntime")?;
let (client, applier) = rt.block_on(async {
let client = Client::try_default()
.await
.context("kube client (kubeconfig / in-cluster ServiceAccount)")?;
let applier = K8sApplier::from_client(client.clone()).await?;
Ok::<_, anyhow::Error>((client, applier))
})?;
Ok(KubeRuntime {
rt,
client,
applier,
target,
manager: "skidbladnir-refresh".to_string(),
})
}
/// **The verb.** Refresh the target to `new_image`, replaying the product's
/// migration `chain` first — the K8s peer of
/// [`Lifecycle::refresh`](crate::service::Lifecycle::refresh). Same migrate
/// model, same rollback-safety, one substrate over.
pub fn refresh(&self, new_image: &str, chain: &[String]) -> Result<KubeRefreshOutcome> {
let plan = RefreshPlan::new(&self.target, new_image, chain);
kube_refresh(self, &plan)
}
fn deployments(&self) -> Api<Deployment> {
Api::namespaced(self.client.clone(), &self.target.namespace)
}
}
impl KubeRefreshOps for KubeRuntime {
fn read_stamp(&self) -> Result<Option<String>> {
self.rt.block_on(async {
let api: Api<ConfigMap> =
Api::namespaced(self.client.clone(), &self.target.namespace);
match api.get_opt(&self.target.stamp_configmap()).await? {
Some(cm) => Ok(cm.data.and_then(|mut d| d.remove("schema_version"))),
None => Ok(None),
}
})
}
fn write_stamp(&self, schema: &str) -> Result<()> {
self.rt.block_on(async {
let api: Api<ConfigMap> =
Api::namespaced(self.client.clone(), &self.target.namespace);
let cm: Value = json!({
"apiVersion": "v1",
"kind": "ConfigMap",
"metadata": {
"name": self.target.stamp_configmap(),
"namespace": self.target.namespace,
"labels": { "app.kubernetes.io/name": self.target.workload },
},
"data": { "schema_version": schema, "product": self.target.product },
});
let obj: ConfigMap = serde_json::from_value(cm)?;
api.patch(
&self.target.stamp_configmap(),
&PatchParams::apply(&self.manager).force(),
&Patch::Apply(&obj),
)
.await
.with_context(|| format!("write migration stamp `{schema}`"))?;
Ok(())
})
}
fn current_image(&self) -> Result<Option<String>> {
self.rt.block_on(async {
let d = match self.deployments().get_opt(&self.target.workload).await? {
Some(d) => d,
None => return Ok(None),
};
Ok(d.spec
.and_then(|s| s.template.spec)
.and_then(|p| {
p.containers
.into_iter()
.find(|c| c.name == self.target.container)
})
.and_then(|c| c.image))
})
}
fn run_migration_job(&self, job: &Value) -> Result<MigrationJobOutcome> {
let outcome = self.rt.block_on(run_job(
&self.client,
&self.target.namespace,
job.clone(),
Duration::from_secs(600),
))?;
Ok(match outcome {
JobOutcome::Succeeded => MigrationJobOutcome::Succeeded,
JobOutcome::Failed => MigrationJobOutcome::Failed,
JobOutcome::TimedOut => MigrationJobOutcome::TimedOut,
})
}
fn apply_deployment(&self, deployment: &Value) -> Result<()> {
self.rt.block_on(self.applier.apply_json_manifests(
&self.target.namespace,
&self.manager,
std::slice::from_ref(deployment),
))
}
fn wait_rollout(&self, timeout: Duration) -> Result<()> {
self.rt.block_on(async {
let api = self.deployments();
let deadline = tokio::time::Instant::now() + timeout;
let mut last: String;
loop {
match api.get(&self.target.workload).await {
Ok(d) => {
let desired = d.spec.as_ref().and_then(|s| s.replicas).unwrap_or(1);
let st = d.status.unwrap_or_default();
let updated = st.updated_replicas.unwrap_or(0);
let ready = st.ready_replicas.unwrap_or(0);
let avail = st.available_replicas.unwrap_or(0);
if updated == desired && ready == desired && avail == desired {
return Ok(());
}
last = format!("desired={desired} updated={updated} ready={ready} available={avail}");
}
Err(e) => last = format!("{e:#}"),
}
if tokio::time::Instant::now() >= deadline {
return Err(anyhow!(
"rollout of {}/{} did not complete within {}s — last seen: {last}",
self.target.namespace,
self.target.workload,
timeout.as_secs()
));
}
tokio::time::sleep(Duration::from_secs(3)).await;
}
})
}
}
}
/// Compile-time proof that the production [`KubeRuntime`](live::KubeRuntime) — the
/// real kube backend — is wired into the unified
/// [`ServiceRefresh`](crate::service_backend::ServiceRefresh) seam, so a product
/// driving the seam can refresh a Kubernetes service exactly as it refreshes a
/// systemd one. `KubeRuntime` can only be built against a live apiserver, so this is
/// a type-level assertion (never called), not a runtime test — the runtime dispatch
/// is proven offline over the scripted mock in the tests below.
#[cfg(all(test, feature = "kubernetes"))]
const _: fn() = || {
fn assert_is_service_backend<T: crate::service_backend::ServiceRefresh>() {}
assert_is_service_backend::<live::KubeRuntime>();
};
// ───────────────────────────── tests ─────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
use std::cell::RefCell;
fn chain(v: &[&str]) -> Vec<String> {
v.iter().map(|s| s.to_string()).collect()
}
fn target() -> RefreshTarget {
RefreshTarget::new("njord", "korp-server", "korp-server", "korp")
.port(50051)
.health(RolloutProbe::Tcp(50051))
.data("korp-data", "/var/lib/korp")
}
// ── golden manifest renders (pure, no cluster) ──
#[test]
fn migration_job_renders_new_image_and_migrate_config_args() {
let job = migration_job(&target(), "registry.local/korp:v2", "korp-v0", "korp-v1");
assert_eq!(job["apiVersion"], "batch/v1");
assert_eq!(job["kind"], "Job");
// Job name is DNS-safe, per-hop, product-scoped.
assert_eq!(job["metadata"]["name"], "korp-migrate-korp-v0-korp-v1");
assert_eq!(job["metadata"]["namespace"], "njord");
// A migration must NOT silently retry a bad hop.
assert_eq!(job["spec"]["backoffLimit"], 0);
let pod = &job["spec"]["template"]["spec"];
assert_eq!(pod["restartPolicy"], "Never");
let c = &pod["containers"][0];
// Image = the NEW image (migrations are a property of the incoming code).
assert_eq!(c["image"], "registry.local/korp:v2");
// The product's own verb, exact argv, zero-shell.
assert_eq!(
c["args"],
json!(["--migrate-config", "from=korp-v0", "to=korp-v1"])
);
// The PVC is mounted so --migrate-config reaches the live data root.
assert_eq!(c["volumeMounts"][0]["mountPath"], "/var/lib/korp");
assert_eq!(
pod["volumes"][0]["persistentVolumeClaim"]["claimName"],
"korp-data"
);
}
#[test]
fn stateless_migration_job_has_no_data_volume() {
let t = RefreshTarget::new("ns", "web", "web", "web");
let job = migration_job(&t, "img:v2", "v0", "v1");
let pod = &job["spec"]["template"]["spec"];
assert!(pod["volumes"].is_null(), "no PVC ⇒ no volumes block");
assert!(
pod["containers"][0]["volumeMounts"].is_null(),
"no PVC ⇒ no volumeMounts"
);
}
#[test]
fn rolling_deployment_renders_new_image_strategy_and_probes() {
let dep = rolling_deployment(&target(), "registry.local/korp:v2");
assert_eq!(dep["kind"], "Deployment");
assert_eq!(dep["metadata"]["name"], "korp-server");
// The swap analog: an explicit RollingUpdate, zero-downtime.
assert_eq!(dep["spec"]["strategy"]["type"], "RollingUpdate");
assert_eq!(
dep["spec"]["strategy"]["rollingUpdate"]["maxUnavailable"],
0
);
assert_eq!(dep["spec"]["strategy"]["rollingUpdate"]["maxSurge"], 1);
let c = &dep["spec"]["template"]["spec"]["containers"][0];
assert_eq!(c["image"], "registry.local/korp:v2");
assert_eq!(c["ports"][0]["containerPort"], 50051);
// Health gate: both readiness + liveness on the declared endpoint.
assert_eq!(c["readinessProbe"]["tcpSocket"]["port"], 50051);
assert_eq!(c["livenessProbe"]["tcpSocket"]["port"], 50051);
// Selector matches the pod label.
assert_eq!(
dep["spec"]["selector"]["matchLabels"]["app.kubernetes.io/name"],
"korp-server"
);
}
#[test]
fn http_probe_renders_httpget_with_leading_slash() {
let t = RefreshTarget::new("ns", "api", "api", "api")
.port(8080)
.health(RolloutProbe::Http {
path: "healthz".into(),
port: 8080,
});
let dep = rolling_deployment(&t, "img:v2");
let c = &dep["spec"]["template"]["spec"]["containers"][0];
assert_eq!(c["readinessProbe"]["httpGet"]["path"], "/healthz");
assert_eq!(c["readinessProbe"]["httpGet"]["port"], 8080);
assert!(c["readinessProbe"]["tcpSocket"].is_null());
}
#[test]
fn no_probe_renders_no_probe_block() {
let t = RefreshTarget::new("ns", "web", "web", "web");
let dep = rolling_deployment(&t, "img:v2");
let c = &dep["spec"]["template"]["spec"]["containers"][0];
assert!(c["readinessProbe"].is_null());
assert!(c["livenessProbe"].is_null());
}
/// The rollback TARGET renders correctly: re-applying the prior image produces
/// the SAME Deployment shape with the prior image — the prior-revision re-apply.
#[test]
fn rollback_target_is_the_prior_image_deployment() {
let forward = rolling_deployment(&target(), "img:v2");
let rollback = rolling_deployment(&target(), "img:v1");
assert_eq!(
rollback["spec"]["template"]["spec"]["containers"][0]["image"],
"img:v1"
);
// Everything but the image is identical ⇒ k8s reuses the prior ReplicaSet.
assert_eq!(forward["metadata"], rollback["metadata"]);
assert_eq!(forward["spec"]["strategy"], rollback["spec"]["strategy"]);
assert_eq!(forward["spec"]["selector"], rollback["spec"]["selector"]);
}
// ── the engine, over a scripted mock (offline) ──
/// A scripted [`KubeRefreshOps`](crate::kube_runtime::KubeRefreshOps) — records every call and can be told to fail a
/// migration hop or the rollout. The K8s analog of the systemd `ScriptedControl`
/// + `RecordingRunner`.
struct ScriptedOps {
calls: RefCell<Vec<String>>,
stamp: RefCell<Option<String>>,
image: RefCell<Option<String>>,
/// The `to` schema whose migration Job fails.
fail_migration_to: Option<String>,
/// Images whose rollout never becomes Ready.
bad_images: Vec<String>,
/// Fail the rollback's wait too.
rollback_also_fails: bool,
}
impl ScriptedOps {
fn new(stamp: Option<&str>, image: Option<&str>) -> Self {
ScriptedOps {
calls: RefCell::new(Vec::new()),
stamp: RefCell::new(stamp.map(Into::into)),
image: RefCell::new(image.map(Into::into)),
fail_migration_to: None,
bad_images: Vec::new(),
rollback_also_fails: false,
}
}
fn log(&self, s: String) {
self.calls.borrow_mut().push(s);
}
fn calls(&self) -> Vec<String> {
self.calls.borrow().clone()
}
}
impl KubeRefreshOps for ScriptedOps {
fn read_stamp(&self) -> Result<Option<String>> {
Ok(self.stamp.borrow().clone())
}
fn write_stamp(&self, schema: &str) -> Result<()> {
self.log(format!("write_stamp {schema}"));
*self.stamp.borrow_mut() = Some(schema.to_string());
Ok(())
}
fn current_image(&self) -> Result<Option<String>> {
self.log("current_image".into());
Ok(self.image.borrow().clone())
}
fn run_migration_job(&self, job: &Value) -> Result<MigrationJobOutcome> {
let to = job["spec"]["template"]["spec"]["containers"][0]["args"][2]
.as_str()
.unwrap()
.to_string();
self.log(format!("migration_job {to}"));
if self.fail_migration_to.as_deref() == Some(to.trim_start_matches("to=")) {
return Ok(MigrationJobOutcome::Failed);
}
Ok(MigrationJobOutcome::Succeeded)
}
fn apply_deployment(&self, deployment: &Value) -> Result<()> {
let img = deployment["spec"]["template"]["spec"]["containers"][0]["image"]
.as_str()
.unwrap()
.to_string();
self.log(format!("apply_deployment {img}"));
*self.image.borrow_mut() = Some(img);
Ok(())
}
fn wait_rollout(&self, _timeout: Duration) -> Result<()> {
let cur = self.image.borrow().clone().unwrap_or_default();
let bad = self.bad_images.contains(&cur)
|| (self.rollback_also_fails && !self.bad_images.contains(&cur));
self.log(format!(
"wait_rollout {cur} -> {}",
if bad { "FAIL" } else { "ok" }
));
if bad {
anyhow::bail!("pods never became ready on {cur}");
}
Ok(())
}
}
/// A good refresh reads the anchor FIRST, migrates the chain in order, applies
/// the new image, and reports the hops.
#[test]
fn a_good_refresh_migrates_then_delivers_and_reports_hops() {
let ops = ScriptedOps::new(Some("korp-v0"), Some("img:v1"));
let ch = chain(&["korp-v0", "korp-v1", "korp-v2"]);
let t = target();
let plan = RefreshPlan::new(&t, "img:v2", &ch);
let out = kube_refresh(&ops, &plan).unwrap();
match &out {
KubeRefreshOutcome::Rolled { from, to, migrated } => {
assert_eq!(from.as_deref(), Some("img:v1"));
assert_eq!(to, "img:v2");
assert_eq!(
migrated.replayed,
vec![
("korp-v0".into(), "korp-v1".into()),
("korp-v1".into(), "korp-v2".into())
]
);
assert!(!migrated.was_noop);
}
other => panic!("expected Rolled, got {other:?}"),
}
assert!(out.is_success());
let calls = ops.calls();
// Anchor read before any mutation.
assert_eq!(calls[0], "current_image");
// Migrations run (and stamp) BEFORE the deployment is applied.
let apply_at = calls
.iter()
.position(|c| c.starts_with("apply_deployment"))
.unwrap();
let last_migrate = calls
.iter()
.rposition(|c| c.starts_with("migration_job"))
.unwrap();
assert!(
last_migrate < apply_at,
"all migration Jobs run before delivery: {calls:?}"
);
// The stamp advanced to head.
assert_eq!(ops.stamp.borrow().as_deref(), Some("korp-v2"));
}
/// `stamped == head` ⇒ zero migration Jobs, and it still delivers + rolls. The
/// idempotency guarantee, inherited from `migration_steps`.
#[test]
fn stamp_at_head_runs_no_migration_jobs() {
let ops = ScriptedOps::new(Some("korp-v2"), Some("img:v1"));
let ch = chain(&["korp-v0", "korp-v1", "korp-v2"]);
let t = target();
let plan = RefreshPlan::new(&t, "img:v2", &ch);
let out = kube_refresh(&ops, &plan).unwrap();
assert!(matches!(out, KubeRefreshOutcome::Rolled { .. }));
assert!(
!ops.calls().iter().any(|c| c.starts_with("migration_job")),
"no hops at head"
);
}
/// A brand-new deployment (no stamp) stamps head and migrates nothing.
#[test]
fn absent_stamp_stamps_head_and_migrates_nothing() {
let ops = ScriptedOps::new(None, None);
let ch = chain(&["korp-v0", "korp-v1"]);
let t = target();
let plan = RefreshPlan::new(&t, "img:v2", &ch);
let out = kube_refresh(&ops, &plan).unwrap();
assert!(matches!(out, KubeRefreshOutcome::Rolled { .. }));
assert!(!ops.calls().iter().any(|c| c.starts_with("migration_job")));
assert_eq!(
ops.stamp.borrow().as_deref(),
Some("korp-v1"),
"head stamped"
);
}
/// A failed migration Job ABORTS before any delivery — the old Deployment is
/// never touched — and leaves the stamp at the last good schema for a precise
/// retry. The K8s analog of a systemd migration that fails before the flip.
#[test]
fn a_failed_migration_aborts_before_delivery() {
let mut ops = ScriptedOps::new(Some("korp-v0"), Some("img:v1"));
ops.fail_migration_to = Some("korp-v2".into());
let ch = chain(&["korp-v0", "korp-v1", "korp-v2"]);
let t = target();
let plan = RefreshPlan::new(&t, "img:v2", &ch);
let out = kube_refresh(&ops, &plan).unwrap();
match &out {
KubeRefreshOutcome::MigrationFailed { hop, stamped, .. } => {
assert_eq!(hop, &("korp-v1".to_string(), "korp-v2".to_string()));
// v0→v1 succeeded and stamped; the failed hop did not advance it.
assert_eq!(stamped, "korp-v1");
}
other => panic!("expected MigrationFailed, got {other:?}"),
}
assert!(!out.is_success());
// The Deployment was NEVER applied — the running service is untouched.
assert!(
!ops.calls()
.iter()
.any(|c| c.starts_with("apply_deployment")),
"nothing may be delivered after a migration failure: {:?}",
ops.calls()
);
assert_eq!(
ops.image.borrow().as_deref(),
Some("img:v1"),
"old image still live"
);
}
/// **The rollback test.** An unready rollout reverts to the prior revision by
/// re-applying the prior image's spec, waits for it, and reports RolledBack —
/// the injected-seam analog of the systemd `unhealthy_refresh_rolls_back`.
#[test]
fn an_unready_rollout_rolls_back_to_the_prior_revision() {
let mut ops = ScriptedOps::new(Some("korp-v1"), Some("img:v1"));
ops.bad_images = vec!["img:v2".into()]; // the new image never becomes Ready
let ch = chain(&["korp-v0", "korp-v1"]);
let t = target();
let plan = RefreshPlan::new(&t, "img:v2", &ch);
let out = kube_refresh(&ops, &plan).unwrap();
match &out {
KubeRefreshOutcome::RolledBack {
to_previous,
failure,
..
} => {
assert_eq!(to_previous, "img:v1");
assert!(failure.contains("never became ready"));
}
other => panic!("expected RolledBack, got {other:?}"),
}
assert!(!out.is_success());
assert!(out.into_result().is_err());
// The prior image was re-applied (the prior revision brought back).
assert_eq!(ops.image.borrow().as_deref(), Some("img:v1"));
let calls = ops.calls();
// Order: apply new → wait (FAIL) → apply prior → wait (ok).
let applies: Vec<&String> = calls
.iter()
.filter(|c| c.starts_with("apply_deployment"))
.collect();
assert_eq!(applies[0], "apply_deployment img:v2");
assert_eq!(applies[1], "apply_deployment img:v1");
}
/// A rollback that also fails is its OWN outcome — the environment may be down,
/// never collapsed into "rollout failed".
#[test]
fn a_failed_rollback_is_its_own_outcome() {
let mut ops = ScriptedOps::new(Some("korp-v1"), Some("img:v1"));
ops.bad_images = vec!["img:v2".into()];
ops.rollback_also_fails = true; // even the prior image won't come back
let ch = chain(&["korp-v1"]);
let t = target();
let plan = RefreshPlan::new(&t, "img:v2", &ch);
let out = kube_refresh(&ops, &plan).unwrap();
assert!(
matches!(out, KubeRefreshOutcome::RollbackFailed { .. }),
"{out:?}"
);
assert!(out.summary().contains("may be DOWN"));
}
/// An unready rollout with no readable anchor says nothing was restored —
/// reporting it as a rollback would be a lie about what is running.
#[test]
fn no_anchor_says_nothing_restored() {
let mut ops = ScriptedOps::new(Some("korp-v1"), None); // no current image
ops.bad_images = vec!["img:v2".into()];
let ch = chain(&["korp-v1"]);
let t = target();
let plan = RefreshPlan::new(&t, "img:v2", &ch);
let out = kube_refresh(&ops, &plan).unwrap();
assert!(
matches!(out, KubeRefreshOutcome::NoAnchor { .. }),
"{out:?}"
);
assert!(out.summary().contains("nothing restored"));
}
/// A stamp outside the chain (downgrade / corruption) is a hard error — inherited
/// straight from `migration_steps`, no silent catch-up.
#[test]
fn a_stamp_outside_the_chain_is_refused() {
let ops = ScriptedOps::new(Some("mystery-v9"), Some("img:v1"));
let ch = chain(&["korp-v0", "korp-v1"]);
let t = target();
let plan = RefreshPlan::new(&t, "img:v2", &ch);
assert!(
kube_refresh(&ops, &plan).is_err(),
"an unknown stamp must be refused"
);
}
/// The dry run is pure and says what it would do (no cluster call).
#[test]
fn the_dry_run_is_pure() {
let ch = chain(&["korp-v0", "korp-v1"]);
let t = target();
let d = RefreshPlan::new(&t, "img:v2", &ch).describe();
assert!(d[0].contains("rollback anchor"));
assert!(d.iter().any(|l| l.contains("RollingUpdate")));
assert!(d.iter().any(|l| l.contains("on failure")));
}
// ── the unified ServiceBackend refresh seam dispatches to this engine ──
/// **The wiring proof.** Driving the unified
/// [`ServiceRefresh`](crate::service_backend::ServiceRefresh) seam's `refresh`
/// verb runs the SAME engine ([`kube_refresh`](crate::kube_runtime::kube_refresh)) the production
/// [`KubeRuntime::refresh`](live::KubeRuntime::refresh) wraps — the K8s peer of
/// the systemd backend's `refresh` delegating to `Lifecycle::refresh`. Proven at
/// the seam over the scripted mock (no cluster): the seam produces the same
/// outcome AND the same recorded engine calls as calling `kube_refresh` directly.
#[test]
fn service_backend_refresh_dispatches_to_the_kube_engine() {
use crate::service_backend::ServiceRefresh;
let ch = chain(&["korp-v0", "korp-v1", "korp-v2"]);
let t = target();
let plan = RefreshPlan::new(&t, "img:v2", &ch);
// Same scripted world, driven two ways.
let direct_ops = ScriptedOps::new(Some("korp-v0"), Some("img:v1"));
let direct = kube_refresh(&direct_ops, &plan).unwrap();
let seam_ops = ScriptedOps::new(Some("korp-v0"), Some("img:v1"));
// The unified verb — NOT kube_refresh — reaching the engine through the trait.
let seam = ServiceRefresh::refresh(&seam_ops, &plan).unwrap();
assert_eq!(
direct, seam,
"the seam must produce the engine's exact outcome"
);
assert!(matches!(seam, KubeRefreshOutcome::Rolled { .. }));
// Anchor-first, migrate-before-deliver — the engine's contract, reached via the seam.
let calls = seam_ops.calls();
assert_eq!(calls[0], "current_image");
let apply_at = calls
.iter()
.position(|c| c.starts_with("apply_deployment"))
.unwrap();
let last_migrate = calls
.iter()
.rposition(|c| c.starts_with("migration_job"))
.unwrap();
assert!(
last_migrate < apply_at,
"the seam ran the real engine: {calls:?}"
);
assert_eq!(
seam_ops.calls(),
direct_ops.calls(),
"identical engine call trace"
);
}
/// The seam carries the engine's rollback safety through unchanged — an unready
/// rollout reverts to the prior revision whether reached directly or via the
/// unified `refresh` verb.
#[test]
fn service_backend_refresh_preserves_rollback() {
use crate::service_backend::ServiceRefresh;
let mut ops = ScriptedOps::new(Some("korp-v1"), Some("img:v1"));
ops.bad_images = vec!["img:v2".into()];
let ch = chain(&["korp-v0", "korp-v1"]);
let t = target();
let plan = RefreshPlan::new(&t, "img:v2", &ch);
let out = ServiceRefresh::refresh(&ops, &plan).unwrap();
assert!(
matches!(out, KubeRefreshOutcome::RolledBack { ref to_previous, .. } if to_previous == "img:v1"),
"{out:?}"
);
assert!(!out.is_success());
}
}