From 90409c7f8496c8b5e4e1b9cde8dedc304f477214 Mon Sep 17 00:00:00 2001 From: Thach Nguyen Date: Thu, 10 Sep 2026 17:31:25 +0000 Subject: [PATCH 1/6] kv-router: skip partial-prefix BlockStored events instead of exiting vLLM (>= the DeepSeek-V4.1 day-0 image) hashes prefixes every prefix_match_unit (hash_block_size) tokens, which can be finer than the cache block size, and publishes the prompt tail that ends inside a cache block as a BlockStored whose block_size is the sub-block length (32/64/96 for a 128-token block). The standalone indexer treated any block_size mismatch on a main-attention event as a fatal --block-size misconfig and exited, so every V4.1 indexer crash-looped. Treat a proper divisor of the configured block size as a partial-prefix entry: drop it (rate-limited warn). Non-divisor mismatches stay fatal. --- lib/kv-router/src/zmq_wire/convert.rs | 35 +++++++++++++++++++++++++++ lib/kv-router/src/zmq_wire/mod.rs | 5 +++- lib/kv-router/src/zmq_wire/tests.rs | 33 +++++++++++++++++++++++++ 3 files changed, 72 insertions(+), 1 deletion(-) diff --git a/lib/kv-router/src/zmq_wire/convert.rs b/lib/kv-router/src/zmq_wire/convert.rs index 820c0eaf48c2..223d11666b1d 100644 --- a/lib/kv-router/src/zmq_wire/convert.rs +++ b/lib/kv-router/src/zmq_wire/convert.rs @@ -43,6 +43,15 @@ impl std::fmt::Display for ConvertError { impl std::error::Error for ConvertError {} +/// True when a BlockStored's `block_size` is smaller than the configured cache +/// block size: a partial-prefix (hash-boundary) entry -- vLLM hashes prefixes +/// every `prefix_match_unit` tokens, so the prompt tail inside a cache block is +/// published with block_size = k * prefix_match_unit (32/64/96 for a 128-token +/// block). A *larger* event block size can only be a --block-size misconfig. +pub fn is_partial_prefix_entry(event_block_size: usize, kv_block_size: u32) -> bool { + event_block_size > 0 && event_block_size < kv_block_size as usize +} + /// Convert a raw event coming from the ZMQ channel into a placement-aware worker event. pub fn convert_event( raw: RawKvEvent, @@ -73,6 +82,32 @@ pub fn convert_event( kv_cache_spec_kind: _, kv_cache_spec_sliding_window: _, } => { + if !block_hashes.is_empty() + && is_partial_prefix_entry(block_size, kv_block_size) + { + // A partial-prefix entry: vLLM hashes prefixes every + // `prefix_match_unit` (hash_block_size) tokens, which can be + // finer than the cache block size, and publishes the prompt + // tail that ends inside a cache block as a BlockStored whose + // `block_size` is that sub-block length (e.g. 32/64/96 for a + // 128-token block). The indexer keys on whole cache blocks, so + // these carry no routable information: drop them instead of + // treating them as a --block-size misconfiguration. + if warning_count.fetch_add(1, Ordering::Relaxed) < 3 { + tracing::warn!( + event_id, + worker_id = worker.worker_id, + dp_rank = worker.dp_rank, + event_block_size = block_size, + configured_block_size = kv_block_size, + "Skipping sub-block BlockStored: a partial-prefix entry \ + (prefix_match_unit < block_size), unless the indexer's \ + --block-size is larger than the engine's -- then the \ + index stays empty; check configured vs event size" + ); + } + return Ok(None); + } if !block_hashes.is_empty() && block_size != kv_block_size as usize { tracing::error!( event_id, diff --git a/lib/kv-router/src/zmq_wire/mod.rs b/lib/kv-router/src/zmq_wire/mod.rs index 3610db87109d..ae0bb20defbf 100644 --- a/lib/kv-router/src/zmq_wire/mod.rs +++ b/lib/kv-router/src/zmq_wire/mod.rs @@ -23,7 +23,10 @@ mod filter; mod tests; mod types; -pub use convert::{ConvertError, convert_event, create_stored_block_from_parts, create_stored_blocks}; +pub use convert::{ + ConvertError, convert_event, create_stored_block_from_parts, create_stored_blocks, + is_partial_prefix_entry, +}; pub use extra_keys::{extra_keys_to_block_mm_infos, parse_mm_hash_from_extra_key}; pub use filter::KvCacheSpecKind; pub use types::{BlockHashValue, ExtraKeyItem, KvEventBatch, KvTokenIds, RawKvEvent}; diff --git a/lib/kv-router/src/zmq_wire/tests.rs b/lib/kv-router/src/zmq_wire/tests.rs index 027fe3716e65..5e3213508832 100644 --- a/lib/kv-router/src/zmq_wire/tests.rs +++ b/lib/kv-router/src/zmq_wire/tests.rs @@ -548,6 +548,39 @@ fn test_convert_event_block_size_mismatch_is_fatal() { ); } +#[test] +fn test_convert_event_partial_prefix_entry_is_skipped() { + // vLLM with prefix_match_unit (hash_block_size) < block_size publishes the + // prompt tail that ends inside a cache block as a BlockStored whose + // block_size is the sub-block length (32/64/96 for a 128-token block). + // That is not a --block-size misconfiguration: drop it, don't exit. + for partial in [32usize, 64, 96] { + let raw_event = RawKvEvent::BlockStored { + block_hashes: vec![BlockHashValue::Unsigned(21)], + parent_block_hash: Some(BlockHashValue::Unsigned(9)), + token_ids: vec![10; partial], + block_size: partial, + medium: None, + lora_name: None, + block_mm_infos: None, + is_eagle: None, + group_idx: Some(4), + kv_cache_spec_kind: Some(KvCacheSpecKind::MlaAttention), + kv_cache_spec_sliding_window: None, + }; + let warning_count = Arc::new(AtomicU32::new(0)); + let result = + convert_event(raw_event, 7, 128, WorkerWithDpRank::new(3, 0), &warning_count) + .expect("partial-prefix entry is not a config error"); + assert!(result.is_none(), "partial entry of {partial} tokens must be dropped"); + } + // Anything smaller is a partial entry; equal/larger is not. + assert!(is_partial_prefix_entry(96, 128)); + assert!(!is_partial_prefix_entry(256, 128)); + assert!(!is_partial_prefix_entry(128, 128)); + assert!(is_partial_prefix_entry(32, 128)); +} + #[test] fn test_convert_event_empty_store_is_not_fatal() { // No blocks to publish -> nothing can mismatch; must not error. From bb6dcfd5c59c868b82f3230b150c9de17b1ff190 Mon Sep 17 00:00:00 2001 From: Shang-Pin Date: Wed, 16 Sep 2026 23:24:18 +0000 Subject: [PATCH 2/6] standalone-indexer: subscribe to every data-parallel rank of a discovered pod Pod discovery registered each engine pod once, as dp_rank 0 on tcp://:. vLLM offsets the KV-event ZMQ port and the /kv_recover port by data_parallel_rank, so on a --data-parallel-size N engine ranks 1..N-1 publish to ports nobody subscribes to and everything they cache is invisible to /query while the engine still hits it. Measured on deepseek-ai/DeepSeek-V4.1-Flash (DP=2): every prefill the scheduler placed on rank 1 (about half) showed 0 in the standalone indexer and in rank 0's /kv_recover dump, and all its blocks in rank 1's dump; the KV ladder read actual 0.78 > h24 0.65 > perfect 0.63 > reality 0.59. Add --watch-dp-size (default 1, so single-rank engines are unchanged). The watcher registers one listener per rank under the pod's instance: dp_rank r, tcp://:, http://:. register() rejects a rank that is already present, so a pod whose registration fails part-way is deregistered before the retry instead of being left half-subscribed forever. Co-Authored-By: Claude Fable 5.1 --- lib/bindings/python/rust/llm/kv.rs | 12 ++ lib/kv-router/src/standalone_indexer/docs.md | 11 ++ lib/kv-router/src/standalone_indexer/mod.rs | 11 +- .../src/standalone_indexer/pod_watcher.rs | 140 +++++++++++++++--- 4 files changed, 151 insertions(+), 23 deletions(-) diff --git a/lib/bindings/python/rust/llm/kv.rs b/lib/bindings/python/rust/llm/kv.rs index 85a033527092..3bec45699ea2 100644 --- a/lib/bindings/python/rust/llm/kv.rs +++ b/lib/bindings/python/rust/llm/kv.rs @@ -118,6 +118,14 @@ struct KvIndexerCli { #[arg(long)] watch_recover_port: Option, + /// 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=` (the stable label the backend stamps @@ -161,11 +169,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, diff --git a/lib/kv-router/src/standalone_indexer/docs.md b/lib/kv-router/src/standalone_indexer/docs.md index 71c50a687f69..aa9537c6d05b 100644 --- a/lib/kv-router/src/standalone_indexer/docs.md +++ b/lib/kv-router/src/standalone_indexer/docs.md @@ -26,6 +26,17 @@ 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`). +## 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 diff --git a/lib/kv-router/src/standalone_indexer/mod.rs b/lib/kv-router/src/standalone_indexer/mod.rs index ee71fd60c16a..8a03ecf151c2 100644 --- a/lib/kv-router/src/standalone_indexer/mod.rs +++ b/lib/kv-router/src/standalone_indexer/mod.rs @@ -124,12 +124,17 @@ pub struct KubeDiscoveryConfig { /// from the model name (`di/model_name=`, 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://:` is the base - /// URL the indexer queries on a detected gap. + /// per-worker gap recovery. `http://:` is the + /// base URL the indexer queries for rank `r` on a detected gap. pub recover_port: Option, + /// 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. diff --git a/lib/kv-router/src/standalone_indexer/pod_watcher.rs b/lib/kv-router/src/standalone_indexer/pod_watcher.rs index 376cf9051af2..ada1a33fb022 100644 --- a/lib/kv-router/src/standalone_indexer/pod_watcher.rs +++ b/lib/kv-router/src/standalone_indexer/pod_watcher.rs @@ -119,6 +119,7 @@ pub fn spawn_pod_watcher( namespace = %config.namespace, label_selector = %config.label_selector, zmq_port = config.zmq_port, + dp_size = config.dp_size, model_name = %config.model_name, block_size = config.block_size, "Starting Kubernetes pod watcher" @@ -259,30 +260,35 @@ async fn reconcile( let instance_id = instance_id_for(name); let ip = &pod.ip; let tenant_id = pod.tenant_id(&config.tenant_id); - let endpoint = format!("tcp://{ip}:{}", config.zmq_port); - let recover_endpoint = config - .recover_port - .map(|port| format!("http://{ip}:{port}")); - match registry - .register( - instance_id, - endpoint, - 0, // dp_rank: single-rank engines - config.model_name.clone(), - tenant_id.clone(), - config.block_size, - recover_endpoint, - Some(name.clone()), - ) - .await - { - Ok(()) => { + let mut failed: Option<(u32, anyhow::Error)> = None; + for rank in rank_endpoints(ip, config) { + if let Err(error) = registry + .register( + instance_id, + rank.endpoint, + rank.dp_rank, + config.model_name.clone(), + tenant_id.clone(), + config.block_size, + rank.recover_endpoint, + Some(name.clone()), + ) + .await + { + failed = Some((rank.dp_rank, error)); + break; + } + } + + match failed { + None => { tracing::info!( pod = %name, ip = %ip, instance_id, tenant_id = %tenant_id, + dp_size = config.dp_size, "Subscribed to engine pod" ); subscribed.insert( @@ -294,19 +300,57 @@ async fn reconcile( }, ); } - Err(error) => { - // Not recorded, so it is retried on the next reconcile. + Some((dp_rank, error)) => { + // Not recorded, so it is retried on the next reconcile. Drop + // any ranks that did register: register() rejects a rank that + // is already present, so a partial instance would never + // converge on retry. tracing::warn!( pod = %name, ip = %ip, + dp_rank, error = %error, "Failed to register engine pod; will retry" ); + if let Err(error) = registry + .deregister(instance_id, &config.model_name, &tenant_id) + .await + { + tracing::debug!( + pod = %name, + error = %error, + "Cleanup deregister was a no-op" + ); + } } } } } +/// One listener to register for a pod: its data-parallel rank and the +/// per-rank event and recovery endpoints. +#[derive(Debug, PartialEq)] +struct RankEndpoint { + dp_rank: u32, + endpoint: String, + recover_endpoint: Option, +} + +/// Endpoints for every data-parallel rank of a pod. vLLM offsets both the ZMQ +/// KV-event port and the `/kv_recover` port by the rank, so rank `r` of a pod +/// at `ip` publishes on `zmq_port + r` and recovers on `recover_port + r`. +fn rank_endpoints(ip: &str, config: &KubeDiscoveryConfig) -> Vec { + (0..config.dp_size.max(1)) + .map(|dp_rank| RankEndpoint { + dp_rank, + endpoint: format!("tcp://{ip}:{}", u32::from(config.zmq_port) + dp_rank), + recover_endpoint: config + .recover_port + .map(|port| format!("http://{ip}:{}", u32::from(port) + dp_rank)), + }) + .collect() +} + /// Derive a stable `WorkerId` from the pod name so the same pod always maps to /// the same registry entry across MODIFIED events. fn instance_id_for(pod_name: &str) -> WorkerId { @@ -456,4 +500,60 @@ mod tests { assert_eq!(instance_id_for("engine-abc"), instance_id_for("engine-abc")); assert_ne!(instance_id_for("engine-abc"), instance_id_for("engine-xyz")); } + + fn discovery(dp_size: u32, recover_port: Option) -> KubeDiscoveryConfig { + KubeDiscoveryConfig { + namespace: "deepinfra".to_string(), + label_selector: "di/model_name=m".to_string(), + zmq_port: 5557, + recover_port, + dp_size, + model_name: "m".to_string(), + tenant_id: "default".to_string(), + block_size: 128, + } + } + + #[test] + fn single_rank_engine_registers_rank_zero_on_base_ports() { + assert_eq!( + rank_endpoints("10.0.0.1", &discovery(1, Some(5559))), + vec![RankEndpoint { + dp_rank: 0, + endpoint: "tcp://10.0.0.1:5557".to_string(), + recover_endpoint: Some("http://10.0.0.1:5559".to_string()), + }] + ); + } + + #[test] + fn data_parallel_ranks_offset_event_and_recovery_ports() { + assert_eq!( + rank_endpoints("10.0.0.1", &discovery(2, Some(5559))), + vec![ + RankEndpoint { + dp_rank: 0, + endpoint: "tcp://10.0.0.1:5557".to_string(), + recover_endpoint: Some("http://10.0.0.1:5559".to_string()), + }, + RankEndpoint { + dp_rank: 1, + endpoint: "tcp://10.0.0.1:5558".to_string(), + recover_endpoint: Some("http://10.0.0.1:5560".to_string()), + }, + ] + ); + } + + #[test] + fn no_recover_port_means_no_recovery_endpoint_on_any_rank() { + let ranks = rank_endpoints("10.0.0.1", &discovery(2, None)); + assert_eq!(ranks.len(), 2); + assert!(ranks.iter().all(|r| r.recover_endpoint.is_none())); + } + + #[test] + fn zero_dp_size_still_registers_rank_zero() { + assert_eq!(rank_endpoints("10.0.0.1", &discovery(0, None)).len(), 1); + } } From 37e1b1a26b6e3aaee3ea805e64cecf29e2891f10 Mon Sep 17 00:00:00 2001 From: Shang-Pin Date: Thu, 17 Sep 2026 21:21:55 +0000 Subject: [PATCH 3/6] standalone-indexer: bound, gate and retry /kv_recover downloads A listener that detects a gap fetches the worker's dump with a reqwest client whose total timeout was 10 s, and a failed fetch was final: the live batch that revealed the gap then advanced the watermark past it, so the missed range was lost for good. On DeepSeek-V4.1-Flash engines (DP=2, 28M-token KV pools) a TreeDump is ~70 MB serialized inside EngineCore. When three indexer flavors started against 39 such pods, each engine got six concurrent dump requests, a third of them ran past 10 s, the indexer logged "error decoding response body", the engine logged BrokenPipeError in kv_events.py do_GET, and the affected listeners carried orphaned chains (ParentBlockNotFound floods) from then on. - --recover-timeout-secs (default 120) replaces the hard-coded 10 s; connect timeout stays short at 5 s. - --recover-concurrency (default 8): a process-wide semaphore around the download, so a fleet-wide (re)subscription queues instead of stampeding every engine. - A failed download is retried up to 3 times with 2 s / 4 s backoff before the gap is given up; the final failure keeps the "kv_recover request failed" substring ops already grep for. Co-Authored-By: Claude Fable 5.1 --- lib/bindings/python/rust/llm/kv.rs | 12 ++ lib/kv-router/src/standalone_indexer/docs.md | 8 ++ .../src/standalone_indexer/listener.rs | 130 ++++++++++++++---- lib/kv-router/src/standalone_indexer/mod.rs | 31 +++++ 4 files changed, 152 insertions(+), 29 deletions(-) diff --git a/lib/bindings/python/rust/llm/kv.rs b/lib/bindings/python/rust/llm/kv.rs index 3bec45699ea2..d29598b32490 100644 --- a/lib/bindings/python/rust/llm/kv.rs +++ b/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 @@ -205,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, })) } diff --git a/lib/kv-router/src/standalone_indexer/docs.md b/lib/kv-router/src/standalone_indexer/docs.md index aa9537c6d05b..12f2c98f8bd9 100644 --- a/lib/kv-router/src/standalone_indexer/docs.md +++ b/lib/kv-router/src/standalone_indexer/docs.md @@ -26,6 +26,14 @@ 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`). +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. + ## Data-parallel engines (`--watch-dp-size`) A vLLM engine with `--data-parallel-size N` runs N ranks per pod, each with its diff --git a/lib/kv-router/src/standalone_indexer/listener.rs b/lib/kv-router/src/standalone_indexer/listener.rs index 3a4350a09d2a..135b9eb7c698 100644 --- a/lib/kv-router/src/standalone_indexer/listener.rs +++ b/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 @@ -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") } diff --git a/lib/kv-router/src/standalone_indexer/mod.rs b/lib/kv-router/src/standalone_indexer/mod.rs index 8a03ecf151c2..921f16bf78a7 100644 --- a/lib/kv-router/src/standalone_indexer/mod.rs +++ b/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 = OnceLock::new(); static ENABLE_LOGGING: OnceLock = OnceLock::new(); +static RECOVER_TIMEOUT_SECS: OnceLock = OnceLock::new(); +static RECOVER_GATE: OnceLock = 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 @@ -168,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<()> { @@ -275,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", From ae7403cbee8e87679d79f9ab8bad106d0768bc42 Mon Sep 17 00:00:00 2001 From: Shang-Pin Date: Thu, 17 Sep 2026 22:30:08 +0000 Subject: [PATCH 4/6] standalone-indexer: unbounded ZMQ receive queue on SUB sockets A listener blocked in /kv_recover (now up to 120 s with the gate) can let its SUB pipe hit the default RCVHWM of 1000. With heartbeats enabled libzmq 4.3.4 then aborts on `Assertion failed: _input_stopped` (zeromq/libzmq#3596, #3937). Hit in prod on frank/DeepSeek-V4.1-Flash kv-indexer:h24: 6 restarts in 10 min while 66 startup TreeDumps were applied under the single H24Indexer mutex. Co-Authored-By: Claude Fable 5.1 --- lib/kv-router/src/standalone_indexer/docs.md | 6 ++++++ lib/kv-router/src/standalone_indexer/zmq.rs | 17 +++++++++++++++++ 2 files changed, 23 insertions(+) diff --git a/lib/kv-router/src/standalone_indexer/docs.md b/lib/kv-router/src/standalone_indexer/docs.md index 12f2c98f8bd9..22a63086f621 100644 --- a/lib/kv-router/src/standalone_indexer/docs.md +++ b/lib/kv-router/src/standalone_indexer/docs.md @@ -26,6 +26,12 @@ 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 diff --git a/lib/kv-router/src/standalone_indexer/zmq.rs b/lib/kv-router/src/standalone_indexer/zmq.rs index 98bd7ddd5a65..3d47f04db6bc 100644 --- a/lib/kv-router/src/standalone_indexer/zmq.rs +++ b/lib/kv-router/src/standalone_indexer/zmq.rs @@ -15,6 +15,10 @@ pub(super) type MultipartMessage = Vec>; pub(super) type SharedSocket = Arc>; const ZMQ_RCVTIMEOUT_MS: i32 = 100; +// Unbounded on purpose: libzmq 4.3.4 asserts (`_input_stopped`) when a +// heartbeating peer restarts a pipe that HWM-stopped input, so a listener +// stalled in recovery must never let its queue fill (zeromq/libzmq#3596). +const ZMQ_RCVHWM: i32 = 0; #[cfg(test)] const ZMQ_SNDTIMEOUT_MS: i32 = 0; const ZMQ_RECONNECT_IVL_MS: i32 = 100; @@ -151,6 +155,7 @@ fn configure_common_socket(socket: &zmq::Socket) -> Result<()> { fn configure_receive_socket(socket: &zmq::Socket) -> Result<()> { configure_common_socket(socket)?; socket.set_rcvtimeo(ZMQ_RCVTIMEOUT_MS)?; + socket.set_rcvhwm(ZMQ_RCVHWM)?; Ok(()) } @@ -203,3 +208,15 @@ pub(super) async fn send_multipart(socket: &SharedSocket, frames: MultipartMessa .collect::>(); poll_fn(|cx| socket.poll_send_multipart(cx, &mut buffer)).await } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn sub_socket_receive_queue_is_unbounded() { + let socket = connect_sub_socket("inproc://rcvhwm-test").unwrap(); + let guard = socket.lock().await; + assert_eq!(guard.socket().get_rcvhwm().unwrap(), 0); + } +} From f4b4f1d20ac442ebc6d72e3c960dc3d779ff4dd2 Mon Sep 17 00:00:00 2001 From: Shang-Pin Date: Thu, 17 Sep 2026 22:39:35 +0000 Subject: [PATCH 5/6] indexer image: stamp archived sources with build time Two concurrent build-indexer-image.sh runs on different branches share the /cargo-target cache mount. git archive gives every file the commit mtime, so the second build saw a kv-router rlib the first build had just written, judged it fresh, and failed compiling the bindings against a struct from the other branch. Touch the tree after extraction. Co-Authored-By: Claude Fable 5.1 --- container/indexer.Dockerfile | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/container/indexer.Dockerfile b/container/indexer.Dockerfile index f4321129a204..191cecc9c4e3 100644 --- a/container/indexer.Dockerfile +++ b/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 From c707ba7adbaf18ef047374c5737ed9495f257717 Mon Sep 17 00:00:00 2001 From: Shang-Pin Date: Tue, 22 Sep 2026 20:51:17 +0000 Subject: [PATCH 6/6] standalone-indexer: hash image blocks from tokens only vLLM attaches each image's identifier to a stored block in extra_keys, and the shared ZMQ normalizer mixed it into the block's tokens hash. The standalone indexer's queriers hash plain token ids: deepapi's probe does, and so do the engine local-indexer TreeDumps served by /kv_recover. Every query therefore stopped matching at a conversation's first image block. Measured on frank/DeepSeek-V4.1-Flash with a live ZMQ capture: for chains whose first image fell in the window, a plain probe matched exactly up to the image block (217, 1098, 1569, 358 blocks) while an image-aware probe matched the whole chain (589, 2432, 3054, 6450). The indexer predicted 0.77 vs actual 0.96 on prompts over 500k tokens. Add ZmqEventNormalizer::with_plain_mm_hashing() and use it in the standalone listener. The library default is unchanged for Dynamo's own router, which queries with image info. Co-Authored-By: Claude Opus 5.5 (1M context) --- lib/kv-router/src/standalone_indexer/docs.md | 12 ++++ .../src/standalone_indexer/listener.rs | 2 +- lib/kv-router/src/zmq_wire/mod.rs | 17 +++++ lib/kv-router/src/zmq_wire/tests.rs | 66 +++++++++++++++++++ lib/kv-router/src/zmq_wire/types.rs | 31 +++++++++ 5 files changed, 127 insertions(+), 1 deletion(-) diff --git a/lib/kv-router/src/standalone_indexer/docs.md b/lib/kv-router/src/standalone_indexer/docs.md index 22a63086f621..1819407421d6 100644 --- a/lib/kv-router/src/standalone_indexer/docs.md +++ b/lib/kv-router/src/standalone_indexer/docs.md @@ -40,6 +40,18 @@ 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 diff --git a/lib/kv-router/src/standalone_indexer/listener.rs b/lib/kv-router/src/standalone_indexer/listener.rs index 135b9eb7c698..0a8fd841a0b2 100644 --- a/lib/kv-router/src/standalone_indexer/listener.rs +++ b/lib/kv-router/src/standalone_indexer/listener.rs @@ -156,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, } } diff --git a/lib/kv-router/src/zmq_wire/mod.rs b/lib/kv-router/src/zmq_wire/mod.rs index ae0bb20defbf..ecc276a685eb 100644 --- a/lib/kv-router/src/zmq_wire/mod.rs +++ b/lib/kv-router/src/zmq_wire/mod.rs @@ -42,6 +42,7 @@ pub struct ZmqEventNormalizer { kv_block_size: u32, warning_count: Arc, group_metadata: FxHashMap<(DpRank, u32), KvCacheGroupMetadata>, + hash_mm: bool, } #[derive(Debug, Clone, Copy)] @@ -77,6 +78,7 @@ impl ZmqEventNormalizer { kv_block_size, warning_count: Arc::new(AtomicU32::new(0)), group_metadata: FxHashMap::default(), + hash_mm: true, } } @@ -85,9 +87,19 @@ impl ZmqEventNormalizer { kv_block_size, warning_count, group_metadata: FxHashMap::default(), + hash_mm: true, } } + /// Hash image blocks from their tokens alone, ignoring the multimodal + /// identifiers vLLM attaches in `extra_keys`. Needed when the querier + /// hashes plain tokens (deepapi's probe, the engine's local-indexer + /// TreeDumps): otherwise every chain diverges at its first image block. + pub fn with_plain_mm_hashing(mut self) -> Self { + self.hash_mm = false; + self + } + pub fn preprocess(&mut self, raw: RawKvEvent, worker: WorkerWithDpRank) -> Option { self.preprocess_with_reason(raw, worker).ok() } @@ -117,6 +129,11 @@ impl ZmqEventNormalizer { event_id: u64, worker: WorkerWithDpRank, ) -> Result, ConvertError> { + let raw = if self.hash_mm { + raw + } else { + raw.without_mm_infos() + }; convert_event( raw, event_id, diff --git a/lib/kv-router/src/zmq_wire/tests.rs b/lib/kv-router/src/zmq_wire/tests.rs index 5e3213508832..b96256989a36 100644 --- a/lib/kv-router/src/zmq_wire/tests.rs +++ b/lib/kv-router/src/zmq_wire/tests.rs @@ -641,3 +641,69 @@ fn test_convert_event_short_token_ids_keeps_parsed_blocks() { other => panic!("expected Stored event, got {other:?}"), } } + +fn stored_image_block(tokens: Vec) -> RawKvEvent { + RawKvEvent::BlockStored { + block_hashes: vec![BlockHashValue::Unsigned(21)], + parent_block_hash: None, + block_size: tokens.len(), + token_ids: tokens, + medium: None, + lora_name: None, + block_mm_infos: Some(vec![Some(BlockExtraInfo { + mm_objects: vec![BlockMmObjectInfo { + mm_hash: 0x5083_86df_2042_9a4f, + offsets: vec![], + }], + })]), + is_eagle: None, + group_idx: None, + kv_cache_spec_kind: None, + kv_cache_spec_sliding_window: None, + } +} + +fn stored_tokens_hash(event: Option) -> u64 { + match event.expect("stored event converts").event.data { + KvCacheEventData::Stored(data) => data.blocks[0].tokens_hash.0, + other => panic!("expected Stored, got {other:?}"), + } +} + +#[test] +fn test_plain_mm_hashing_matches_token_only_probe() { + let tokens: Vec = (0..4).map(|t| 129_264 + t).collect(); + let token_only = compute_block_hash_for_seq( + &tokens, + 4, + BlockHashOptions { + block_mm_infos: None, + lora_name: None, + is_eagle: None, + }, + )[0] + .0; + let worker = WorkerWithDpRank::new(3, 0); + + let mut plain = ZmqEventNormalizer::new(4).with_plain_mm_hashing(); + let plain_hash = stored_tokens_hash( + plain + .normalize(stored_image_block(tokens.clone()), 1, worker) + .unwrap(), + ); + assert_eq!( + plain_hash, token_only, + "plain mode must ignore the image hash" + ); + + let mut mm_aware = ZmqEventNormalizer::new(4); + let mm_hash = stored_tokens_hash( + mm_aware + .normalize(stored_image_block(tokens), 1, worker) + .unwrap(), + ); + assert_ne!( + mm_hash, token_only, + "default mode still mixes the image hash in" + ); +} diff --git a/lib/kv-router/src/zmq_wire/types.rs b/lib/kv-router/src/zmq_wire/types.rs index a0d7ff7a3975..f8a80b1be212 100644 --- a/lib/kv-router/src/zmq_wire/types.rs +++ b/lib/kv-router/src/zmq_wire/types.rs @@ -97,6 +97,37 @@ pub enum RawKvEvent { } impl RawKvEvent { + pub fn without_mm_infos(self) -> Self { + match self { + Self::BlockStored { + block_hashes, + parent_block_hash, + token_ids, + block_size, + medium, + lora_name, + block_mm_infos: _, + is_eagle, + group_idx, + kv_cache_spec_kind, + kv_cache_spec_sliding_window, + } => Self::BlockStored { + block_hashes, + parent_block_hash, + token_ids, + block_size, + medium, + lora_name, + block_mm_infos: None, + is_eagle, + group_idx, + kv_cache_spec_kind, + kv_cache_spec_sliding_window, + }, + other => other, + } + } + pub fn event_type_label(&self) -> &'static str { match self { Self::BlockStored { .. } => "stored",