refactor(discovery)!: split worker reconciliation and mesh-router discovery

.github/workflows/e2e-kind-discovery.yml

@@ -6,7 +6,12 @@ name: E2E Kind Discovery
 on:
   pull_request:
     paths:
-      - model_gateway/src/service_discovery.rs
+      - model_gateway/src/service_discovery/**
+      - model_gateway/src/mesh_discovery/**
+      # The start/skip gating for both discoveries now lives in `startup`,
+      # and `main` owns the CLI -> discovery-config conversion feeding it.
+      - model_gateway/src/server.rs
+      - model_gateway/src/main.rs
       - model_gateway/tests/k8s_discovery_test.rs
       - e2e_test/kind_discovery/**
       - .github/workflows/e2e-kind-discovery.yml

bindings/python/src/lib.rs

@@ -1488,9 +1488,20 @@ impl Router {
                 worker_ports_annotation: self.worker_ports_annotation.clone(),
                 kv_connector_annotation: self.kv_connector_annotation.clone(),
                 kv_engine_id_annotation: self.kv_engine_id_annotation.clone(),
+                model_id_source,
+            })
+        } else {
+            None
+        };
+
+        // Mesh-router discovery now has its own task and lifetime, but stays
+        // gated on the legacy service-discovery flag until it gains its own
+        // config surface with the tagged provider configuration.
+        let mesh_discovery_config = if self.service_discovery && !self.router_selector.is_empty() {
+            Some(mesh_discovery::MeshDiscoveryConfig {
+                namespace: self.service_discovery_namespace.clone(),
                 router_selector: self.router_selector.clone(),
                 router_mesh_port_annotation: "sglang.ai/mesh-port".to_string(),
-                model_id_source,
             })
         } else {
             None
@@ -1522,6 +1533,7 @@ impl Router {
                 log_level: self.log_level.clone(),
                 log_json: self.log_json,
                 service_discovery_config,
+                mesh_discovery_config,
                 prometheus_config,
                 request_timeout_secs: self.request_timeout_secs,
                 request_id_headers: self.request_id_headers.clone(),
@@ -1549,6 +1561,16 @@ impl Router {
                             })
                         })
                         .transpose()?;
+                    // Mirrors the CLI check in `main.rs`: port 0 parses as a
+                    // socket address but is undialable once gossiped to peers,
+                    // and reaches router discovery as the fallback mesh port
+                    // for Pods with no usable annotation.
+                    if self.mesh_port == 0 {
+                        return Err(pyo3::exceptions::PyValueError::new_err(
+                            "Invalid value for mesh_port='0': mesh port cannot be 0; peers dial \
+                             the advertised port, so it must be a fixed, routable one",
+                        ));
+                    }
                     let bind_addr =
                         Self::parse_mesh_socket_addr(&self.mesh_host, self.mesh_port, "mesh_host")?;
                     let (advertise_host, advertise_field) =

crates/rl/COUPLING.md

@@ -16,7 +16,7 @@ Test-only files that the mount also touches, none of them a new surface:
 `model_gateway/tests/rl_control_plane_test.rs` (gateway-level `/v1/rl` tests),
 `model_gateway/tests/common/mock_worker.rs` (engine-native RL routes on the
 mock), and the three `#[cfg(test)]` `AppContext { .. }` literals in
-`src/service_discovery.rs`, `src/workflow/steps/local/drain_workers.rs`, and
+`src/service_discovery/testing.rs`, `src/workflow/steps/local/drain_workers.rs`, and
 `src/workflow/steps/local/update_worker_properties.rs`, which gain `rl: None`
 because the struct grew a field. The gateway-level test relies on
 `TestRouterConfig` disabling health checks, so the mock stopped mid-test

e2e_test/kind_discovery/test_in_cluster_journey.py

@@ -80,6 +80,16 @@ def engine_pod_ips(label: str = "app=engines-incluster") -> set[str]:
     }
 
 
+def _wait_until_gone(label: str, timeout: float = 120.0) -> None:
+    """Block until no Pod matching `label` is listed as live."""
+    start = time.monotonic()
+    while time.monotonic() - start < timeout:
+        if not engine_pod_ips(label):
+            return
+        time.sleep(1)
+    raise AssertionError(f"timed out waiting for pods matching {label!r} to go away")
+
+
 def expected_urls(pod_ips: set[str]) -> set[str]:
     return {f"http://{ip}:{port}" for ip in pod_ips for port in ENGINE_PORTS}
 
@@ -236,6 +246,13 @@ def test_misconfigured_deployment_falls_back_and_is_diagnosable(self, incluster)
         )
 
         kubectl("delete", "deployment", "engines-typo", "--wait=true")
+        # `--wait` covers the Deployment object only: its Pod is garbage
+        # collected asynchronously and can still be listed, without a
+        # deletionTimestamp, right after the call returns. Snapshotting the
+        # fleet here would fold the doomed typo Pod into `expected` at four
+        # ports -- a set the gateway can never produce, since that Pod only
+        # ever contributed one worker.
+        _wait_until_gone("app=engines-incluster,variant=typo")
         self._wait_for_urls(
             incluster,
             expected_urls(engine_pod_ips()),

model_gateway/src/lib.rs

@@ -3,6 +3,7 @@ pub mod config;
 pub mod endpoints;
 pub mod health;
 pub mod mesh;
+pub mod mesh_discovery;
 pub mod middleware;
 pub mod observability;
 pub mod policies;

model_gateway/src/main.rs

@@ -19,6 +19,7 @@ use smg::{
         RoutingKeyOverrideConfig, RoutingMode, SchemaConfig, TenantApiKeyEntry,
         TokenizerCacheConfig, TraceConfig,
     },
+    mesh_discovery::MeshDiscoveryConfig,
     observability::{
         metrics::{register_jemalloc_as_global_allocator, PrometheusConfig},
         otel_trace::{is_otel_enabled, shutdown_otel},
@@ -1402,6 +1403,21 @@ impl CliArgs {
             })
             .transpose()?;
 
+        // Port 0 parses fine as a socket address, but it is meaningless as an
+        // advertised one: peers gossip this port and cannot dial an ephemeral
+        // bind. It would also reach router discovery as the fallback mesh port
+        // for Pods whose annotation is missing or invalid, publishing `IP:0`
+        // peers into the cluster state.
+        if self.mesh_port == 0 {
+            return Err(ConfigError::InvalidValue {
+                field: "mesh_port".to_string(),
+                value: self.mesh_port.to_string(),
+                reason: "mesh port cannot be 0; peers dial the advertised port, \
+                         so it must be a fixed, routable one"
+                    .to_string(),
+            });
+        }
+
         let bind_addr = Self::parse_mesh_socket_addr(&self.mesh_host, self.mesh_port, "mesh_host")?;
         let (advertise_host, advertise_field) =
             if let Some(host) = self.mesh_advertise_host.as_deref() {
@@ -1967,27 +1983,17 @@ impl CliArgs {
 
     fn to_server_config(&self, router_config: RouterConfig) -> ConfigResult<ServerConfig> {
         let service_discovery_config = if self.service_discovery {
-            // Get router discovery config from router_config.discovery if available
-            let (
-                router_selector,
-                router_mesh_port_annotation,
-                kv_connector_annotation,
-                kv_engine_id_annotation,
-            ) = router_config
+            let (kv_connector_annotation, kv_engine_id_annotation) = router_config
                 .discovery
                 .as_ref()
                 .map(|d| {
                     (
-                        d.router_selector.clone(),
-                        d.router_mesh_port_annotation.clone(),
                         d.kv_connector_annotation.clone(),
                         d.kv_engine_id_annotation.clone(),
                     )
                 })
                 .unwrap_or_else(|| {
                     (
-                        HashMap::new(),
-                        "sglang.ai/mesh-port".to_string(),
                         self.kv_connector_annotation.clone(),
                         self.kv_engine_id_annotation.clone(),
                     )
@@ -2025,14 +2031,26 @@ impl CliArgs {
                 worker_ports_annotation: "smg.ai/worker-ports".to_string(),
                 kv_connector_annotation,
                 kv_engine_id_annotation,
-                router_selector,
-                router_mesh_port_annotation,
                 model_id_source,
             })
         } else {
             None
         };
 
+        // Mesh-router discovery now has its own task and lifetime, but its
+        // configuration still arrives inside `discovery`, so it stays reachable
+        // only under `--service-discovery`. Moving it onto its own config
+        // surface lands with the tagged provider configuration.
+        let mesh_discovery_config = router_config
+            .discovery
+            .as_ref()
+            .map(|d| MeshDiscoveryConfig {
+                namespace: d.namespace.clone(),
+                router_selector: d.router_selector.clone(),
+                router_mesh_port_annotation: d.router_mesh_port_annotation.clone(),
+            })
+            .filter(MeshDiscoveryConfig::is_enabled);
+
         let prometheus_config = Some(PrometheusConfig {
             port: self.prometheus_port,
             host: self.prometheus_host.clone(),
@@ -2067,6 +2085,7 @@ impl CliArgs {
             log_level: Some(self.log_level.clone()),
             log_json: self.log_json,
             service_discovery_config,
+            mesh_discovery_config,
             prometheus_config,
             request_timeout_secs: self.request_timeout_secs,
             request_id_headers: if self.request_id_headers.is_empty() {
@@ -2447,6 +2466,89 @@ mod tests {
         assert_eq!(format!("{canonical:?}"), format!("{aliased:?}"));
     }
 
+    /// Mesh-router discovery has its own task and lifetime, but is still
+    /// configured through `discovery`, which only `--service-discovery`
+    /// populates. `--router-selector` alone therefore still does nothing; it
+    /// gains its own config surface with the tagged provider configuration.
+    #[test]
+    fn router_selector_alone_does_not_yet_configure_mesh_discovery() {
+        let cli = cli_args_from(&["--router-selector", "role=router"]);
+        let router = cli.to_router_config(vec![], vec![]).unwrap();
+        assert!(router.discovery.is_none());
+
+        let server = cli.to_server_config(router).unwrap();
+        assert!(server.service_discovery_config.is_none());
+        assert!(server.mesh_discovery_config.is_none());
+    }
+
+    #[test]
+    fn mesh_discovery_is_absent_without_a_router_selector() {
+        let cli = cli_args_from(&["--service-discovery", "--selector", "app=worker"]);
+        let router = cli.to_router_config(vec![], vec![]).unwrap();
+        let server = cli.to_server_config(router).unwrap();
+        assert!(server.service_discovery_config.is_some());
+        assert!(server.mesh_discovery_config.is_none());
+    }
+
+    /// The router selector reaches the new independent mesh task unchanged.
+    #[test]
+    fn service_discovery_router_selector_reaches_mesh_discovery() {
+        let cli = cli_args_from(&[
+            "--service-discovery",
+            "--selector",
+            "app=worker",
+            "--router-selector",
+            "role=router",
+        ]);
+        let router = cli.to_router_config(vec![], vec![]).unwrap();
+        let server = cli.to_server_config(router).unwrap();
+        let mesh = server.mesh_discovery_config.expect("mesh discovery config");
+        assert_eq!(
+            mesh.router_selector.get("role").map(String::as_str),
+            Some("router")
+        );
+    }
+
+    /// Port 0 parses as a socket address, so nothing downstream rejects it:
+    /// it would be advertised to peers and used as the fallback mesh port for
+    /// router Pods with no usable annotation, publishing `IP:0` peers.
+    #[test]
+    fn mesh_port_zero_is_rejected() {
+        let cli = cli_args_from(&[
+            "--enable-mesh",
+            "--mesh-advertise-host",
+            "10.0.0.1",
+            "--mesh-port",
+            "0",
+        ]);
+        let router = cli.to_router_config(vec![], vec![]).unwrap();
+        // `ServerConfig` is not `Debug`, so unwrap the error by hand.
+        let Err(err) = cli.to_server_config(router) else {
+            panic!("mesh port 0 must be rejected");
+        };
+        assert!(
+            format!("{err}").contains("mesh port cannot be 0"),
+            "unexpected error: {err}"
+        );
+    }
+
+    /// The same configuration with a real port still builds, so the guard is
+    /// not rejecting every meshed setup.
+    #[test]
+    fn mesh_port_nonzero_is_accepted() {
+        let cli = cli_args_from(&[
+            "--enable-mesh",
+            "--mesh-advertise-host",
+            "10.0.0.1",
+            "--mesh-port",
+            "39527",
+        ]);
+        let router = cli.to_router_config(vec![], vec![]).unwrap();
+        let server = cli.to_server_config(router).unwrap();
+        let mesh = server.mesh_server_config.expect("mesh server config");
+        assert_eq!(mesh.advertise_addr.port(), 39527);
+    }
+
     #[test]
     fn kv_annotation_flags_flow_into_both_configs() {
         let cli = cli_args_from(&[