fix(epp): tolerate transient EndpointSlice port absence during startup

deploy/inference-gateway/ext-proc/src/peer_discovery.rs

@@ -28,26 +28,53 @@ type Store = kube::runtime::reflector::Store<EndpointSlice>;
 /// Resolve the required aggregated replica-sync port from the peer Service's
 /// EndpointSlices. Every slice must expose the same named `replica-agg` port;
 /// missing or inconsistent ports fail EPP startup before replica sync is built.
+/// How many times to retry the EndpointSlice LIST when the observed slice is
+/// transiently incomplete (e.g. every pod restarting at once). The window is
+/// short (hundreds of ms), so a few bounded retries smooth it out without
+/// masking a genuine misconfiguration, which still fails after retries
+/// exhaust.
+const PORT_RESOLUTION_RETRIES: usize = 5;
+const PORT_RESOLUTION_INITIAL_BACKOFF_MS: u64 = 100;
+
 pub async fn resolve_replica_sync_port(namespace: &str, service_name: &str) -> Result<u16> {
     use kube::{Api, Client, api::ListParams};
 
     let client = Client::try_default()
         .await
         .context("building Kubernetes client for EPP peer port resolution")?;
     let slices: Api<EndpointSlice> = Api::namespaced(client, namespace);
-    let list = slices
-        .list(&ListParams::default().labels(&format!("{SERVICE_NAME_LABEL}={service_name}")))
-        .await
-        .with_context(|| {
-            format!("listing EndpointSlices for EPP peer Service {namespace}/{service_name}")
-        })?;
 
-    replica_sync_port(list.items.iter()).with_context(|| {
-        format!(
-            "resolving named port {REPLICA_AGG_PORT_NAME:?} for EPP peer Service \
-             {namespace}/{service_name}"
-        )
-    })
+    let mut backoff = std::time::Duration::from_millis(PORT_RESOLUTION_INITIAL_BACKOFF_MS);
+    for attempt in 0..PORT_RESOLUTION_RETRIES {
+        let list = slices
+            .list(&ListParams::default().labels(&format!("{SERVICE_NAME_LABEL}={service_name}")))
+            .await
+            .with_context(|| {
+                format!("listing EndpointSlices for EPP peer Service {namespace}/{service_name}")
+            })?;
+        match replica_sync_port(list.items.iter()) {
+            Ok(port) => return Ok(port),
+            Err(error) if attempt + 1 < PORT_RESOLUTION_RETRIES => {
+                tracing::warn!(
+                    %error,
+                    attempt,
+                    backoff_ms = backoff.as_millis(),
+                    "EPP peer port resolution saw transient EndpointSlice state; retrying"
+                );
+                tokio::time::sleep(backoff).await;
+                backoff = backoff.saturating_mul(2);
+            }
+            Err(error) => {
+                return Err(error).with_context(|| {
+                    format!(
+                        "resolving named port {REPLICA_AGG_PORT_NAME:?} for EPP peer Service \
+                         {namespace}/{service_name}"
+                    )
+                });
+            }
+        }
+    }
+    unreachable!("retry loop always returns or exhausts")
 }
 
 fn replica_sync_port<'a>(slices: impl Iterator<Item = &'a EndpointSlice>) -> Result<u16> {
@@ -62,27 +89,37 @@ fn replica_sync_port<'a>(slices: impl Iterator<Item = &'a EndpointSlice>) -> Res
             .as_deref()
             .unwrap_or_default()
             .iter()
-            // Only a TCP `replica-agg` port satisfies the contract: the replica
-            // plane binds and dials `tcp://`. Kubernetes defaults `protocol` to
-            // TCP when absent, so treat `None` as TCP and reject explicit
-            // UDP/SCTP rather than let a mismatched port through.
-            .filter(|port| {
-                port.name.as_deref() == Some(REPLICA_AGG_PORT_NAME)
-                    && port
-                        .protocol
-                        .as_deref()
-                        .is_none_or(|protocol| protocol.eq_ignore_ascii_case("TCP"))
-            });
-        let endpoint_port = matches.next().with_context(|| {
-            format!(
-                "EndpointSlice {slice_name} does not expose named port \
-                 {REPLICA_AGG_PORT_NAME:?}"
-            )
-        })?;
+            .filter(|port| port.name.as_deref() == Some(REPLICA_AGG_PORT_NAME));
+        // Skip a slice that does not expose the named port at all. The
+        // EndpointSlice controller rewrites slices while pods churn, so a LIST
+        // can observe a slice mid-update without its ports; the Service is the
+        // single source of truth for ports, so this is a transient race, not a
+        // misconfiguration. Erroring on it made replica startup crash (and
+        // restart) whenever every pod is restarted at once. Genuine
+        // misconfiguration is still caught below: no slice exposing the port,
+        // or conflicting values across slices.
+        let Some(endpoint_port) = matches.next() else {
+            tracing::debug!(
+                slice_name,
+                "EndpointSlice does not expose replica-agg port; skipping transient slice"
+            );
+            continue;
+        };
         anyhow::ensure!(
             matches.next().is_none(),
             "EndpointSlice {slice_name} exposes named port {REPLICA_AGG_PORT_NAME:?} more than once"
         );
+        // Only a TCP `replica-agg` port satisfies the contract: the replica
+        // plane binds and dials `tcp://`. Kubernetes defaults `protocol` to
+        // TCP when absent, so treat `None` as TCP and reject explicit
+        // UDP/SCTP rather than let a mismatched port through.
+        anyhow::ensure!(
+            endpoint_port
+                .protocol
+                .as_deref()
+                .is_none_or(|protocol| protocol.eq_ignore_ascii_case("TCP")),
+            "EndpointSlice {slice_name} named port {REPLICA_AGG_PORT_NAME:?} must use TCP"
+        );
         let raw_port = endpoint_port.port.with_context(|| {
             format!(
                 "EndpointSlice {slice_name} named port {REPLICA_AGG_PORT_NAME:?} has no port number"
@@ -101,6 +138,10 @@ fn replica_sync_port<'a>(slices: impl Iterator<Item = &'a EndpointSlice>) -> Res
     }
 
     anyhow::ensure!(slice_count > 0, "peer Service has no EndpointSlices");
+    anyhow::ensure!(
+        !resolved.is_empty(),
+        "no EndpointSlice exposes named port {REPLICA_AGG_PORT_NAME:?}"
+    );
     anyhow::ensure!(
         resolved.len() == 1,
         "named port {REPLICA_AGG_PORT_NAME:?} resolves to inconsistent ports {resolved:?}"
@@ -434,6 +475,18 @@ mod tests {
         assert!(error.contains(REPLICA_AGG_PORT_NAME));
     }
 
+    #[test]
+    fn skips_slice_missing_named_port_alongside_one_that_exposes_it() {
+        // A slice being rewritten during pod churn can momentarily lack the
+        // named port; the Service is the source of truth for ports, so
+        // resolution must succeed as long as another slice exposes it.
+        let slices = [
+            slice_with(&["10.0.0.1"], false, "IPv4"),
+            slice_with_replica_port(Some(9092)),
+        ];
+        assert_eq!(replica_sync_port(slices.iter()).unwrap(), 9092);
+    }
+
     #[test]
     fn rejects_inconsistent_replica_agg_named_ports() {
         let slices = [