feat(epp): recover embedded KV index from peers before readiness

deploy/inference-gateway/ext-proc/examples/onramp/README.md

@@ -23,7 +23,8 @@ KV-aware selection is provided by the runtime-free
 [selection service](../../../../../docs/fern/pages/developer-guide/knowledge-base/modular-components/router/standalone-selection.md),
 which the EPP runs **in-process**: the EPP and the selection service are compiled into one binary,
 so there is no separate selector Deployment and no HTTP hop. The EPP can run single-replica, or
-**replicated** with cross-replica active-load sync between EPP pods (see
+**replicated** with cross-replica active-load sync and startup KV-index recovery between EPP pods
+(see
 [Replicated mode](../../../../../docs/fern/pages/kubernetes/kv-aware-routing/vanilla-vllm-onramp.mdx#epp-replication)).
 
 Whether the EPP uses the Dynamo runtime or not is controlled with the `DYN_EPP_MODE` environment
@@ -66,8 +67,9 @@ flowchart LR
 - Operator-managed lifecycle for Workers, Services, `InferencePool`, and EPP resources.
 - Request migration, rejection, cancellation - overall admission control
 - Data parallelism (The standalone mode which targets DP=1.)
-- Cross-replica KV-index warm-up when new replica re-warms from live traffic + replay.
-- Initial worker cache-state synchronization instead of rebuilding the index only from live traffic.
+- Active-reservation or load-state snapshots. Replica lifecycle events converge only after a replica
+  joins.
+- An atomic KV-index snapshot plus live-event handoff or event-cursor protocol.
 - Per-tenant KV cache isolation with x-tenant-id / cache_salt. This requires per-engine support and as such is not supported in the Standalone mode.
 - Management of Transient disconnects. In the Dynamo mode the KV-cache updates the worker sent during the gap are recovered from the worker's **replay** socket when `DYN_EPP_KV_EVENT_REPLAY_PORT` is set (and the vLLM worker exposes one); otherwise the index refreshes from new traffic.
 - Dropped events / gaps management. The `SelectionCore` indexer does seq-watermark gap detection and replays missed events from the worker's replay socket when `DYN_EPP_KV_EVENT_REPLAY_PORT` is configured. Without a replay socket, gaps are dropped and the index re-warms from new traffic.

deploy/inference-gateway/ext-proc/examples/onramp/agg.yaml

@@ -258,12 +258,13 @@ spec:
             # the vLLM engine's --max-num-batched-tokens.
             - name: DYN_EPP_MAX_NUM_BATCHED_TOKENS
               value: "8192"
-            # Replication: watch THIS Deployment's own Service
-            # (EndpointSlices) to find sibling EPP replicas and sync active load
-            # over its required named replica-agg port. Omit this env to run a
-            # single fully-local replica (and set replicas: 1).
+            # Replication: watch the peer Service (EndpointSlices) to find
+            # sibling EPP replicas and sync active load over its required named
+            # replica-agg port, and to recover the KV index over selection-http.
+            # Omit this env to run a single fully-local replica (and set
+            # replicas: 1).
             - name: DYN_EPP_PEER_SERVICE
-              value: dynamo-epp
+              value: dynamo-epp-peer
             # The reflector watches pods in the EPP's own namespace.
             - name: POD_NAMESPACE
               valueFrom:
@@ -282,6 +283,9 @@ spec:
             # ZMQ replica-sync between EPP replicas (embedded replication).
             - name: replica-agg
               containerPort: 9092
+            # Internal HTTP dump endpoint for joining EPP KV-index recovery.
+            - name: selection-http
+              containerPort: 9093
           readinessProbe:
             grpc:
               port: 9003
@@ -295,7 +299,11 @@ spec:
               drop:
                 - ALL
 ---
-# GAIE calls the gRPC port. EPP replicas use `replica-agg` for load-state sync.
+# GAIE calls the gRPC port on the request Service; EPP replicas talk to each
+# other over the dedicated peer Service below. Keeping the peer plane on its
+# own Service makes the EndpointSlice discovery (and its port contract)
+# explicit: the EPP reads `replica-agg` / `selection-http` from
+# `dynamo-epp-peer`, and the request Service carries only gRPC.
 apiVersion: v1
 kind: Service
 metadata:
@@ -308,9 +316,61 @@ spec:
       appProtocol: http2
       port: 9002
       targetPort: grpc
+---
+# Peer Service: `replica-agg` for lifecycle replica sync, `selection-http` for
+# startup KV-index recovery. The EPP watches THIS Service's EndpointSlices for
+# sibling replicas and resolves the named peer ports from its spec.
+apiVersion: v1
+kind: Service
+metadata:
+  name: dynamo-epp-peer
+spec:
+  selector:
+    app: dynamo-epp
+  ports:
     - name: replica-agg
       port: 9092
       targetPort: replica-agg
+    - name: selection-http
+      port: 9093
+      targetPort: selection-http
+---
+# The peer Service's `selection-http` serves the KV-index snapshot
+# (`GET /dump`) without authentication, so it is reachable only from sibling
+# EPP pods; the same peer-only restriction applies to `replica-agg` lifecycle
+# sync. The GAIE gateway still reaches gRPC (9002) and the kubelet still probes
+# grpc-health (9003) from any source; KV events flow *out* of the EPP to
+# workers, so no inbound rule is needed for them. Clusters without a
+# NetworkPolicy CNI simply ignore this object; operators with their own network
+# baseline may drop or replace it.
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: dynamo-epp
+  labels:
+    app: dynamo-epp
+spec:
+  podSelector:
+    matchLabels:
+      app: dynamo-epp
+  policyTypes:
+    - Ingress
+  ingress:
+    - from:
+        - podSelector:
+            matchLabels:
+              app: dynamo-epp
+      ports:
+        - protocol: TCP
+          port: replica-agg
+        - protocol: TCP
+          port: selection-http
+    - from: [] # any source: GAIE gateway + kubelet probes
+      ports:
+        - protocol: TCP
+          port: grpc
+        - protocol: TCP
+          port: grpc-health
 ---
 # InferencePool selects the raw vLLM pods and delegates endpoint selection to
 # the EPP. The EPP reads this object (read-only) to learn the pod selector and

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

@@ -47,14 +47,8 @@ pub struct EppRouter {
     // Kept alive for the lifetime of the router; the reconcile loop runs on it.
     _adapter: TopologyAdapter,
     reflector_ready: Arc<AtomicBool>,
-    /// Peer-discovery readiness (replicated mode only): `None` when replication
-    /// is off, else a flag that latches `true` after the initial peer-set sync
-    /// (EndpointSlice LIST + reconcile). ANDed with `reflector_ready` to form the
-    /// health signal, so a replica does not serve before its peers are discovered
-    /// and its replica-sync sockets are connected. Note: this proves only that
-    /// future load deltas will flow — it does NOT bootstrap the load already in
-    /// flight on peers; that converges from live deltas as pre-existing requests
-    /// drain (same warm-up shape as the KV index).
+    /// Replication bootstrap readiness (replicated mode only): initial peer
+    /// discovery plus KV-index recovery, or authoritative no-peer bootstrap.
     peer_ready: Option<Arc<AtomicBool>>,
     model_name: String,
     /// Bounds total concurrent in-flight `pick()`s. HTTP/2 stream multiplexing
@@ -69,15 +63,17 @@ pub struct EppRouter {
 impl EppRouter {
     /// Assemble the standalone runtime from the validated selector config.
     pub async fn from_selector(cfg: EppStandaloneConfig) -> Result<Self> {
-        let selector = Arc::new(Selector::new(&cfg).await?);
+        let selector = Selector::new(&cfg).await?;
+        Self::from_built_selector(cfg, selector).await
+    }
+
+    pub(crate) async fn from_built_selector(
+        cfg: EppStandaloneConfig,
+        selector: Selector,
+    ) -> Result<Self> {
+        let selector = Arc::new(selector);
         let (renderer, reflector, reflector_ready) = Self::dependencies(&cfg).await?;
-        Ok(Self::from_selector_parts(
-            cfg,
-            renderer,
-            reflector,
-            reflector_ready,
-            selector,
-        ))
+        Self::from_selector_parts(cfg, renderer, reflector, reflector_ready, selector).await
     }
 
     /// Assemble a custom EPP image around a prebuilt selection service.
@@ -87,13 +83,7 @@ impl EppRouter {
     ) -> Result<Self> {
         let selector = Arc::new(Selector::from_service(&cfg, service).await?);
         let (renderer, reflector, reflector_ready) = Self::dependencies(&cfg).await?;
-        Ok(Self::from_selector_parts(
-            cfg,
-            renderer,
-            reflector,
-            reflector_ready,
-            selector,
-        ))
+        Self::from_selector_parts(cfg, renderer, reflector, reflector_ready, selector).await
     }
 
     async fn dependencies(
@@ -111,23 +101,29 @@ impl EppRouter {
         Ok((renderer, reflector, reflector_ready))
     }
 
-    fn from_selector_parts(
+    async fn from_selector_parts(
         cfg: EppStandaloneConfig,
         renderer: VllmRenderClient,
         reflector: Arc<PodDiscovery>,
         reflector_ready: Arc<AtomicBool>,
         selector: Arc<Selector>,
-    ) -> Self {
+    ) -> Result<Self> {
         let peer_ready = selector.peer_ready();
 
         let defaults = RegistrationDefaults::from_config(&cfg);
         let adapter =
             TopologyAdapter::spawn(reflector.as_ref().clone(), selector.clone(), defaults);
 
+        // Subscribe-first: the topology adapter registers workers (their ZMQ
+        // KV-event listeners begin buffering live events) before the peer dump,
+        // so recovery overlaps the live event stream instead of leaving a gap.
+        // `is_ready` stays false (and /dump stays 503) until recovery completes.
+        selector.start_peer_recovery().await?;
+
         // Readiness is driven solely by the live pod+pool signal (see `is_ready`);
         // we do not block startup on a schedulable worker. A valid, empty pool is
         // ready immediately and returns 503 per-request until capacity appears.
-        Self {
+        Ok(Self {
             renderer,
             reflector,
             selector,
@@ -136,12 +132,11 @@ impl EppRouter {
             peer_ready,
             model_name: cfg.model_name,
             inflight: Arc::new(Semaphore::new(cfg.max_inflight_requests)),
-        }
+        })
     }
 
-    /// Overall EPP readiness for the gRPC health signal: the pod reflector is
-    /// ready (workers synced + pool resolved) AND, in replicated mode, the peer
-    /// set has finished its initial sync. Polled by the health mirror in `main`.
+    /// Overall EPP readiness: worker discovery is ready and replicated mode has
+    /// completed peer discovery plus KV-index recovery/bootstrap.
     pub fn is_ready(&self) -> bool {
         compute_ready(
             self.reflector_ready.load(Ordering::Acquire),
@@ -195,8 +190,7 @@ impl EppRouter {
     }
 }
 
-/// Overall EPP health: pod readiness AND, when replicated (`peer_ready = Some`),
-/// the initial peer sync. `None` means no replication → pod readiness alone.
+/// Overall EPP health: pod readiness AND replication bootstrap readiness.
 fn compute_ready(pod_ready: bool, peer_ready: Option<bool>) -> bool {
     pod_ready && peer_ready.unwrap_or(true)
 }

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

@@ -19,6 +19,7 @@ pub mod epp_standalone_config;
 pub mod inference_pool;
 pub mod metrics;
 pub mod peer_discovery;
+mod peer_http;
 pub mod picker;
 pub mod pod_discovery;
 pub mod proto;

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

@@ -18,7 +18,9 @@ use axum::{
     routing::get,
 };
 use dynamo_llm::http::service::metrics::generate_log_buckets;
-use prometheus::{Encoder, HistogramOpts, HistogramVec, Registry, TEXT_FORMAT, TextEncoder};
+use prometheus::{
+    Encoder, HistogramOpts, HistogramVec, IntGaugeVec, Opts, Registry, TEXT_FORMAT, TextEncoder,
+};
 
 /// Port the `/metrics` endpoint binds to unless `DYN_EPP_METRICS_PORT` says
 /// otherwise. Distinct from the ext_proc gRPC port (9002) and the health port
@@ -89,6 +91,54 @@ pub fn observe_cached_tokens(cached_tokens: u64) {
         .observe(cached_tokens as f64);
 }
 
+/// Possible startup KV-index recovery outcomes for one EPP replica. The gauge
+/// reports which state is current (1 for it, 0 for the others), so a scrape can
+/// distinguish "recovered from a peer" from "bootstrapped empty" and from
+/// "recovery disabled" — the difference between a normal cold start and a
+/// silent full-index loss.
+pub const KV_RECOVERY_RECOVERED: &str = "recovered";
+pub const KV_RECOVERY_EMPTY_BOOTSTRAP: &str = "empty_bootstrap";
+pub const KV_RECOVERY_DISABLED: &str = "recovery_disabled";
+
+/// Current startup KV-index recovery outcome for this replica. Set once at
+/// startup. A `1` appears on exactly one `state` label; the others are `0`.
+///
+/// Alerts consume the *transition* (a replica that previously exported
+/// `recovered` flipping to `empty_bootstrap` means a full-index loss, whereas
+/// `empty_bootstrap` from first deployment is normal) or a *stuck* `disabled` /
+/// `empty_bootstrap` lasting past the expected upgrade window — not the bare
+/// state, which is expected during first deploy and rolling upgrades.
+static KV_RECOVERY_STATE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
+    let gauge = IntGaugeVec::new(
+        Opts::new(
+            "dynamo_epp_kv_recovery_state",
+            "Startup KV-index peer-recovery outcome for this EPP replica: \
+             recovered (restored from a peer dump), empty_bootstrap (no eligible \
+             serving peer), or recovery_disabled (peer Service lacks the \
+             selection-http port). Exactly one label is 1.",
+        ),
+        &["state"],
+    )
+    .expect("kv_recovery_state gauge options are statically valid");
+    REGISTRY
+        .register(Box::new(gauge.clone()))
+        .expect("kv_recovery_state is the only registrant of its name");
+    gauge
+});
+
+/// Mark this replica's startup KV-index recovery outcome.
+pub fn set_kv_recovery_state(state: &str) {
+    for candidate in [
+        KV_RECOVERY_RECOVERED,
+        KV_RECOVERY_EMPTY_BOOTSTRAP,
+        KV_RECOVERY_DISABLED,
+    ] {
+        KV_RECOVERY_STATE
+            .with_label_values(&[candidate])
+            .set(i64::from(candidate == state));
+    }
+}
+
 /// Serve `/metrics` until the process exits.
 pub async fn serve(port: u16) -> anyhow::Result<()> {
     let app = AxumRouter::new().route("/metrics", get(render));