Replace engine copies that aren't serving when a rollout stalls

functions/compose-model-replica/function/backends/base.py

@@ -55,6 +55,8 @@ def build(
         provider_config: str,
         serving_label: str,
         stack: str,
+        *,
+        observed: Mapping[str, fnv1.Resource],
     ) -> dict[str, k8sobjv1alpha1.Object]: ...
 
     def ready_copies(self, engine: v1alpha1.Engine, observed: Mapping[str, fnv1.Resource]) -> int:
@@ -589,6 +591,56 @@ def ready_replicas(engine: v1alpha1.Engine, observed: Mapping[str, fnv1.Resource
     return int(status.get("readyReplicas") or 0)
 
 
+def stalled_copies_down(engine: v1alpha1.Engine, observed: Mapping[str, fnv1.Resource], key: str) -> int:
+    """How many of an engine's copies aren't available once its Deployment's
+    rollout has stalled, or 0 while it hasn't. key is the response key the
+    Deployment is composed under.
+
+    A rollout has stalled once the Deployment controller marks it Progressing
+    False with reason ProgressDeadlineExceeded. It checks the deadline when it
+    next syncs the Deployment after progressDeadlineSeconds (600 by default)
+    with no progress, so this can take longer than the deadline. It marks it
+    Progressing True again as soon as the rollout makes progress, such as a pod
+    becoming available or an old pod being removed.
+    """
+    status = observed_status(observed, key) or {}
+    stalled = any(
+        c.get("type") == "Progressing" and c.get("status") == "False" and c.get("reason") == "ProgressDeadlineExceeded"
+        for c in status.get("conditions") or []
+    )
+    if not stalled:
+        return 0
+    copies = int(engine.copies or 1)
+    return copies - min(int(status.get("availableReplicas") or 0), copies)
+
+
+# Kubernetes' default for both of a rolling update's bounds. Surge rounds up and
+# unavailable rounds down, so an engine of fewer than four copies starts a new
+# pod before it stops an old one.
+_ROLLING_UPDATE_DEFAULT = "25%"
+
+
+def rolling_update(copies: int, down: int) -> dict:
+    """A Deployment strategy that surges, and replaces the copies down in a
+    stalled rollout without waiting for their replacements.
+
+    Waiting would deadlock when the only devices a new pod fits on are held by
+    an old pod that isn't serving, such as one crash-looping on a bad config.
+    Counting copies down only once the rollout has stalled means a copy that's
+    briefly unready while the rollout still progresses isn't removed. A copy
+    that's available stays up until its replacement is available.
+
+    The bounds stay at Kubernetes' defaults until more copies are down than
+    they allow, so a healthy engine's Deployment is composed as it would be
+    defaulted.
+    """
+    max_unavailable: int | str = down if down > copies // 4 else _ROLLING_UPDATE_DEFAULT
+    return {
+        "type": "RollingUpdate",
+        "rollingUpdate": {"maxSurge": _ROLLING_UPDATE_DEFAULT, "maxUnavailable": max_unavailable},
+    }
+
+
 def serving_label(replica: v1alpha1.ModelReplica) -> str:
     """The serving label value a replica's serving pods share.
 

functions/compose-model-replica/function/backends/grove.py

@@ -61,6 +61,11 @@ def build(
         provider_config: str,
         serving_label: str,
         stack: str,
+        *,
+        # Unused here: the workload's own default rollout replaces a gang
+        # before it starts the new one, so it never waits on devices an old
+        # gang holds.
+        observed: Mapping[str, fnv1.Resource],  # noqa: ARG002
     ) -> dict[str, k8sobjv1alpha1.Object]:
         leader = base.engine_member(engine, base.ROLE_LEADER)
         worker = base.engine_member(engine, base.ROLE_WORKER)

functions/compose-model-replica/function/backends/llmd.py

@@ -60,6 +60,11 @@ def build(
         # backend. Unused here: this backend is Standard-only, so there's no
         # stack to switch on.
         stack: str,  # noqa: ARG002
+        *,
+        # Unused here: the workload's own default rollout replaces a gang
+        # before it starts the new one, so it never waits on devices an old
+        # gang holds.
+        observed: Mapping[str, fnv1.Resource],  # noqa: ARG002
     ) -> dict[str, k8sobjv1alpha1.Object]:
         leader = base.engine_member(engine, base.ROLE_LEADER)
         worker = base.engine_member(engine, base.ROLE_WORKER)

functions/compose-model-replica/function/backends/native.py

@@ -42,6 +42,8 @@ def build(
         provider_config: str,
         serving_label: str,
         stack: str,
+        *,
+        observed: Mapping[str, fnv1.Resource],
     ) -> dict[str, k8sobjv1alpha1.Object]:
         member = base.engine_member(engine, base.ROLE_STANDALONE)
         # select_backend dispatches the native backend only for an engine with a
@@ -103,13 +105,16 @@ def build(
         if tmpl.spec.imagePullSecrets:
             pod_spec["imagePullSecrets"] = [s.model_dump(exclude_none=True) for s in tmpl.spec.imagePullSecrets]
 
+        copies = int(engine.copies or 1)
+        key = base.workload_key(engine)
         deployment = {
             "apiVersion": "apps/v1",
             "kind": "Deployment",
             "metadata": {"name": name, "namespace": base.remote_namespace(replica)},
             "spec": {
-                "replicas": int(engine.copies or 1),
+                "replicas": copies,
                 "selector": {"matchLabels": selector},
+                "strategy": base.rolling_update(copies, base.stalled_copies_down(engine, observed, key)),
                 "template": {
                     "metadata": base.pod_metadata(member, pod_labels, replica=replica, engine=engine),
                     "spec": pod_spec,
@@ -118,9 +123,7 @@ def build(
         }
 
         composed = {
-            base.workload_key(engine): base.wrap_object(
-                provider_config, deployment, cel_query=base.ALL_COPIES_READY_CEL, watch=True
-            ),
+            key: base.wrap_object(provider_config, deployment, cel_query=base.ALL_COPIES_READY_CEL, watch=True),
         }
         # Gated like the container's resources above, and for the same reason.
         if member.deviceRequests:

functions/compose-model-replica/function/fn.py

@@ -192,7 +192,7 @@ def compose_model_serving(self) -> None:
         composed: dict[str, k8sobjv1alpha1.Object] = {}
         for engine in self.xr.spec.engines:
             backend = _BACKENDS[base.select_backend(engine, stack)]()
-            composed.update(backend.build(self.xr, engine, pc, label, stack))
+            composed.update(backend.build(self.xr, engine, pc, label, stack, observed=self.req.observed.resources))
         composed = routing.apply(composed, self.xr, pc)
         for key, obj in composed.items():
             resource.update(self.rsp.desired.resources[key], obj)