[Router] Add bucket-aware policy domains and native cache indexing (#38108)

Signed-off-by: Vincent Gao <vincentbo@linux.alibaba.com> Co-authored-by: inkcherry <mingzhi.liu@amd.com> Co-authored-by: yangbodong22011 <13137470+yangbodong22011@users.noreply.github.com>
2026-09-06 19:47:51 +08:00
parent a176ba2f7b
commit 5bebe7a033
76 changed files with 5842 additions and 3639 deletions
@@ -32,4 +32,5 @@ tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
 zeromq = { version = "0.6", default-features = false, features = ["tokio-runtime", "tcp-transport"] }

 [build-dependencies]
+protoc-bin-vendored = "3.2.0"
 tonic-prost-build = "0.14.6"
@@ -133,7 +133,8 @@ sgl-router \
  --model-id <model-id> \
  --tokenizer-path <huggingface-repo-or-tokenizer> \
  --worker-urls http://127.0.0.1:30000 \
-  --policy cache_aware_zmq \
+  --policy cache_aware \
+  --cache-prefix-provider indexer \
  --kv-indexer-endpoint http://127.0.0.1:50051 \
  --kv-indexer-query-timeout-ms 100 \
  --kv-indexer-query-max-inflight 32
@@ -2,9 +2,14 @@
 // SPDX-License-Identifier: Apache-2.0

 fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Parent hashes need proto3 presence semantics to distinguish roots from
+    // valid hash values. Use the bundled compiler to keep codegen stable.
+    let mut config = tonic_prost_build::Config::new();
+    config.protoc_executable(protoc_bin_vendored::protoc_bin_path()?);
+
    tonic_prost_build::configure()
        .build_client(true)
        .build_server(true)
-        .compile_protos(&["proto/kv_indexer.proto"], &["proto"])?;
+        .compile_with_config(config, &["proto/kv_indexer.proto"], &["proto"])?;
    Ok(())
 }
@@ -71,6 +71,11 @@ message ExternalKvAction {
  // REPORT only. Per-hash token count (block_size), index-aligned with `hashes`,
  // used to accumulate SWA trailing windows. Empty when not supplied (legacy).
  repeated uint32 block_sizes = 5;
+
+  // REPORT only. Parent of hashes[0]; absent means hashes[0] is a root block.
+  // Every later hash is the child of the preceding hash. The Indexer and Bridge
+  // are deployed together, so this new protocol does not support old senders.
+  optional sfixed64 parent_block_hash = 6;
 }

 message ApplyExternalKvBatchRequest {
@@ -6,13 +6,15 @@ use std::sync::Arc;
 use std::{env, io};

 use sgl_kv_indexer::{
-    server_builder, shutdown_signal, stamp_arrival, InMemoryKvIndexerBackend, KvIndexerBackend,
-    KvIndexerService, DEFAULT_PREFIX_QUERY_MAX_INFLIGHT, MAX_CONCURRENT_STREAMS,
+    server_builder_with_max_concurrent_streams, shutdown_signal, stamp_arrival,
+    InMemoryKvIndexerBackend, KvIndexerBackend, KvIndexerService,
+    DEFAULT_PREFIX_QUERY_MAX_INFLIGHT, MAX_CONCURRENT_STREAMS,
 };
 use tonic::service::interceptor::InterceptedService;
 use tracing::info;

 const PREFIX_QUERY_MAX_INFLIGHT_ENV: &str = "KV_INDEXER_PREFIX_QUERY_MAX_INFLIGHT";
+const MAX_CONCURRENT_STREAMS_ENV: &str = "KV_INDEXER_MAX_CONCURRENT_STREAMS";

 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
@@ -26,6 +28,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        .unwrap_or_else(|_| "[::1]:50051".to_string())
        .parse::<SocketAddr>()?;
    let prefix_query_max_inflight = prefix_query_max_inflight_from_env()?;
+    let max_concurrent_streams = max_concurrent_streams_from_env()?;

    let backend: Arc<dyn KvIndexerBackend> = Arc::new(InMemoryKvIndexerBackend::new());
    // The interceptor timestamps each request before its own task is queued,
@@ -39,10 +42,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    info!(
        %addr,
        prefix_query_max_inflight,
-        max_concurrent_streams = MAX_CONCURRENT_STREAMS,
+        max_concurrent_streams,
        "starting single-server in-memory SGLang KV Indexer"
    );
-    server_builder()
+    server_builder_with_max_concurrent_streams(max_concurrent_streams)
        .add_service(service)
        .serve_with_shutdown(addr, shutdown_signal())
        .await?;
@@ -77,6 +80,33 @@ fn parse_prefix_query_max_inflight(raw: &str) -> io::Result<usize> {
    Ok(value)
 }

+fn max_concurrent_streams_from_env() -> io::Result<u32> {
+    match env::var(MAX_CONCURRENT_STREAMS_ENV) {
+        Ok(raw) => parse_max_concurrent_streams(&raw),
+        Err(env::VarError::NotPresent) => Ok(MAX_CONCURRENT_STREAMS),
+        Err(env::VarError::NotUnicode(_)) => Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            format!("{MAX_CONCURRENT_STREAMS_ENV} must be valid UTF-8"),
+        )),
+    }
+}
+
+fn parse_max_concurrent_streams(raw: &str) -> io::Result<u32> {
+    let value = raw.parse::<u32>().map_err(|_| {
+        io::Error::new(
+            io::ErrorKind::InvalidInput,
+            format!("{MAX_CONCURRENT_STREAMS_ENV} must be a positive integer, got {raw:?}"),
+        )
+    })?;
+    if value == 0 {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            format!("{MAX_CONCURRENT_STREAMS_ENV} must be greater than zero"),
+        ));
+    }
+    Ok(value)
+}
+
 #[cfg(test)]
 mod tests {
    use super::*;
@@ -91,4 +121,16 @@ mod tests {
        assert!(parse_prefix_query_max_inflight("0").is_err());
        assert!(parse_prefix_query_max_inflight("many").is_err());
    }
+
+    #[test]
+    fn parses_positive_stream_limit() {
+        assert_eq!(parse_max_concurrent_streams("512").unwrap(), 512);
+    }
+
+    #[test]
+    fn rejects_invalid_stream_limit() {
+        assert!(parse_max_concurrent_streams("0").is_err());
+        assert!(parse_max_concurrent_streams("many").is_err());
+        assert!(parse_max_concurrent_streams("4294967296").is_err());
+    }
 }
@@ -150,6 +150,7 @@ fn classify_rpc(status: Status) -> BridgeError {
 enum Action {
    Report {
        tier: i32,
+        parent_block_hash: Option<i64>,
        hashes: Vec<i64>,
        masks: Vec<Option<u32>>,
        block_sizes: Vec<Option<u32>>,
@@ -171,19 +172,27 @@ impl EventActions {
    /// with an immediately-preceding store to the same tier and never across a
    /// revoke/clear, so the final per-hash state is preserved. All hashes here
    /// share the event's component mask and block size.
-    fn report(&mut self, tier: i32, hashes: Vec<i64>, mask: Option<u32>, block_size: Option<u32>) {
+    fn report(
+        &mut self,
+        tier: i32,
+        parent_block_hash: Option<i64>,
+        hashes: Vec<i64>,
+        mask: Option<u32>,
+        block_size: Option<u32>,
+    ) {
        if hashes.is_empty() {
            return;
        }
        let n = hashes.len();
        if let Some(Action::Report {
            tier: last_tier,
+            parent_block_hash: _,
            hashes: last,
            masks,
            block_sizes,
        }) = self.actions.last_mut()
        {
-            if *last_tier == tier {
+            if *last_tier == tier && parent_block_hash == last.last().copied() {
                last.extend(hashes);
                masks.extend(std::iter::repeat_n(mask, n));
                block_sizes.extend(std::iter::repeat_n(block_size, n));
@@ -192,6 +201,7 @@ impl EventActions {
        }
        self.actions.push(Action::Report {
            tier,
+            parent_block_hash,
            hashes,
            masks: vec![mask; n],
            block_sizes: vec![block_size; n],
@@ -401,6 +411,7 @@ fn build_apply_request(
        match action {
            Action::Report {
                tier,
+                parent_block_hash,
                hashes,
                masks,
                block_sizes,
@@ -413,6 +424,7 @@ fn build_apply_request(
                // the backend keeps the whole-block fast path.
                component_masks: encode_component_masks(&masks),
                block_sizes: encode_block_sizes(&block_sizes),
+                parent_block_hash,
            }),
            Action::Revoke { tier, hashes } => actions.push(ExternalKvAction {
                r#type: ExternalKvActionType::ActionRevoke as i32,
@@ -420,6 +432,7 @@ fn build_apply_request(
                hashes,
                component_masks: Vec::new(),
                block_sizes: Vec::new(),
+                parent_block_hash: None,
            }),
            Action::ClearAll => {
                for tier in &config.clear_tiers {
@@ -429,6 +442,7 @@ fn build_apply_request(
                        hashes: Vec::new(),
                        component_masks: Vec::new(),
                        block_sizes: Vec::new(),
+                        parent_block_hash: None,
                    });
                }
            }
@@ -499,6 +513,15 @@ fn split_action(action: ExternalKvAction) -> Vec<ExternalKvAction> {
                hashes: action.hashes[start..end].to_vec(),
                component_masks: slice_or_empty(&action.component_masks, start, end),
                block_sizes: slice_or_empty(&action.block_sizes, start, end),
+                parent_block_hash: if action.r#type == ExternalKvActionType::ActionReport as i32 {
+                    if start == 0 {
+                        action.parent_block_hash
+                    } else {
+                        Some(action.hashes[start - 1])
+                    }
+                } else {
+                    None
+                },
            }
        })
        .collect()
@@ -599,7 +622,13 @@ fn decode_event(event: &Value, actions: &mut EventActions) -> Result<(), BridgeE
                Some(_) => Some(decode_block_size(&event[4])?),
                None => None,
            };
-            actions.report(tier, decode_hashes(&event[1])?, mask, block_size);
+            actions.report(
+                tier,
+                decode_optional_hash(&event[2], "BlockStored.parent_block_hash")?,
+                decode_hashes(&event[1])?,
+                mask,
+                block_size,
+            );
        }
        "BlockRemoved" => {
            if event.len() < 3 {
@@ -623,26 +652,29 @@ fn decode_event(event: &Value, actions: &mut EventActions) -> Result<(), BridgeE
 fn decode_hashes(value: &Value) -> Result<Vec<i64>, BridgeError> {
    expect_array(value, "block_hashes")?
        .iter()
-        .map(|value| {
-            if let Some(value) = value.as_i64() {
-                return Ok(value);
-            }
-            // SGLang folds the unsigned top 64 bits of the SHA-256 into the
-            // signed range by subtracting 2^64 (`hash_str_to_int64`), which is
-            // two's complement, so a producer that serialises the unsigned half
-            // instead is carrying identical bits. Reinterpreting recovers the
-            // hash the router queries for; refusing the value would instead skip
-            // the whole event and lose every placement it carried.
-            if let Some(value) = value.as_u64() {
-                return Ok(value as i64);
-            }
-            Err(BridgeError::Decode(
-                "block hash must be an integer".to_string(),
-            ))
-        })
+        .map(|value| decode_hash(value, "block hash"))
        .collect()
 }

+fn decode_hash(value: &Value, field: &str) -> Result<i64, BridgeError> {
+    if let Some(value) = value.as_i64() {
+        return Ok(value);
+    }
+    // SGLang folds the unsigned top 64 bits of the SHA-256 into the signed
+    // range by subtracting 2^64. Reinterpreting recovers the same bits.
+    if let Some(value) = value.as_u64() {
+        return Ok(value as i64);
+    }
+    Err(BridgeError::Decode(format!("{field} must be an integer")))
+}
+
+fn decode_optional_hash(value: &Value, field: &str) -> Result<Option<i64>, BridgeError> {
+    if matches!(value, Value::Nil) {
+        return Ok(None);
+    }
+    decode_hash(value, field).map(Some)
+}
+
 /// Decodes the optional `component_types` slot of a `BlockStored` into a component
 /// bitmask. `nil` maps to `None`, a legacy whole-block store; an array of labels
 /// folds into a bitmask, and labels this build does not model are ignored.
@@ -836,10 +868,14 @@ mod tests {
    }

    fn stored(hashes: &[i64], medium: &str) -> Value {
+        stored_with_parent(hashes, None, medium)
+    }
+
+    fn stored_with_parent(hashes: &[i64], parent: Option<i64>, medium: &str) -> Value {
        Value::Array(vec![
            Value::String("BlockStored".into()),
            ints(hashes),
-            Value::Nil,         // parent_block_hash
+            parent.map_or(Value::Nil, Value::from),
            ints(&[1]),         // token_ids
            Value::from(1_i64), // block_size
            Value::Nil,         // lora_id
@@ -850,10 +886,20 @@ mod tests {
    /// A component-aware `BlockStored` (8-element schema): trailing
    /// `component_types` slot plus a concrete `block_size` token count.
    fn stored_c(hashes: &[i64], medium: &str, block_size: i64, components: Value) -> Value {
+        stored_c_with_parent(hashes, None, medium, block_size, components)
+    }
+
+    fn stored_c_with_parent(
+        hashes: &[i64],
+        parent: Option<i64>,
+        medium: &str,
+        block_size: i64,
+        components: Value,
+    ) -> Value {
        Value::Array(vec![
            Value::String("BlockStored".into()),
            ints(hashes),
-            Value::Nil, // parent_block_hash
+            parent.map_or(Value::Nil, Value::from),
            ints(&[1]), // token_ids
            Value::from(block_size),
            Value::Nil, // lora_id
@@ -868,8 +914,13 @@ mod tests {

    /// Legacy (whole-block) report action expectation.
    fn rep(tier: i32, hashes: &[&str]) -> Action {
+        rep_with_parent(tier, None, hashes)
+    }
+
+    fn rep_with_parent(tier: i32, parent_block_hash: Option<i64>, hashes: &[&str]) -> Action {
        Action::Report {
            tier,
+            parent_block_hash,
            hashes: hashes.iter().map(|h| h.parse().unwrap()).collect(),
            masks: vec![None; hashes.len()],
            block_sizes: vec![None; hashes.len()],
@@ -948,6 +999,7 @@ mod tests {
            hashes: hashes.iter().map(|h| h.parse().unwrap()).collect(),
            component_masks: Vec::new(),
            block_sizes: Vec::new(),
+            parent_block_hash: None,
        }
    }

@@ -958,6 +1010,7 @@ mod tests {
            hashes: hashes.iter().map(|h| h.parse().unwrap()).collect(),
            component_masks: Vec::new(),
            block_sizes: Vec::new(),
+            parent_block_hash: None,
        }
    }

@@ -968,6 +1021,7 @@ mod tests {
            hashes: Vec::new(),
            component_masks: Vec::new(),
            block_sizes: Vec::new(),
+            parent_block_hash: None,
        }
    }

@@ -986,6 +1040,19 @@ mod tests {
        assert_eq!(request.worker_address, "127.0.0.1:9000");
    }

+    #[test]
+    fn request_carries_parent_block_hash() {
+        let config = test_config(vec![hbm()]);
+        let request = request_of(
+            &config,
+            0,
+            vec![stored_with_parent(&[2, 3], Some(1), "GPU")],
+        );
+        assert_eq!(request.actions.len(), 1);
+        assert_eq!(request.actions[0].parent_block_hash, Some(1));
+        assert_eq!(request.actions[0].hashes, vec![2, 3]);
+    }
+
    #[test]
    fn oversized_report_is_split_with_aligned_metadata() {
        let count = MAX_HASHES_PER_REQUEST + 1;
@@ -998,6 +1065,7 @@ mod tests {
                hashes: (0..count).map(|index| index as i64).collect(),
                component_masks: (0..count as u32).collect(),
                block_sizes: (0..count as u32).map(|index| index + 1).collect(),
+                parent_block_hash: None,
            }],
            worker_address: "http://worker-1".into(),
            cache_spec: None,
@@ -1007,6 +1075,7 @@ mod tests {

        assert_eq!(batches.len(), 2);
        assert_eq!(batches[0].actions[0].hashes.len(), MAX_HASHES_PER_REQUEST);
+        assert_eq!(batches[0].actions[0].parent_block_hash, None);
        assert_eq!(
            batches[1].actions[0].hashes,
            vec![MAX_HASHES_PER_REQUEST as i64]
@@ -1015,6 +1084,10 @@ mod tests {
            batches[1].actions[0].component_masks,
            vec![MAX_HASHES_PER_REQUEST as u32]
        );
+        assert_eq!(
+            batches[1].actions[0].parent_block_hash,
+            Some(MAX_HASHES_PER_REQUEST as i64 - 1)
+        );
        assert_eq!(
            batches[1].actions[0].block_sizes,
            vec![MAX_HASHES_PER_REQUEST as u32 + 1]
@@ -1167,11 +1240,22 @@ mod tests {
    #[test]
    fn adjacent_same_tier_stores_coalesce() {
        assert_eq!(
-            actions_of(vec![stored(&[1], "GPU"), stored(&[2], "GPU")]),
+            actions_of(vec![
+                stored(&[1], "GPU"),
+                stored_with_parent(&[2], Some(1), "GPU")
+            ]),
            vec![rep(hbm(), &["1", "2"])]
        );
    }

+    #[test]
+    fn same_tier_stores_on_different_chains_do_not_coalesce() {
+        assert_eq!(
+            actions_of(vec![stored(&[1], "GPU"), stored(&[2], "GPU")]),
+            vec![rep(hbm(), &["1"]), rep(hbm(), &["2"])]
+        );
+    }
+
    #[test]
    fn different_tier_stores_do_not_coalesce() {
        assert_eq!(
@@ -1219,7 +1303,7 @@ mod tests {
        assert_eq!(
            decode_event_batch(&payload).unwrap().actions,
            vec![
-                rep(hbm(), &["1234567890123", "-987654321"]),
+                rep_with_parent(hbm(), Some(42), &["1234567890123", "-987654321"]),
                rev(ssd(), &["100", "200"]),
                Action::ClearAll,
            ]
@@ -1352,6 +1436,7 @@ mod tests {
            actions_of(vec![stored_c(&[1], "GPU", 64, strv(&["full", "swa"]))]),
            vec![Action::Report {
                tier: hbm(),
+                parent_block_hash: None,
                hashes: vec![1],
                masks: vec![Some(
                    crate::service::COMPONENT_FULL | crate::service::COMPONENT_SWA
@@ -1379,7 +1464,7 @@ mod tests {
            0,
            vec![
                stored_c(&[1], "GPU", 64, strv(&["full", "swa"])),
-                stored_c(&[2], "GPU", 32, strv(&["full"])),
+                stored_c_with_parent(&[2], Some(1), "GPU", 32, strv(&["full"])),
            ],
        );
        assert_eq!(request.actions.len(), 1);
@@ -24,9 +24,10 @@ pub use client::{
 };
 pub use memory_backend::InMemoryKvIndexerBackend;
 pub use service::{
-    component_bit, server_builder, BlockComponents, KvIndexerBackend, KvIndexerService,
-    WorkerPrefixInput, COMPONENT_FULL, COMPONENT_MAMBA, COMPONENT_SWA,
-    DEFAULT_PREFIX_QUERY_MAX_INFLIGHT, MAX_CONCURRENT_STREAMS, MAX_GRPC_DECODING_MESSAGE_SIZE,
+    component_bit, server_builder, server_builder_with_max_concurrent_streams, BlockComponents,
+    KvIndexerBackend, KvIndexerService, WorkerPrefixInput, COMPONENT_FULL, COMPONENT_MAMBA,
+    COMPONENT_SWA, DEFAULT_PREFIX_QUERY_MAX_INFLIGHT, MAX_CONCURRENT_STREAMS,
+    MAX_GRPC_DECODING_MESSAGE_SIZE,
 };
 pub use shutdown::shutdown_signal;
 /// Re-exported because [`PrefixIndexError::Rejected`] carries it, so callers can
@@ -7,26 +7,42 @@
 //! atomic and every query a consistent snapshot. The state is soft: not shared
 //! with another server, and lost when the process exits.

-use std::collections::{BTreeMap, HashMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
 use std::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};

 use tonic::Status;

 use crate::pb::{
-    ApplyExternalKvBatchRequest, ApplyExternalKvBatchResponse, ExternalKvActionType,
-    ExternalKvNodeMatch, GetExternalKvHitCountsRequest, GetExternalKvHitCountsResponse,
-    HitCountEntry, MatchExternalKvPrefixRequest, MatchExternalKvPrefixResponse,
-    MatchExternalKvRequest, MatchExternalKvResponse, TierHashes, WorkerCacheSpec,
+    ApplyExternalKvBatchRequest, ApplyExternalKvBatchResponse, ExternalKvAction,
+    ExternalKvActionType, ExternalKvNodeMatch, GetExternalKvHitCountsRequest,
+    GetExternalKvHitCountsResponse, HitCountEntry, MatchExternalKvPrefixRequest,
+    MatchExternalKvPrefixResponse, MatchExternalKvRequest, MatchExternalKvResponse, TierHashes,
+    TierType, WorkerCacheSpec,
 };
-use crate::service::{assemble_prefix_response, prefix_limit, WorkerPrefixScanner};
+use crate::service::{assemble_prefix_response, prefix_limit, WorkerPrefixScanner, COMPONENT_FULL};
 use crate::{BlockComponents, KvIndexerBackend, WorkerPrefixInput};

+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+enum ParentLink {
+    #[default]
+    Unknown,
+    Root,
+    Hash(i64),
+}
+
 #[derive(Debug, Default)]
 struct BlockRecord {
+    /// Chain relationship reported by the worker. Prefix-derived state is valid
+    /// only along links rooted at `Root`.
+    parent: ParentLink,
+    children: HashSet<i64>,
    /// Shared block token count. A zero value means legacy/unspecified.
    token_count: u32,
    /// Resident component snapshot for each `(worker, tier)`.
    placements: HashMap<(String, i32), u32>,
+    /// Workers for which the root-to-this-block prefix is complete and this
+    /// boundary is servable by the Legacy/FULL-only fast path.
+    prefix_complete_workers: HashSet<String>,
 }

 #[derive(Debug, Default)]
@@ -35,6 +51,9 @@ struct WorkerRecord {
    spec: Option<WorkerCacheSpec>,
    /// Reverse index used by CLEAR_ALL_AT_TIER.
    holdings: HashMap<i32, HashSet<i64>>,
+    /// Number of non-legacy component placements. A spec-less worker can use
+    /// the derived fast path only while this is zero.
+    component_placement_count: usize,
 }

 #[derive(Debug, Default)]
@@ -59,6 +78,12 @@ struct PrefixCandidate {
    scanner: WorkerPrefixScanner,
 }

+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum FastPathKind {
+    Legacy,
+    FullOnly { tier_mask: u32 },
+}
+
 /// Single-process, soft-state KV placement index.
 #[derive(Debug, Default)]
 pub struct InMemoryKvIndexerBackend {
@@ -89,6 +114,10 @@ impl InMemoryKvIndexerBackend {
        let mut state = self.write_state()?;
        let worker_id = req.worker_id;

+        validate_actions(&state, &req.actions)?;
+
+        let previous_fast_path = state.workers.get(&worker_id).and_then(fast_path_kind);
+
        // Address and spec are snapshots carried on every batch. Empty address
        // makes the worker unroutable; absent spec returns it to legacy mode.
        {
@@ -97,16 +126,21 @@ impl InMemoryKvIndexerBackend {
            worker.spec = req.cache_spec;
        }

+        let mut dirty_roots = Vec::new();
+        let mut reported_chains = Vec::new();
+        let mut revoked_hashes = Vec::new();
+        // Only a fast-path worker identity change requires a full recompute.
+        // REPORT, REVOKE, and CLEAR enqueue their affected hashes directly.
+        let mut recompute_from_graph_roots = false;
        for action in req.actions {
            match ExternalKvActionType::try_from(action.r#type) {
                Ok(ExternalKvActionType::ActionReport) => {
                    let has_masks = !action.component_masks.is_empty();
                    let has_sizes = !action.block_sizes.is_empty();
+                    let hashes = action.hashes;
+                    apply_report_chain(&mut state, action.parent_block_hash, &hashes);

-                    // REPORT is a REPLACE snapshot. Keep the final occurrence
-                    // when a coalesced action repeats one hash.
-                    let mut last_by_hash: HashMap<i64, (u32, u32)> = HashMap::new();
-                    for (index, hash) in action.hashes.into_iter().enumerate() {
+                    for (index, hash) in hashes.iter().copied().enumerate() {
                        let mask = if has_masks {
                            action.component_masks[index]
                        } else {
@@ -117,18 +151,22 @@ impl InMemoryKvIndexerBackend {
                        } else {
                            0
                        };
-                        last_by_hash.insert(hash, (mask, token_count));
-                    }
-
-                    for (hash, (mask, token_count)) in last_by_hash {
-                        let block = state.blocks.entry(hash).or_default();
-                        block
+                        let old_mask = state
+                            .blocks
+                            .entry(hash)
+                            .or_default()
                            .placements
                            .insert((worker_id.clone(), action.tier), mask);
+                        adjust_component_placement_count(
+                            &mut state,
+                            &worker_id,
+                            old_mask,
+                            Some(mask),
+                        );
                        // A legacy report carries no size, so 0 means
                        // "unknown" and must not erase a known count.
                        if token_count > 0 {
-                            block.token_count = token_count;
+                            state.blocks.entry(hash).or_default().token_count = token_count;
                        }
                        state
                            .workers
@@ -139,10 +177,15 @@ impl InMemoryKvIndexerBackend {
                            .or_default()
                            .insert(hash);
                    }
+                    if !hashes.is_empty() {
+                        reported_chains.push(hashes);
+                    }
                }
                Ok(ExternalKvActionType::ActionRevoke) => {
                    for hash in action.hashes {
                        revoke_one(&mut state, &worker_id, &hash, action.tier);
+                        dirty_roots.push(hash);
+                        revoked_hashes.push(hash);
                    }
                }
                Ok(ExternalKvActionType::ActionClearAllAtTier) => {
@@ -154,6 +197,8 @@ impl InMemoryKvIndexerBackend {
                        .unwrap_or_default();
                    for hash in hashes {
                        revoke_one(&mut state, &worker_id, &hash, action.tier);
+                        dirty_roots.push(hash);
+                        revoked_hashes.push(hash);
                    }
                }
                Ok(ExternalKvActionType::ActionUnknown) | Err(_) => {
@@ -162,6 +207,31 @@ impl InMemoryKvIndexerBackend {
            }
        }

+        let current_fast_path = state.workers.get(&worker_id).and_then(fast_path_kind);
+        if previous_fast_path != current_fast_path {
+            for block in state.blocks.values_mut() {
+                block.prefix_complete_workers.remove(&worker_id);
+            }
+            recompute_from_graph_roots = true;
+        }
+        if recompute_from_graph_roots {
+            dirty_roots = state
+                .blocks
+                .iter()
+                .filter_map(|(hash, block)| (block.parent == ParentLink::Root).then_some(*hash))
+                .collect();
+        } else {
+            for hashes in reported_chains {
+                dirty_roots.extend(refresh_linear_report_chain_prefix_completeness(
+                    &mut state, &worker_id, &hashes,
+                ));
+            }
+        }
+        recompute_worker_subtrees(&mut state, &worker_id, dirty_roots);
+        for hash in revoked_hashes {
+            prune_empty_leaf(&mut state, hash);
+        }
+
        Ok(ApplyExternalKvBatchResponse {})
    }

@@ -302,11 +372,28 @@ impl InMemoryKvIndexerBackend {
                blocks_read: 1,
            });
        };
+        let known_prefix_len = known_request_prefix_len(&state, hashes).unwrap_or(0);
+        let fast_worker_ids: HashSet<&str> = if known_prefix_len > 0 {
+            first
+                .prefix_complete_workers
+                .iter()
+                .filter_map(|worker_id| {
+                    state.workers.get(worker_id).and_then(|worker| {
+                        (fast_path_kind(worker).is_some() && !worker.address.is_empty())
+                            .then_some(worker_id.as_str())
+                    })
+                })
+                .collect()
+        } else {
+            HashSet::new()
+        };
        let mut seen = HashSet::new();
        let mut candidates: Vec<PrefixCandidate> = first
            .placements
            .keys()
-            .filter(|(worker, _)| seen.insert(worker.as_str()))
+            .filter(|(worker, _)| {
+                !fast_worker_ids.contains(worker.as_str()) && seen.insert(worker.as_str())
+            })
            .map(|(worker, _)| {
                let metadata = state.workers.get(worker);
                PrefixCandidate {
@@ -332,41 +419,61 @@ impl InMemoryKvIndexerBackend {
                tier_masks: Vec::new(),
            })
            .collect();
-
-        for hash in hashes {
-            present.fill(false);
-            for block in &mut block_views {
-                block.token_count = 0;
-                block.tier_masks.clear();
+        let mut entries = Vec::with_capacity(fast_worker_ids.len() + candidates.len());
+        let mut unresolved = fast_worker_ids;
+        for (index, hash) in hashes[..known_prefix_len].iter().enumerate().rev() {
+            if unresolved.is_empty() {
+                break;
            }
            if let Some(block) = state.blocks.get(hash) {
-                for ((worker, tier), mask) in &block.placements {
-                    let Some(&index) = candidate_by_id.get(worker) else {
+                for worker_id in &block.prefix_complete_workers {
+                    if !unresolved.remove(worker_id.as_str()) {
                        continue;
-                    };
-                    present[index] = true;
-                    block_views[index].token_count = block.token_count;
-                    block_views[index].tier_masks.push((*tier, *mask));
+                    }
+                    if let Some(worker) = state.workers.get(worker_id) {
+                        entries.push((
+                            worker_id.clone(),
+                            worker.address.clone(),
+                            (index + 1) as u32,
+                        ));
+                    }
                }
            }
-            for (index, candidate) in candidates.iter_mut().enumerate() {
-                candidate
-                    .scanner
-                    .push(present[index].then_some(&block_views[index]));
-            }
        }

-        let entries = candidates
-            .into_iter()
-            .filter_map(|candidate| {
-                let prefix = candidate.scanner.prefix();
-                (!candidate.address.is_empty() && prefix > 0).then_some((
-                    candidate.worker_id,
-                    candidate.address,
-                    prefix,
-                ))
-            })
-            .collect();
+        if !candidates.is_empty() {
+            for hash in hashes {
+                present.fill(false);
+                for block in &mut block_views {
+                    block.token_count = 0;
+                    block.tier_masks.clear();
+                }
+                if let Some(block) = state.blocks.get(hash) {
+                    for ((worker, tier), mask) in &block.placements {
+                        let Some(&index) = candidate_by_id.get(worker) else {
+                            continue;
+                        };
+                        present[index] = true;
+                        block_views[index].token_count = block.token_count;
+                        block_views[index].tier_masks.push((*tier, *mask));
+                    }
+                }
+                for (index, candidate) in candidates.iter_mut().enumerate() {
+                    candidate
+                        .scanner
+                        .push(present[index].then_some(&block_views[index]));
+                }
+            }
+        }
+
+        entries.extend(candidates.into_iter().filter_map(|candidate| {
+            let prefix = candidate.scanner.prefix();
+            (!candidate.address.is_empty() && prefix > 0).then_some((
+                candidate.worker_id,
+                candidate.address,
+                prefix,
+            ))
+        }));
        Ok(assemble_prefix_response(entries, limit as u32))
    }

@@ -392,12 +499,319 @@ impl InMemoryKvIndexerBackend {
    }
 }

-fn revoke_one(state: &mut State, worker_id: &str, hash: &i64, tier: i32) {
-    let mut remove_block = false;
-    if let Some(block) = state.blocks.get_mut(hash) {
-        block.placements.remove(&(worker_id.to_string(), tier));
-        remove_block = block.placements.is_empty();
+fn fast_path_kind(worker: &WorkerRecord) -> Option<FastPathKind> {
+    match worker.spec.as_ref() {
+        None if worker.component_placement_count == 0 => Some(FastPathKind::Legacy),
+        Some(spec) if spec.version <= 1 && spec.components == COMPONENT_FULL => {
+            Some(FastPathKind::FullOnly {
+                tier_mask: spec.full_tier_mask,
+            })
+        }
+        _ => None,
    }
+}
+
+fn tier_in_mask(mask: u32, tier: i32) -> bool {
+    tier >= 0 && mask & (1u32 << tier) != 0
+}
+
+fn block_servable(state: &State, hash: i64, worker_id: &str, kind: FastPathKind) -> bool {
+    match kind {
+        // A globally component-free legacy worker only needs membership, which
+        // its reverse holdings index answers without scanning every other
+        // worker placed on this popular block.
+        FastPathKind::Legacy => state.workers.get(worker_id).is_some_and(|worker| {
+            worker
+                .holdings
+                .values()
+                .any(|hashes| hashes.contains(&hash))
+        }),
+        FastPathKind::FullOnly { tier_mask } => {
+            let indexer_tiers =
+                (1 << (TierType::TierHbm as u32)) | (1 << (TierType::TierDram as u32));
+            state.blocks.get(&hash).is_some_and(|block| {
+                block.placements.iter().any(|((worker, tier), mask)| {
+                    worker == worker_id
+                        && mask & COMPONENT_FULL != 0
+                        && tier_in_mask(indexer_tiers & tier_mask, *tier)
+                })
+            })
+        }
+    }
+}
+
+#[cfg(test)]
+fn link_report_chain(
+    state: &mut State,
+    parent_block_hash: Option<i64>,
+    hashes: &[i64],
+) -> Result<(), Status> {
+    let mut planned_parents = HashMap::with_capacity(hashes.len());
+    validate_report_chain(state, &mut planned_parents, parent_block_hash, hashes)?;
+    validate_parent_graph_acyclic(state, &planned_parents)?;
+
+    apply_report_chain(state, parent_block_hash, hashes);
+    Ok(())
+}
+
+fn apply_report_chain(state: &mut State, parent_block_hash: Option<i64>, hashes: &[i64]) {
+    let mut parent = parent_block_hash.map_or(ParentLink::Root, ParentLink::Hash);
+    for hash in hashes {
+        if let ParentLink::Hash(parent_hash) = parent {
+            state
+                .blocks
+                .entry(parent_hash)
+                .or_default()
+                .children
+                .insert(*hash);
+        }
+        state.blocks.entry(*hash).or_default().parent = parent;
+        parent = ParentLink::Hash(*hash);
+    }
+}
+
+fn validate_actions(state: &State, actions: &[ExternalKvAction]) -> Result<(), Status> {
+    let mut planned_parents = HashMap::new();
+    for action in actions {
+        match ExternalKvActionType::try_from(action.r#type) {
+            Ok(ExternalKvActionType::ActionReport) => validate_report_chain(
+                state,
+                &mut planned_parents,
+                action.parent_block_hash,
+                &action.hashes,
+            )?,
+            Ok(ExternalKvActionType::ActionRevoke)
+            | Ok(ExternalKvActionType::ActionClearAllAtTier) => {}
+            Ok(ExternalKvActionType::ActionUnknown) | Err(_) => {
+                return Err(Status::invalid_argument("unsupported action type"));
+            }
+        }
+    }
+    validate_parent_graph_acyclic(state, &planned_parents)
+}
+
+fn validate_report_chain(
+    state: &State,
+    planned_parents: &mut HashMap<i64, ParentLink>,
+    parent_block_hash: Option<i64>,
+    hashes: &[i64],
+) -> Result<(), Status> {
+    let mut parent = parent_block_hash.map_or(ParentLink::Root, ParentLink::Hash);
+    for hash in hashes {
+        if parent == ParentLink::Hash(*hash) {
+            return Err(Status::invalid_argument(
+                "block hash cannot be its own parent",
+            ));
+        }
+        let existing = planned_parents
+            .get(hash)
+            .copied()
+            .or_else(|| state.blocks.get(hash).map(|block| block.parent))
+            .unwrap_or_default();
+        if existing != ParentLink::Unknown && existing != parent {
+            return Err(Status::invalid_argument(format!(
+                "block hash {hash} was reported with conflicting parents"
+            )));
+        }
+        planned_parents.insert(*hash, parent);
+        parent = ParentLink::Hash(*hash);
+    }
+    Ok(())
+}
+
+fn validate_parent_graph_acyclic(
+    state: &State,
+    planned_parents: &HashMap<i64, ParentLink>,
+) -> Result<(), Status> {
+    let mut complete = HashSet::new();
+    for start in planned_parents.keys().copied() {
+        if complete.contains(&start) {
+            continue;
+        }
+
+        let mut path = Vec::new();
+        let mut on_path = HashSet::new();
+        let mut current = start;
+        loop {
+            if complete.contains(&current) {
+                break;
+            }
+            if !on_path.insert(current) {
+                return Err(Status::invalid_argument(
+                    "report would create a parent cycle",
+                ));
+            }
+            path.push(current);
+
+            let parent = planned_parents
+                .get(&current)
+                .copied()
+                .or_else(|| state.blocks.get(&current).map(|block| block.parent))
+                .unwrap_or_default();
+            match parent {
+                ParentLink::Hash(parent) => current = parent,
+                ParentLink::Unknown | ParentLink::Root => break,
+            }
+        }
+        complete.extend(path);
+    }
+    Ok(())
+}
+
+fn adjust_component_placement_count(
+    state: &mut State,
+    worker_id: &str,
+    old_mask: Option<u32>,
+    new_mask: Option<u32>,
+) {
+    let worker = state.workers.entry(worker_id.to_string()).or_default();
+    if old_mask.is_some_and(|mask| mask != 0) {
+        worker.component_placement_count = worker.component_placement_count.saturating_sub(1);
+    }
+    if new_mask.is_some_and(|mask| mask != 0) {
+        worker.component_placement_count = worker.component_placement_count.saturating_add(1);
+    }
+}
+
+fn recompute_worker_subtrees(
+    state: &mut State,
+    worker_id: &str,
+    roots: impl IntoIterator<Item = i64>,
+) {
+    let kind = state.workers.get(worker_id).and_then(fast_path_kind);
+    let mut queue: VecDeque<i64> = roots.into_iter().collect();
+    let mut visited = HashSet::new();
+    while let Some(hash) = queue.pop_front() {
+        if !visited.insert(hash) {
+            continue;
+        }
+        let Some(block) = state.blocks.get(&hash) else {
+            continue;
+        };
+        let parent_complete = match block.parent {
+            ParentLink::Unknown => false,
+            ParentLink::Root => true,
+            ParentLink::Hash(parent) => state
+                .blocks
+                .get(&parent)
+                .is_some_and(|parent| parent.prefix_complete_workers.contains(worker_id)),
+        };
+        let complete = kind
+            .is_some_and(|kind| parent_complete && block_servable(state, hash, worker_id, kind));
+        let children: Vec<i64> = block.children.iter().copied().collect();
+        let block = state.blocks.get_mut(&hash).expect("block exists");
+        if complete {
+            block.prefix_complete_workers.insert(worker_id.to_string());
+        } else {
+            block.prefix_complete_workers.remove(worker_id);
+        }
+        queue.extend(children);
+    }
+}
+
+/// Returns direct children held by this worker but outside the current REPORT chain.
+fn external_children_held_by_worker(
+    state: &State,
+    worker_id: &str,
+    reported_hashes: &HashSet<i64>,
+    parent: i64,
+) -> Vec<i64> {
+    state
+        .blocks
+        .get(&parent)
+        .into_iter()
+        .flat_map(|block| block.children.iter().copied())
+        .filter(|child| {
+            !reported_hashes.contains(child)
+                && state.blocks.get(child).is_some_and(|child| {
+                    child
+                        .placements
+                        .keys()
+                        .any(|(worker, _)| worker == worker_id)
+                })
+        })
+        .collect()
+}
+
+/// Refreshes derived prefix state along a closed linear REPORT chain.
+///
+/// The caller has verified that no node in the chain has an external child.
+fn refresh_linear_report_chain_prefix_completeness(
+    state: &mut State,
+    worker_id: &str,
+    hashes: &[i64],
+) -> Vec<i64> {
+    let kind = state.workers.get(worker_id).and_then(fast_path_kind);
+    let reported_hashes: HashSet<i64> = hashes.iter().copied().collect();
+    let mut external_dirty_roots = Vec::new();
+    let mut parent_complete = hashes
+        .first()
+        .and_then(|hash| state.blocks.get(hash))
+        .is_some_and(|block| match block.parent {
+            ParentLink::Root => true,
+            ParentLink::Hash(parent) => state
+                .blocks
+                .get(&parent)
+                .is_some_and(|parent| parent.prefix_complete_workers.contains(worker_id)),
+            ParentLink::Unknown => false,
+        });
+
+    for hash in hashes {
+        let was_complete = state
+            .blocks
+            .get(hash)
+            .is_some_and(|block| block.prefix_complete_workers.contains(worker_id));
+        let complete = kind
+            .is_some_and(|kind| parent_complete && block_servable(state, *hash, worker_id, kind));
+        if was_complete != complete {
+            external_dirty_roots.extend(external_children_held_by_worker(
+                state,
+                worker_id,
+                &reported_hashes,
+                *hash,
+            ));
+        }
+        let Some(block) = state.blocks.get_mut(hash) else {
+            continue;
+        };
+        if complete {
+            block.prefix_complete_workers.insert(worker_id.to_string());
+        } else {
+            block.prefix_complete_workers.remove(worker_id);
+        }
+        parent_complete = complete;
+    }
+    external_dirty_roots
+}
+
+/// Returns the length of the longest leading request chain already known to the
+/// Indexer. A missing block starts the normal uncached suffix; a present block
+/// with the wrong parent is a chain conflict and disables the derived fast path.
+fn known_request_prefix_len(state: &State, hashes: &[i64]) -> Option<usize> {
+    let mut known = 0;
+    for (index, hash) in hashes.iter().enumerate() {
+        let expected = if index == 0 {
+            ParentLink::Root
+        } else {
+            ParentLink::Hash(hashes[index - 1])
+        };
+        let Some(block) = state.blocks.get(hash) else {
+            break;
+        };
+        if block.parent != expected {
+            return None;
+        }
+        known += 1;
+    }
+    Some(known)
+}
+
+fn revoke_one(state: &mut State, worker_id: &str, hash: &i64, tier: i32) {
+    let mut removed_mask = None;
+    if let Some(block) = state.blocks.get_mut(hash) {
+        removed_mask = block.placements.remove(&(worker_id.to_string(), tier));
+    }
+    adjust_component_placement_count(state, worker_id, removed_mask, None);

    if let Some(worker) = state.workers.get_mut(worker_id) {
        if let Some(hashes) = worker.holdings.get_mut(&tier) {
@@ -408,12 +822,38 @@ fn revoke_one(state: &mut State, worker_id: &str, hash: &i64, tier: i32) {
        }
    }

-    if remove_block {
-        state.blocks.remove(hash);
+    if state
+        .blocks
+        .get(hash)
+        .is_some_and(|block| block.placements.is_empty())
+    {
        state.hit_counts.remove(hash);
    }
 }

+fn prune_empty_leaf(state: &mut State, mut hash: i64) {
+    loop {
+        let Some(block) = state.blocks.get(&hash) else {
+            return;
+        };
+        if !block.placements.is_empty()
+            || !block.children.is_empty()
+            || !block.prefix_complete_workers.is_empty()
+        {
+            return;
+        }
+        let parent = block.parent;
+        state.blocks.remove(&hash);
+        let ParentLink::Hash(parent_hash) = parent else {
+            return;
+        };
+        if let Some(parent) = state.blocks.get_mut(&parent_hash) {
+            parent.children.remove(&hash);
+        }
+        hash = parent_hash;
+    }
+}
+
 fn dedup_preserve_order(hashes: &[i64]) -> Vec<i64> {
    let mut seen = HashSet::new();
    hashes
@@ -497,4 +937,74 @@ mod tests {
        drop(read_guard);
        query.join().unwrap();
    }
+
+    #[test]
+    fn known_request_prefix_stops_at_uncached_suffix_and_rejects_conflicts() {
+        let mut state = State::default();
+        link_report_chain(&mut state, None, &[1, 2, 3]).unwrap();
+        link_report_chain(&mut state, None, &[9]).unwrap();
+
+        assert_eq!(known_request_prefix_len(&state, &[1, 2, 3, 4, 5]), Some(3));
+        assert_eq!(known_request_prefix_len(&state, &[1, 9]), None);
+    }
+    #[test]
+    fn conflicting_report_chain_does_not_mutate_the_graph() {
+        let mut state = State::default();
+
+        let error = link_report_chain(&mut state, None, &[1, 2, 1]).unwrap_err();
+
+        assert_eq!(error.code(), tonic::Code::InvalidArgument);
+        assert!(state.blocks.is_empty());
+    }
+
+    #[test]
+    fn cyclic_report_chain_does_not_mutate_the_graph() {
+        let mut state = State::default();
+
+        let error = link_report_chain(&mut state, Some(2), &[1, 2]).unwrap_err();
+
+        assert_eq!(error.code(), tonic::Code::InvalidArgument);
+        assert!(state.blocks.is_empty());
+    }
+
+    #[test]
+    fn cyclic_report_chain_through_existing_graph_is_rejected() {
+        let mut state = State::default();
+        link_report_chain(&mut state, Some(2), &[1]).unwrap();
+
+        let error = link_report_chain(&mut state, Some(1), &[2]).unwrap_err();
+
+        assert_eq!(error.code(), tonic::Code::InvalidArgument);
+        assert_eq!(state.blocks[&1].parent, ParentLink::Hash(2));
+        assert_eq!(state.blocks[&2].parent, ParentLink::Unknown);
+    }
+
+    #[test]
+    fn external_children_only_include_the_reporting_workers_branch() {
+        let mut state = State::default();
+
+        link_report_chain(&mut state, None, &[1, 2, 3]).unwrap();
+        link_report_chain(&mut state, Some(1), &[4]).unwrap();
+        state
+            .blocks
+            .get_mut(&4)
+            .unwrap()
+            .placements
+            .insert(("worker-b".into(), TierType::TierHbm as i32), 0);
+        let reported_hashes: HashSet<i64> = [1, 2, 3].into_iter().collect();
+        assert!(
+            external_children_held_by_worker(&state, "worker-a", &reported_hashes, 1,).is_empty()
+        );
+
+        state
+            .blocks
+            .get_mut(&4)
+            .unwrap()
+            .placements
+            .insert(("worker-a".into(), TierType::TierHbm as i32), 0);
+        assert_eq!(
+            external_children_held_by_worker(&state, "worker-a", &reported_hashes, 1,),
+            vec![4]
+        );
+    }
 }
@@ -182,7 +182,16 @@ where
 /// [`KvIndexerService::into_server`]: that sets the per-message ceiling, this
 /// bounds how many messages can be in flight against it at once.
 pub fn server_builder() -> Server {
-    Server::builder().max_concurrent_streams(MAX_CONCURRENT_STREAMS)
+    server_builder_with_max_concurrent_streams(MAX_CONCURRENT_STREAMS)
+}
+
+/// A transport builder with an explicit stream bound for high-fanout fleets.
+///
+/// The default entry point keeps the stable 64-stream behavior, while the
+/// standalone Indexer binary can raise the bound when it has one bridge per
+/// worker.
+pub fn server_builder_with_max_concurrent_streams(max_concurrent_streams: u32) -> Server {
+    Server::builder().max_concurrent_streams(max_concurrent_streams)
 }

 #[tonic::async_trait]
@@ -775,6 +784,7 @@ mod tests {
            hashes: hashes.iter().map(|h| h.parse().unwrap()).collect(),
            component_masks: Vec::new(),
            block_sizes: Vec::new(),
+            parent_block_hash: None,
        }
    }

@@ -14,12 +14,22 @@ pub fn dram() -> i32 {
 }

 pub fn action(kind: ExternalKvActionType, tier: i32, hashes: &[i64]) -> ExternalKvAction {
+    action_with_parent(kind, tier, None, hashes)
+}
+
+pub fn action_with_parent(
+    kind: ExternalKvActionType,
+    tier: i32,
+    parent_block_hash: Option<i64>,
+    hashes: &[i64],
+) -> ExternalKvAction {
    ExternalKvAction {
        r#type: kind as i32,
        tier,
        hashes: hashes.to_vec(),
        component_masks: Vec::new(),
        block_sizes: Vec::new(),
+        parent_block_hash,
    }
 }

@@ -31,6 +41,17 @@ pub fn component_report(
    hashes: &[i64],
    masks: &[u32],
    block_sizes: &[u32],
+) -> ExternalKvAction {
+    component_report_with_parent(tier, None, hashes, masks, block_sizes)
+}
+
+#[allow(dead_code)]
+pub fn component_report_with_parent(
+    tier: i32,
+    parent_block_hash: Option<i64>,
+    hashes: &[i64],
+    masks: &[u32],
+    block_sizes: &[u32],
 ) -> ExternalKvAction {
    ExternalKvAction {
        r#type: ExternalKvActionType::ActionReport as i32,
@@ -38,6 +59,7 @@ pub fn component_report(
        hashes: hashes.to_vec(),
        component_masks: masks.to_vec(),
        block_sizes: block_sizes.to_vec(),
+        parent_block_hash,
    }
 }

@@ -34,7 +34,7 @@ use sgl_kv_indexer::{
    PrefixIndex, PrefixIndexConfig, MAX_GRPC_DECODING_MESSAGE_SIZE,
 };
 use test_id::nanos;
-use test_kv::{action, apply_request, hbm};
+use test_kv::{action, action_with_parent, apply_request, hbm};
 use test_net::free_addr;

 async fn start_backend(
@@ -211,6 +211,27 @@ fn apply_report(
    )
 }

+fn apply_report_with_parent(
+    worker: &str,
+    addr: &str,
+    seq: u64,
+    tier: i32,
+    parent_block_hash: Option<i64>,
+    hashes: &[i64],
+) -> ApplyExternalKvBatchRequest {
+    apply_request(
+        worker,
+        addr,
+        seq,
+        vec![action_with_parent(
+            ExternalKvActionType::ActionReport,
+            tier,
+            parent_block_hash,
+            hashes,
+        )],
+    )
+}
+
 #[tokio::test]
 async fn multiple_workers_share_one_indexer_server() {
    let mut indexer = start().await;
@@ -225,7 +246,7 @@ async fn multiple_workers_share_one_indexer_server() {
            "10.0.0.1:9000",
            1,
            hbm(),
-            &[hash_0, shared_hash],
+            &[shared_hash, hash_0],
        ))
        .await
        .expect("apply worker-0");
@@ -235,7 +256,7 @@ async fn multiple_workers_share_one_indexer_server() {
            "10.0.0.2:9000",
            1,
            hbm(),
-            &[hash_1, shared_hash],
+            &[shared_hash, hash_1],
        ))
        .await
        .expect("apply worker-1");
@@ -309,6 +330,7 @@ async fn validation_errors_map_to_invalid_argument_over_grpc() {
            hashes: vec![1],
            component_masks: Vec::new(),
            block_sizes: Vec::new(),
+            parent_block_hash: None,
        }],
    };
    let err = c
@@ -318,6 +340,76 @@ async fn validation_errors_map_to_invalid_argument_over_grpc() {
    assert_eq!(err.code(), Code::InvalidArgument);
 }

+#[tokio::test]
+async fn rejected_batch_is_atomic_over_grpc() {
+    let mut c = start().await;
+    c.apply_external_kv_batch(apply_report("w", "old-address", 1, hbm(), &[1, 2]))
+        .await
+        .expect("seed chain");
+
+    let err = c
+        .apply_external_kv_batch(apply_request(
+            "w",
+            "new-address",
+            2,
+            vec![
+                action(ExternalKvActionType::ActionReport, hbm(), &[3]),
+                action_with_parent(ExternalKvActionType::ActionReport, hbm(), Some(9), &[2]),
+            ],
+        ))
+        .await
+        .expect_err("conflicting parent must reject the whole batch");
+    assert_eq!(err.code(), Code::InvalidArgument);
+
+    let old = c
+        .match_external_kv(MatchExternalKvRequest {
+            hashes: vec![1],
+            count_as_hit: false,
+        })
+        .await
+        .expect("query original state")
+        .into_inner();
+    assert_eq!(old.matches.len(), 1);
+    assert_eq!(old.matches[0].address, "old-address");
+
+    let leaked = c
+        .match_external_kv(MatchExternalKvRequest {
+            hashes: vec![3],
+            count_as_hit: false,
+        })
+        .await
+        .expect("query rejected action")
+        .into_inner();
+    assert!(leaked.matches.is_empty());
+}
+
+#[tokio::test]
+async fn cyclic_report_is_rejected_over_grpc() {
+    let mut c = start().await;
+    let err = c
+        .apply_external_kv_batch(apply_report_with_parent(
+            "w",
+            "address",
+            1,
+            hbm(),
+            Some(2),
+            &[1, 2],
+        ))
+        .await
+        .expect_err("cyclic report must be rejected");
+    assert_eq!(err.code(), Code::InvalidArgument);
+
+    let response = c
+        .match_external_kv(MatchExternalKvRequest {
+            hashes: vec![1, 2],
+            count_as_hit: false,
+        })
+        .await
+        .expect("query rejected report")
+        .into_inner();
+    assert!(response.matches.is_empty());
+}
+
 #[tokio::test]
 async fn match_prefix_over_grpc() {
    let mut c = start().await;
@@ -358,12 +450,14 @@ async fn prefix_query_scans_more_than_one_apply_chunk_over_grpc() {
    let mut indexer = start().await;
    let hashes: Vec<i64> = (0..=APPLY_CHUNK_SIZE as i64).collect();
    for (seq, chunk) in hashes.chunks(APPLY_CHUNK_SIZE).enumerate() {
+        let parent_block_hash = (seq > 0).then_some(chunk[0] - 1);
        indexer
-            .apply_external_kv_batch(apply_report(
+            .apply_external_kv_batch(apply_report_with_parent(
                "large-prefix-worker",
                "10.0.0.1:9000",
                seq as u64,
                hbm(),
+                parent_block_hash,
                chunk,
            ))
            .await
@@ -487,7 +581,7 @@ async fn start_recording_deadlines(
 /// the only thing letting the indexer shed a query whose caller gave up.
 #[tokio::test]
 async fn router_client_publishes_its_deadline_on_the_wire() {
-    let (index, seen) = start_recording_deadlines(Duration::from_millis(250)).await;
+    let (index, seen) = start_recording_deadlines(Duration::from_secs(2)).await;

    index
        .match_prefix(vec![1, 2, 3])
@@ -17,7 +17,10 @@ use sgl_kv_indexer::pb::{
 use sgl_kv_indexer::{
    InMemoryKvIndexerBackend, KvIndexerBackend, WorkerPrefixInput, COMPONENT_FULL, COMPONENT_SWA,
 };
-use test_kv::{action, apply_request as apply_req, component_report, dram, hbm};
+use test_kv::{
+    action, action_with_parent, apply_request as apply_req, component_report,
+    component_report_with_parent, dram, hbm,
+};
 use tonic::Status;

 fn backend() -> InMemoryKvIndexerBackend {
@@ -81,6 +84,38 @@ itest!(report_then_match_returns_worker_and_address, b, {
    assert!(tiers_for(&resp, "w1", 3).is_empty());
 });

+itest!(rejected_batch_does_not_publish_earlier_actions, b, {
+    b.apply_external_kv_batch(apply_req(
+        "w1",
+        "old-address",
+        1,
+        vec![action(ExternalKvActionType::ActionReport, hbm(), &[1, 2])],
+    ))
+    .await
+    .unwrap();
+
+    let error = b
+        .apply_external_kv_batch(apply_req(
+            "w1",
+            "new-address",
+            2,
+            vec![
+                action(ExternalKvActionType::ActionReport, hbm(), &[3]),
+                action_with_parent(ExternalKvActionType::ActionReport, hbm(), Some(9), &[2]),
+            ],
+        ))
+        .await
+        .unwrap_err();
+    assert_eq!(error.code(), tonic::Code::InvalidArgument);
+
+    let old = b.match_external_kv(match_req(&[1], false)).await.unwrap();
+    assert_eq!(old.matches.len(), 1);
+    assert_eq!(old.matches[0].address, "old-address");
+
+    let leaked = b.match_external_kv(match_req(&[3], false)).await.unwrap();
+    assert!(leaked.matches.is_empty());
+});
+
 itest!(large_request_preserves_complete_ordered_results, b, {
    // Exercise a large write and read while preserving complete ordered results.
    let expected_hashes: Vec<i64> = (0..300).collect();
@@ -587,10 +622,19 @@ async fn prefix_fast_path_matches_default_impl() {
    fast.apply_external_kv_batch(report("w-short", "10.0.0.2:1", 1, &[1, 2]))
        .await
        .unwrap();
-    // w-hole holds 1, 3, 4 but not 2: strict prefix must be 1.
-    fast.apply_external_kv_batch(report("w-hole", "10.0.0.3:1", 1, &[1, 3, 4]))
+    // w-hole first learns the same chain, then loses block 2 while descendants
+    // remain placed: strict prefix must be 1.
+    fast.apply_external_kv_batch(report("w-hole", "10.0.0.3:1", 1, &[1, 2, 3, 4]))
        .await
        .unwrap();
+    fast.apply_external_kv_batch(apply_req(
+        "w-hole",
+        "10.0.0.3:1",
+        2,
+        vec![action(ExternalKvActionType::ActionRevoke, hbm(), &[2])],
+    ))
+    .await
+    .unwrap();
    // w-noaddr is unroutable and must be excluded by both paths.
    fast.apply_external_kv_batch(report("w-noaddr", "", 1, &[1, 2]))
        .await
@@ -626,6 +670,64 @@ async fn prefix_fast_path_matches_default_impl() {
    assert!(fast_resp.blocks_read >= 1);
 }

+#[tokio::test]
+async fn prefix_fast_path_returns_worker_depths_with_uncached_suffix() {
+    let (fast, reference) = shared_state_pair();
+    fast.apply_external_kv_batch(report("w-long", "10.0.0.1:1", 1, &[1, 2, 3]))
+        .await
+        .unwrap();
+    fast.apply_external_kv_batch(report("w-short", "10.0.0.2:1", 1, &[1, 2]))
+        .await
+        .unwrap();
+
+    // Blocks 4 and 5 are the newly appended turn and are not cached anywhere.
+    // They must cap the maximum prefix without disabling the known-prefix path.
+    let query = [1, 2, 3, 4, 5];
+    let fast_response = fast
+        .match_external_kv_prefix(prefix_req(&query))
+        .await
+        .unwrap();
+    let reference_response = reference
+        .match_external_kv_prefix(prefix_req(&query))
+        .await
+        .unwrap();
+
+    assert_eq!(
+        prefix_pairs(&fast_response),
+        prefix_pairs(&reference_response)
+    );
+    assert_eq!(
+        prefix_pairs(&fast_response),
+        vec![("w-long".to_string(), 3), ("w-short".to_string(), 2)]
+    );
+}
+
+#[tokio::test]
+async fn prefix_fast_path_falls_back_on_existing_parent_conflict() {
+    let (fast, reference) = shared_state_pair();
+    fast.apply_external_kv_batch(report("w1", "10.0.0.1:1", 1, &[1, 2]))
+        .await
+        .unwrap();
+    // Hash 9 is an independent root, not a child of hash 1.
+    fast.apply_external_kv_batch(report("w1", "10.0.0.1:1", 2, &[9]))
+        .await
+        .unwrap();
+
+    let query = [1, 9];
+    let fast_response = fast
+        .match_external_kv_prefix(prefix_req(&query))
+        .await
+        .unwrap();
+    let reference_response = reference
+        .match_external_kv_prefix(prefix_req(&query))
+        .await
+        .unwrap();
+    assert_eq!(
+        prefix_pairs(&fast_response),
+        prefix_pairs(&reference_response)
+    );
+}
+
 #[tokio::test]
 async fn prefix_first_block_miss_reads_one_block() {
    let b = backend();
@@ -662,6 +764,156 @@ async fn prefix_max_blocks_caps_the_scan() {
    assert_eq!(resp.matches[0].matched_prefix_blocks, 2);
 }

+#[tokio::test]
+async fn prefix_complete_revoke_and_restore_propagates_to_descendants() {
+    let (fast, reference) = shared_state_pair();
+    fast.apply_external_kv_batch(report("w1", "10.0.0.1:1", 1, &[1, 2, 3, 4]))
+        .await
+        .unwrap();
+
+    fast.apply_external_kv_batch(apply_req(
+        "w1",
+        "10.0.0.1:1",
+        2,
+        vec![action(ExternalKvActionType::ActionRevoke, hbm(), &[2])],
+    ))
+    .await
+    .unwrap();
+    let after_revoke = fast
+        .match_external_kv_prefix(prefix_req(&[1, 2, 3, 4]))
+        .await
+        .unwrap();
+    assert_eq!(prefix_pairs(&after_revoke), vec![("w1".to_string(), 1)]);
+
+    fast.apply_external_kv_batch(apply_req(
+        "w1",
+        "10.0.0.1:1",
+        3,
+        vec![action_with_parent(
+            ExternalKvActionType::ActionReport,
+            hbm(),
+            Some(1),
+            &[2],
+        )],
+    ))
+    .await
+    .unwrap();
+    let restored = fast
+        .match_external_kv_prefix(prefix_req(&[1, 2, 3, 4]))
+        .await
+        .unwrap();
+    let expected = reference
+        .match_external_kv_prefix(prefix_req(&[1, 2, 3, 4]))
+        .await
+        .unwrap();
+    assert_eq!(prefix_pairs(&restored), vec![("w1".to_string(), 4)]);
+    assert_eq!(prefix_pairs(&restored), prefix_pairs(&expected));
+}
+
+#[tokio::test]
+async fn prefix_complete_fast_path_preserves_cache_hit_rate() {
+    let (fast, reference) = shared_state_pair();
+    let query: Vec<i64> = (1..=64).collect();
+    let worker_count = 32usize;
+    let mut expected_prefix_sum = 0u64;
+    for worker in 0..worker_count {
+        let depth = 1 + (worker * 7 % query.len());
+        expected_prefix_sum += depth as u64;
+        fast.apply_external_kv_batch(report(
+            &format!("w-{worker:02}"),
+            &format!("http://worker-{worker:02}"),
+            1,
+            &query[..depth],
+        ))
+        .await
+        .unwrap();
+    }
+
+    let fast_response = fast
+        .match_external_kv_prefix(prefix_req(&query))
+        .await
+        .unwrap();
+    let reference_response = reference
+        .match_external_kv_prefix(prefix_req(&query))
+        .await
+        .unwrap();
+    assert_eq!(
+        prefix_pairs(&fast_response),
+        prefix_pairs(&reference_response)
+    );
+    assert_eq!(fast_response.matches.len(), worker_count);
+
+    let fast_prefix_sum: u64 = fast_response
+        .matches
+        .iter()
+        .map(|item| item.matched_prefix_blocks as u64)
+        .sum();
+    let reference_prefix_sum: u64 = reference_response
+        .matches
+        .iter()
+        .map(|item| item.matched_prefix_blocks as u64)
+        .sum();
+    assert_eq!(fast_prefix_sum, expected_prefix_sum);
+    assert_eq!(fast_prefix_sum, reference_prefix_sum);
+    let hit_rate = fast_prefix_sum as f64 / (worker_count * query.len()) as f64;
+    let reference_hit_rate = reference_prefix_sum as f64 / (worker_count * query.len()) as f64;
+    assert!((hit_rate - reference_hit_rate).abs() < f64::EPSILON);
+}
+
+#[tokio::test]
+async fn full_only_prefix_complete_fast_path_matches_reference() {
+    let (fast, reference) = shared_state_pair();
+    let full_spec = WorkerCacheSpec {
+        version: 1,
+        components: COMPONENT_FULL,
+        swa_window_tokens: 0,
+        full_tier_mask: (1 << hbm()) | (1 << dram()),
+        swa_tier_mask: 0,
+        mamba_tier_mask: 0,
+    };
+    fast.apply_external_kv_batch(apply_with_spec(
+        "w-full",
+        "10.0.0.1:1",
+        1,
+        full_spec,
+        vec![component_report(
+            hbm(),
+            &[11, 12, 13, 14],
+            &[COMPONENT_FULL; 4],
+            &[16; 4],
+        )],
+    ))
+    .await
+    .unwrap();
+    fast.apply_external_kv_batch(apply_with_spec(
+        "w-full",
+        "10.0.0.1:1",
+        2,
+        full_spec,
+        vec![action(ExternalKvActionType::ActionRevoke, hbm(), &[13])],
+    ))
+    .await
+    .unwrap();
+
+    let query = [11, 12, 13, 14];
+    let fast_response = fast
+        .match_external_kv_prefix(prefix_req(&query))
+        .await
+        .unwrap();
+    let reference_response = reference
+        .match_external_kv_prefix(prefix_req(&query))
+        .await
+        .unwrap();
+    assert_eq!(
+        prefix_pairs(&fast_response),
+        prefix_pairs(&reference_response)
+    );
+    assert_eq!(
+        prefix_pairs(&fast_response),
+        vec![("w-full".to_string(), 2)]
+    );
+}
+
 // --- component-aware placement & prefix -------------------------------------

 /// A hybrid-SWA spec: full servable from HBM+DRAM, swa a 100-token trailing
@@ -766,7 +1018,13 @@ async fn partial_eviction_replace_shrinks_component_set() {
        "10.0.0.1:1",
        2,
        swa_spec(),
-        vec![component_report(hbm(), &[2], &[COMPONENT_FULL], &[80])],
+        vec![component_report_with_parent(
+            hbm(),
+            Some(1),
+            &[2],
+            &[COMPONENT_FULL],
+            &[80],
+        )],
    ))
    .await
    .unwrap();
@@ -827,12 +1085,10 @@ async fn duplicate_hash_in_one_report_keeps_last_snapshot() {
        "10.0.0.1:1",
        1,
        swa_spec(),
-        vec![component_report(
-            hbm(),
-            &[1, 1],
-            &[COMPONENT_FULL | COMPONENT_SWA, COMPONENT_FULL],
-            &[80, 80],
-        )],
+        vec![
+            component_report(hbm(), &[1], &[COMPONENT_FULL | COMPONENT_SWA], &[80]),
+            component_report(hbm(), &[1], &[COMPONENT_FULL], &[80]),
+        ],
    ))
    .await
    .unwrap();