standalone-indexer (h24 line): bound, gate and retry /kv_recover downloads

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
@@ -221,6 +231,8 @@ where
             enable_logging: cli.enable_logging,
             h24: cli.h24,
             h24_horizon_secs: cli.h24_horizon_secs,
+            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,20 @@ 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.
+
 ## Data-parallel engines (`--watch-dp-size`)
 
 A vLLM engine with `--data-parallel-size N` runs N ranks per pod, each with its

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")
     }

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

@@ -53,6 +53,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
@@ -178,6 +199,12 @@ pub struct IndexerConfig {
     /// stored or touched within this window are dropped. Only meaningful
     /// with `h24`.
     pub h24_horizon_secs: u64,
+    /// 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<()> {
@@ -285,6 +312,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",

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

@@ -15,6 +15,10 @@ pub(super) type MultipartMessage = Vec<Vec<u8>>;
 pub(super) type SharedSocket = Arc<Mutex<ZmqSocket>>;
 
 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::<VecDeque<_>>();
     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);
+    }
+}