kv-router: skip partial-prefix BlockStored events instead of exiting

container/indexer.Dockerfile

@@ -45,6 +45,11 @@ ENV PATH=/opt/maturin/bin:$PATH
 # Source tarball (git archive), auto-extracted by ADD. rust-toolchain.toml lands
 # at /src so rustup resolves 1.93.1 for the whole workspace.
 ADD dynamo-src.tar /src
+# git archive stamps every file with the commit time, and cargo judges a path
+# crate fresh by mtime against the shared /cargo-target cache. A sibling build
+# (another branch) can therefore leave a newer artifact that cargo reuses with
+# the wrong contents; stamping the sources with the build time prevents that.
+RUN find /src -type f -exec touch {} +
 
 # Build the release wheel. kv-indexer-metrics adds the `dynamo.indexer` binary
 # + Prometheus /metrics. nixl-sys' build script runs bindgen, so point it at

lib/bindings/python/rust/llm/kv.rs

@@ -89,6 +89,16 @@ struct KvIndexerCli {
     #[arg(long, default_value_t = 0.75)]
     evict_memory_threshold: f64,
 
+    /// Total timeout in seconds for one engine `GET /kv_recover` download
+    /// (connect + body). A full TreeDump of a large engine is tens of MB.
+    #[arg(long, default_value_t = 120)]
+    recover_timeout_secs: u64,
+
+    /// Maximum concurrent `/kv_recover` downloads across all listeners of this
+    /// indexer. Bounds the load a fleet-wide (re)subscription puts on engines.
+    #[arg(long, default_value_t = 8)]
+    recover_concurrency: usize,
+
     /// Emit verbose audit logs on the `kv_audit` tracing target: one line per
     /// query (block hashes + full indexer response) and one line per
     /// store/evict/clear event ingested from the engine. Filter with
@@ -118,6 +128,14 @@ struct KvIndexerCli {
     #[arg(long)]
     watch_recover_port: Option<u16>,
 
+    /// Data-parallel ranks per discovered engine pod (vLLM
+    /// `--data-parallel-size`). Rank r publishes KV events on
+    /// `--watch-zmq-port + r` and serves recovery on `--watch-recover-port + r`;
+    /// every rank is subscribed under the pod's instance. Leaving this at 1 on
+    /// a DP engine silently indexes only rank 0's cache.
+    #[arg(long, default_value_t = 1)]
+    watch_dp_size: u32,
+
     /// Model name whose engine pods to discover, e.g. "openai/gpt-oss-120b".
     /// Unless --watch-label overrides it, the pod watch uses the selector
     /// `di/model_name=<sanitized name>` (the stable label the backend stamps
@@ -161,11 +179,15 @@ where
                 let block_size = cli.block_size.ok_or_else(|| {
                     anyhow::anyhow!("--block-size is required when --watch-namespace is set")
                 })?;
+                if cli.watch_dp_size == 0 {
+                    anyhow::bail!("--watch-dp-size must be at least 1");
+                }
                 Some(KubeDiscoveryConfig {
                     namespace,
                     label_selector,
                     zmq_port: cli.watch_zmq_port,
                     recover_port: cli.watch_recover_port,
+                    dp_size: cli.watch_dp_size,
                     model_name: cli.watch_model_name.unwrap_or_else(|| cli.model_name.clone()),
                     tenant_id: cli.tenant_id.clone(),
                     block_size,
@@ -193,6 +215,8 @@ where
             evict_retention_secs: cli.evict_retention_secs,
             evict_memory_threshold: cli.evict_memory_threshold,
             enable_logging: cli.enable_logging,
+            recover_timeout_secs: cli.recover_timeout_secs,
+            recover_concurrency: cli.recover_concurrency,
             kube_discovery,
         }))
     }

lib/kv-router/src/standalone_indexer/docs.md

@@ -26,6 +26,43 @@ If no `recover_endpoint` is configured, gaps are logged and the dropped batches
 are lost. Implementation lives in `listener.rs` (`recover_gap`,
 `apply_recover_response`, `apply_recovered_events`).
 
+The SUB sockets run with an unbounded receive queue (`ZMQ_RCVHWM = 0`,
+`zmq.rs`): a listener blocked in recovery must never HWM-stop its pipe, because
+libzmq 4.3.4 aborts on `_input_stopped` when a heartbeating peer restarts such
+a pipe (zeromq/libzmq#3596). Seen in prod as a crash loop of the h24 indexer
+while 66 startup TreeDumps were being applied under the single h24 mutex.
+
+A download runs under a process-wide gate (`--recover-concurrency`, default 8)
+with a total timeout of `--recover-timeout-secs` (default 120), and a failed
+download is retried up to 3 times with 2/4 s backoff before the gap is given up
+(`"kv_recover request failed; giving up, batches lost"`). A large engine's
+TreeDump is tens of MB serialized inside the engine process; with the previous
+10 s timeout and no gate, a fleet-wide (re)subscription lost about a third of
+its recoveries on 39 DP=2 pods.
+
+## Image blocks hash from tokens only
+
+vLLM attaches each image's identifier to a stored block in `extra_keys`, and the
+shared ZMQ normalizer would mix it into that block's tokens hash. The standalone
+indexer's queriers do not: deepapi's probe hashes plain token ids and the
+engine's local-indexer TreeDumps (`/kv_recover`) carry token-only hashes. With
+the image hash mixed in, every query stopped matching at a conversation's first
+image block. The listener therefore builds its normalizer with
+`with_plain_mm_hashing()`. Trade-off: two prompts that share text but carry
+different images at the same position are indexed as the same prefix, so the
+indexer can over-report a hit the engine will not give.
+
+## Data-parallel engines (`--watch-dp-size`)
+
+A vLLM engine with `--data-parallel-size N` runs N ranks per pod, each with its
+own KV cache and its own event stream: rank `r` publishes on `zmq_port + r` and
+serves `/kv_recover` on `kv_recover_port + r`. Pod discovery registers one
+listener per rank under the pod's instance (`--watch-dp-size N`, default 1),
+with `dp_rank = r` and the per-rank endpoints, so `/workers` shows N
+`listeners` per pod. With the default on a DP engine only rank 0's cache is
+indexed and every prefill scheduled on another rank is invisible to `/query`
+while the engine still hits it (`pod_watcher.rs`, `rank_endpoints`).
+
 ## Audit logging (`--enable-logging`)
 
 Pass `--enable-logging` to `python -m dynamo.indexer` to turn on verbose audit

lib/kv-router/src/standalone_indexer/listener.rs

@@ -20,6 +20,15 @@ use super::zmq::{MultipartMessage, SharedSocket, connect_sub_socket, recv_multip
 
 const WATERMARK_UNSET: u64 = u64::MAX;
 
+/// Attempts per detected gap before the missed batches are given up as lost.
+const RECOVER_ATTEMPTS: u32 = 3;
+
+/// Delay before retry `attempt` (0-based) of a failed `/kv_recover` download:
+/// 2 s, 4 s, 8 s, ... capped at 32 s.
+fn recover_backoff(attempt: u32) -> std::time::Duration {
+    std::time::Duration::from_secs(2u64 << attempt.min(4))
+}
+
 fn cursor_from_watermark(watermark: u64) -> CursorState {
     if watermark == WATERMARK_UNSET {
         CursorState::Initial
@@ -147,7 +156,7 @@ impl ListenerLoop {
             http_client,
             watermark,
             pending_evictions,
-            normalizer: ZmqEventNormalizer::new(block_size),
+            normalizer: ZmqEventNormalizer::new(block_size).with_plain_mm_hashing(),
             messages_processed: 0,
         }
     }
@@ -213,39 +222,78 @@ impl ListenerLoop {
         };
 
         let url = format!("{}/kv_recover", recover_endpoint.trim_end_matches('/'));
-        let client = self.http_client.clone();
         let cancel = self.cancel.clone();
         let worker_id = self.worker_id;
         let dp_rank = self.dp_rank;
 
-        let fetch = async move {
-            let response = client
-                .get(&url)
-                .query(&[("start", start_seq), ("end", end_seq)])
-                .send()
-                .await?;
-            if !response.status().is_success() {
-                anyhow::bail!("kv_recover returned status {}", response.status());
-            }
-            let body: WorkerKvQueryResponse = response.json().await?;
-            anyhow::Ok(body)
-        };
+        for attempt in 0..RECOVER_ATTEMPTS {
+            // One dump at a time per permit: a fleet-wide (re)subscription
+            // otherwise asks every engine for several large TreeDumps at once
+            // and the downloads run past the client timeout.
+            let permit = tokio::select! {
+                _ = cancel.cancelled() => {
+                    tracing::debug!(worker_id, dp_rank, "Recovery cancelled");
+                    return 0;
+                }
+                permit = super::recover_gate().acquire() => {
+                    permit.expect("recovery gate is never closed")
+                }
+            };
 
-        let response = tokio::select! {
-            _ = cancel.cancelled() => {
-                tracing::debug!(worker_id, dp_rank, "Recovery cancelled");
-                return 0;
-            }
-            result = fetch => match result {
-                Ok(body) => body,
-                Err(error) => {
-                    tracing::error!(worker_id, dp_rank, error = %error, "kv_recover request failed");
+            let client = self.http_client.clone();
+            let request_url = url.clone();
+            let fetch = async move {
+                let response = client
+                    .get(&request_url)
+                    .query(&[("start", start_seq), ("end", end_seq)])
+                    .send()
+                    .await?;
+                if !response.status().is_success() {
+                    anyhow::bail!("kv_recover returned status {}", response.status());
+                }
+                let body: WorkerKvQueryResponse = response.json().await?;
+                anyhow::Ok(body)
+            };
+
+            let result = tokio::select! {
+                _ = cancel.cancelled() => {
+                    tracing::debug!(worker_id, dp_rank, "Recovery cancelled");
                     return 0;
                 }
-            }
-        };
+                result = fetch => result,
+            };
+            drop(permit);
 
-        self.apply_recover_response(response).await
+            let error = match result {
+                Ok(body) => return self.apply_recover_response(body).await,
+                Err(error) => error,
+            };
+            if attempt + 1 == RECOVER_ATTEMPTS {
+                tracing::error!(
+                    worker_id,
+                    dp_rank,
+                    attempts = RECOVER_ATTEMPTS,
+                    gap_size = end_seq.saturating_sub(start_seq),
+                    error = %error,
+                    "kv_recover request failed; giving up, batches lost"
+                );
+                return 0;
+            }
+            let delay = recover_backoff(attempt);
+            tracing::warn!(
+                worker_id,
+                dp_rank,
+                attempt = attempt + 1,
+                retry_in_secs = delay.as_secs(),
+                error = %error,
+                "kv_recover request failed; retrying"
+            );
+            tokio::select! {
+                _ = cancel.cancelled() => return 0,
+                _ = tokio::time::sleep(delay) => {}
+            }
+        }
+        0
     }
 
     /// Apply a [`WorkerKvQueryResponse`] to this listener's indexer and advance
@@ -598,12 +646,16 @@ async fn run_listener(
     .await
 }
 
-/// Build the HTTP client used for `/kv_recover` gap-recovery requests. The 10s
-/// timeout bounds the whole request (connect + body read). Recovery is a
-/// low-frequency, on-gap operation, so a fresh client per listener is fine.
+/// Build the HTTP client used for `/kv_recover` gap-recovery requests. The
+/// total timeout bounds the whole request (connect + body read) and comes from
+/// `--recover-timeout-secs`; a large engine's TreeDump is tens of MB produced
+/// inside the engine process, which the old 10 s bound did not cover under
+/// concurrent recoveries. Recovery is a low-frequency, on-gap operation, so a
+/// fresh client per listener is fine.
 fn build_recover_client() -> reqwest::Client {
     reqwest::Client::builder()
-        .timeout(std::time::Duration::from_secs(10))
+        .connect_timeout(std::time::Duration::from_secs(5))
+        .timeout(std::time::Duration::from_secs(super::recover_timeout_secs()))
         .build()
         .unwrap_or_else(|error| {
             tracing::warn!(error = %error, "Failed to build recover HTTP client; using default");
@@ -899,6 +951,26 @@ mod tests {
         assert_eq!(msg, vec![b"probe".to_vec()]);
     }
 
+    #[test]
+    fn recover_backoff_doubles_and_caps() {
+        assert_eq!(super::recover_backoff(0).as_secs(), 2);
+        assert_eq!(super::recover_backoff(1).as_secs(), 4);
+        assert_eq!(super::recover_backoff(2).as_secs(), 8);
+        assert_eq!(super::recover_backoff(10).as_secs(), 32);
+    }
+
+    #[test]
+    fn recover_gate_defaults_when_unconfigured() {
+        assert_eq!(
+            crate::standalone_indexer::recover_gate().available_permits(),
+            crate::standalone_indexer::DEFAULT_RECOVER_CONCURRENCY
+        );
+        assert_eq!(
+            crate::standalone_indexer::recover_timeout_secs(),
+            crate::standalone_indexer::DEFAULT_RECOVER_TIMEOUT_SECS
+        );
+    }
+
     fn reserve_open_port() -> std::net::TcpListener {
         std::net::TcpListener::bind("127.0.0.1:0").expect("failed to bind probe listener")
     }

lib/kv-router/src/standalone_indexer/mod.rs

@@ -52,6 +52,27 @@ use server::{AppState, create_router};
 // exactly once and read many times.
 static KEEP_EVICTIONS: OnceLock<bool> = OnceLock::new();
 static ENABLE_LOGGING: OnceLock<bool> = OnceLock::new();
+static RECOVER_TIMEOUT_SECS: OnceLock<u64> = OnceLock::new();
+static RECOVER_GATE: OnceLock<tokio::sync::Semaphore> = OnceLock::new();
+
+pub(crate) const DEFAULT_RECOVER_TIMEOUT_SECS: u64 = 120;
+pub(crate) const DEFAULT_RECOVER_CONCURRENCY: usize = 8;
+
+/// Total timeout (connect + body) for one `GET /kv_recover` download. A full
+/// TreeDump of a large engine is tens of MB serialized inside the engine
+/// process, so this is minutes, not seconds.
+pub(crate) fn recover_timeout_secs() -> u64 {
+    *RECOVER_TIMEOUT_SECS
+        .get()
+        .unwrap_or(&DEFAULT_RECOVER_TIMEOUT_SECS)
+}
+
+/// Process-wide cap on concurrent `/kv_recover` downloads. Every listener of a
+/// freshly started indexer detects a gap at once; without the cap each engine
+/// is asked for one dump per flavor per dp_rank simultaneously.
+pub(crate) fn recover_gate() -> &'static tokio::sync::Semaphore {
+    RECOVER_GATE.get_or_init(|| tokio::sync::Semaphore::new(DEFAULT_RECOVER_CONCURRENCY))
+}
 
 /// Returns `true` when this indexer instance parks `Removed` events in the
 /// per-listener pending-evictions buffer (and drops `Cleared`) instead of
@@ -124,12 +145,17 @@ pub struct KubeDiscoveryConfig {
     /// from the model name (`di/model_name=<sanitized>`, spans every
     /// engine_hash) or supplied raw, e.g. `engine_hash=d4b7a85131172ca6`.
     pub label_selector: String,
-    /// ZMQ KV-event port the engines publish on (e.g. 5557).
+    /// ZMQ KV-event port the engines publish on (e.g. 5557). Data-parallel
+    /// rank `r` publishes on `zmq_port + r`.
     pub zmq_port: u16,
     /// Optional HTTP port serving `GET /kv_recover` on the engines, used for
-    /// per-worker gap recovery. `http://<pod-ip>:<recover_port>` is the base
-    /// URL the indexer queries on a detected gap.
+    /// per-worker gap recovery. `http://<pod-ip>:<recover_port + r>` is the
+    /// base URL the indexer queries for rank `r` on a detected gap.
     pub recover_port: Option<u16>,
+    /// Data-parallel ranks per engine pod. Each rank has its own KV cache and
+    /// its own event stream, so every rank is registered as a listener of the
+    /// same instance. Engines without data parallelism run one rank.
+    pub dp_size: u32,
     /// Model name discovered pods are registered under.
     pub model_name: String,
     /// Tenant id discovered pods are registered under.
@@ -163,6 +189,12 @@ pub struct IndexerConfig {
     /// Emit verbose per-query and per-event audit logs on the `kv_audit`
     /// tracing target. See [`logging_enabled`].
     pub enable_logging: bool,
+    /// Total timeout in seconds for one `/kv_recover` download. See
+    /// [`recover_timeout_secs`].
+    pub recover_timeout_secs: u64,
+    /// Maximum concurrent `/kv_recover` downloads across all listeners. See
+    /// [`recover_gate`].
+    pub recover_concurrency: usize,
 }
 
 pub(super) fn validate_zmq_endpoint(endpoint: &str) -> anyhow::Result<()> {
@@ -270,6 +302,10 @@ pub async fn run_server(config: IndexerConfig) -> anyhow::Result<()> {
     // run_server is called once, so we discard the result.
     let _ = KEEP_EVICTIONS.set(config.keep_evictions);
     let _ = ENABLE_LOGGING.set(config.enable_logging);
+    let _ = RECOVER_TIMEOUT_SECS.set(config.recover_timeout_secs.max(1));
+    let _ = RECOVER_GATE.set(tokio::sync::Semaphore::new(
+        config.recover_concurrency.max(1),
+    ));
     if config.enable_logging {
         tracing::info!(
             target: "kv_audit",