From 37e1b1a26b6e3aaee3ea805e64cecf29e2891f10 Mon Sep 17 00:00:00 2001 From: Shang-Pin Date: Thu, 17 Sep 2026 21:21:55 +0000 Subject: [PATCH 1/3] 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 2/3] 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 3/3] 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