feat(kv-router): Route vLLM CPU KV events to HostPinned and count lower-tier applies

lib/kv-router/src/indexer/local.rs

@@ -213,6 +213,9 @@ pub struct LocalKvIndexer {
     /// This stays separate from `event_buffer` so dump wait/build state can be
     /// managed on the async path without holding the buffer lock across `.await`.
     recovery_cache: Arc<RecoverySnapshotCache>,
+    /// Shared metrics handle, also wired into lazily created lower-tier
+    /// indexers so HostPinned/Disk/External traffic is counted too.
+    metrics: Arc<KvIndexerMetrics>,
     /// Maximum number of events to keep in buffer
     max_buffer_size: usize, // Router sets this to WORKER_KV_INDEXER_BUFFER_SIZE
     #[cfg(test)]
@@ -230,7 +233,8 @@ impl LocalKvIndexer {
         max_buffer_size: usize,
     ) -> Self {
         Self {
-            indexer: KvIndexer::new(token, kv_block_size, metrics),
+            indexer: KvIndexer::new(token, kv_block_size, metrics.clone()),
+            metrics,
             lower_tier_indexers: Arc::new(Mutex::new(HashMap::new())),
             event_buffer: Mutex::new(VecDeque::with_capacity(max_buffer_size)),
             recovery_cache: Arc::new(RecoverySnapshotCache::new()),
@@ -684,10 +688,11 @@ impl LocalKvIndexer {
         indexers
             .entry(storage_tier)
             .or_insert_with(|| {
-                Arc::new(ThreadPoolIndexer::new(
+                Arc::new(ThreadPoolIndexer::new_with_metrics(
                     LowerTierIndexer::new(),
                     1,
                     self.block_size(),
+                    Some(self.metrics.clone()),
                 ))
             })
             .clone()

lib/kv-router/src/indexer/lower_tier.rs

@@ -20,7 +20,7 @@ use std::sync::Arc;
 use dashmap::DashMap;
 use rustc_hash::{FxBuildHasher, FxHashMap, FxHashSet};
 
-use super::{KvIndexerMetrics, SyncIndexer, WorkerLookupStats, WorkerTask};
+use super::{EventKind, KvIndexerMetrics, SyncIndexer, WorkerLookupStats, WorkerTask};
 use crate::protocols::{
     ExternalSequenceBlockHash, KvCacheEvent, KvCacheEventData, KvCacheEventError, KvCacheStoreData,
     KvCacheStoredBlockData, LocalBlockHash, OverlapScores, RouterEvent, WorkerWithDpRank,
@@ -499,23 +499,33 @@ impl SyncIndexer for LowerTierIndexer {
     fn worker(
         &self,
         event_receiver: flume::Receiver<WorkerTask>,
-        _metrics: Option<Arc<KvIndexerMetrics>>,
+        metrics: Option<Arc<KvIndexerMetrics>>,
     ) -> anyhow::Result<()> {
         let mut worker_blocks = WorkerBlockIndex::default();
+        let counters = metrics.as_ref().map(|m| m.prebind());
 
         while let Ok(task) = event_receiver.recv() {
             match task {
                 WorkerTask::Event(event) => {
-                    if let Err(error) = self.apply_event(&mut worker_blocks, event) {
+                    let kind = EventKind::of(&event.event.data);
+                    let result = self.apply_event(&mut worker_blocks, event);
+                    if let Err(ref error) = result {
                         tracing::warn!(%error, "Failed to apply lower-tier event");
                     }
+                    if let Some(ref c) = counters {
+                        c.inc(kind, result);
+                    }
                 }
                 WorkerTask::EventWithAck { event, resp } => {
+                    let kind = EventKind::of(&event.event.data);
                     let result = self.apply_event(&mut worker_blocks, event);
                     let applied = result.is_ok();
-                    if let Err(error) = result {
+                    if let Err(ref error) = result {
                         tracing::warn!(%error, "Failed to apply lower-tier event");
                     }
+                    if let Some(ref c) = counters {
+                        c.inc(kind, result);
+                    }
                     let _ = resp.send(applied);
                 }
                 WorkerTask::Anchor { worker, anchor } => {

lib/kv-router/src/indexer/lower_tier_indexers.rs

@@ -18,7 +18,7 @@ use std::collections::HashMap;
 use std::sync::{Arc, RwLock};
 
 use crate::indexer::{
-    LowerTierContinuation, LowerTierIndexer, LowerTierMatchDetails, MatchDetails,
+    KvIndexerMetrics, LowerTierContinuation, LowerTierIndexer, LowerTierMatchDetails, MatchDetails,
     ThreadPoolIndexer, WireTieredMatchDetails,
 };
 use crate::protocols::{LocalBlockHash, StorageTier};
@@ -27,20 +27,36 @@ use crate::protocols::{LocalBlockHash, StorageTier};
 /// non-device [`StorageTier`] that has received at least one event.
 #[derive(Clone)]
 pub struct LowerTierIndexers {
+    metrics: Option<Arc<KvIndexerMetrics>>,
     num_threads: usize,
     block_size: u32,
     indexers: Arc<RwLock<HashMap<StorageTier, Arc<ThreadPoolIndexer<LowerTierIndexer>>>>>,
 }
 
 impl LowerTierIndexers {
+    /// Metrics-less constructor for call sites without a `KvIndexerMetrics` handle.
+    /// Router production assembly should use [`new_with_metrics`](Self::new_with_metrics)
+    /// so lower-tier traffic is included in `kv_cache_events_applied`.
     pub fn new(num_threads: usize, block_size: u32) -> Self {
+        Self::new_with_metrics(num_threads, block_size, None)
+    }
+
+    /// Same as [`new`](Self::new) but wires `kv_cache_events_applied`
+    /// counters into every lazily created per-tier indexer, matching the
+    /// observability of the device-tier path.
+    pub fn new_with_metrics(
+        num_threads: usize,
+        block_size: u32,
+        metrics: Option<Arc<KvIndexerMetrics>>,
+    ) -> Self {
         assert!(
             num_threads > 0,
             "lower-tier indexer threads must be non-zero"
         );
         Self {
             num_threads,
             block_size,
+            metrics,
             indexers: Arc::new(RwLock::new(HashMap::new())),
         }
     }
@@ -60,10 +76,11 @@ impl LowerTierIndexers {
             .unwrap()
             .entry(storage_tier)
             .or_insert_with(|| {
-                Arc::new(ThreadPoolIndexer::new(
+                Arc::new(ThreadPoolIndexer::new_with_metrics(
                     LowerTierIndexer::new(),
                     self.num_threads,
                     self.block_size,
+                    self.metrics.clone(),
                 ))
             })
             .clone()

lib/kv-router/src/protocols.rs

@@ -343,7 +343,7 @@ impl StorageTier {
     pub fn from_kv_medium(medium: &str) -> Option<Self> {
         match medium {
             "GPU" | "DEVICE" => Some(Self::Device),
-            "CPU_PINNED" | "CPU_TIER1" => Some(Self::HostPinned),
+            "CPU" | "CPU_PINNED" | "CPU_TIER1" => Some(Self::HostPinned),
             "CPU_TIER2" | "DISK" | "NVME" => Some(Self::Disk),
             "EXTERNAL" | "NETWORK" | "REMOTE" | "SHARED" => Some(Self::External),
             _ => None,

lib/kv-router/src/zmq_wire/tests.rs

@@ -2,13 +2,13 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use std::sync::Arc;
-use std::sync::atomic::AtomicU32;
+use std::sync::atomic::{AtomicU32, Ordering};
 
 use rmp_serde::{from_slice, to_vec};
 
 use crate::protocols::{
     BlockExtraInfo, BlockHashOptions, BlockMmObjectInfo, ExternalSequenceBlockHash,
-    KvCacheEventData, WorkerWithDpRank, compute_block_hash_for_seq,
+    KvCacheEventData, StorageTier, WorkerWithDpRank, compute_block_hash_for_seq,
 };
 
 use super::filter::KvCacheSpecKind;
@@ -525,3 +525,99 @@ fn test_convert_event_bigram_emits_eagle_windows() {
         other => panic!("expected Stored event, got {other:?}"),
     }
 }
+
+struct CpuBlockStoredFixture<'a> {
+    block_hashes: &'a [u64],
+    token_ids: &'a [u32],
+    block_size: usize,
+    parent_block_hash: Option<u64>,
+}
+
+fn cpu_block_stored(fixture: CpuBlockStoredFixture<'_>) -> RawKvEvent {
+    RawKvEvent::BlockStored {
+        block_hashes: fixture
+            .block_hashes
+            .iter()
+            .copied()
+            .map(BlockHashValue::Unsigned)
+            .collect(),
+        parent_block_hash: fixture.parent_block_hash.map(BlockHashValue::Unsigned),
+        token_ids: fixture.token_ids.to_vec(),
+        block_size: fixture.block_size,
+        medium: Some("CPU".to_string()),
+        lora_name: None,
+        block_mm_infos: None,
+        is_eagle: None,
+        group_idx: None,
+        kv_cache_spec_kind: None,
+        kv_cache_spec_sliding_window: None,
+    }
+}
+
+#[test]
+fn cpu_event_with_placeholder_payload_is_dropped_safely() {
+    let raw = cpu_block_stored(CpuBlockStoredFixture {
+        block_hashes: &[201, 202, 203],
+        token_ids: &[],
+        block_size: 0,
+        parent_block_hash: None,
+    });
+    let warning_count = Arc::new(AtomicU32::new(0));
+    let placement = convert_event(
+        raw,
+        42,
+        16,
+        WorkerWithDpRank::new(7, 0),
+        &warning_count,
+        None,
+    )
+    .unwrap();
+
+    assert_eq!(placement.placement.tier, StorageTier::HostPinned);
+    match placement.event.data {
+        KvCacheEventData::Stored(store_data) => {
+            assert!(store_data.parent_hash.is_none());
+            assert!(store_data.blocks.is_empty());
+        }
+        other => panic!("expected Stored event, got {other:?}"),
+    }
+    assert!(warning_count.load(Ordering::Relaxed) >= 1);
+}
+
+#[test]
+fn cpu_event_with_full_payload_is_indexable() {
+    let raw = cpu_block_stored(CpuBlockStoredFixture {
+        block_hashes: &[201, 202],
+        token_ids: &[10, 11, 12, 13, 14, 15, 16, 17],
+        block_size: 4,
+        parent_block_hash: Some(200),
+    });
+    let warning_count = Arc::new(AtomicU32::new(0));
+    let placement = convert_event(
+        raw,
+        43,
+        4,
+        WorkerWithDpRank::new(7, 0),
+        &warning_count,
+        None,
+    )
+    .unwrap();
+
+    assert_eq!(placement.placement.tier, StorageTier::HostPinned);
+    match placement.event.data {
+        KvCacheEventData::Stored(store_data) => {
+            assert_eq!(store_data.parent_hash, Some(ExternalSequenceBlockHash(200)));
+            assert_eq!(store_data.blocks.len(), 2);
+            assert_eq!(
+                store_data.blocks[0].block_hash,
+                ExternalSequenceBlockHash(201)
+            );
+            assert_eq!(
+                store_data.blocks[1].block_hash,
+                ExternalSequenceBlockHash(202)
+            );
+        }
+        other => panic!("expected Stored event, got {other:?}"),
+    }
+    assert_eq!(warning_count.load(Ordering::Relaxed), 0);
+}

lib/llm/src/kv_router/indexer/mod.rs

@@ -116,12 +116,13 @@ impl Indexer {
                         ConcurrentRadixTreeCompressed::new(),
                         kv_router_config.router_event_threads as usize,
                         block_size,
-                        Some(kv_indexer_metrics),
+                        Some(kv_indexer_metrics.clone()),
                         prune_config,
                     )),
-                    lower_tier: LowerTierIndexers::new(
+                    lower_tier: LowerTierIndexers::new_with_metrics(
                         kv_router_config.router_event_threads as usize,
                         block_size,
+                        Some(kv_indexer_metrics),
                     ),
                     approx: None,
                     primary_records_routing_decisions: true,
@@ -133,10 +134,14 @@ impl Indexer {
                 primary: KvIndexer::new_with_pruning(
                     cancellation_token,
                     block_size,
-                    kv_indexer_metrics,
+                    kv_indexer_metrics.clone(),
                     prune_config,
                 ),
-                lower_tier: LowerTierIndexers::new(1, block_size),
+                lower_tier: LowerTierIndexers::new_with_metrics(
+                    1,
+                    block_size,
+                    Some(kv_indexer_metrics),
+                ),
                 approx: None,
                 primary_records_routing_decisions: true,
             });
@@ -151,11 +156,12 @@ impl Indexer {
                     ConcurrentRadixTreeCompressed::new(),
                     kv_router_config.router_event_threads as usize,
                     block_size,
-                    Some(kv_indexer_metrics),
+                    Some(kv_indexer_metrics.clone()),
                 )),
-                lower_tier: LowerTierIndexers::new(
+                lower_tier: LowerTierIndexers::new_with_metrics(
                     kv_router_config.router_event_threads as usize,
                     block_size,
+                    Some(kv_indexer_metrics),
                 ),
                 approx,
                 primary_records_routing_decisions: false,
@@ -169,10 +175,14 @@ impl Indexer {
             primary: KvIndexer::new_with_pruning(
                 cancellation_token,
                 block_size,
-                kv_indexer_metrics,
+                kv_indexer_metrics.clone(),
                 None,
             ),
-            lower_tier: LowerTierIndexers::new(1, block_size),
+            lower_tier: LowerTierIndexers::new_with_metrics(
+                1,
+                block_size,
+                Some(kv_indexer_metrics),
+            ),
             approx,
             primary_records_routing_decisions: false,
         })