fix(kv-router): filter non-main ZMQ KV event groups

[PeaBrane]

Motivation

DeepSeek V4 vLLM KV events include a group_idx that separates the main/full compressed KV group from SWA and other auxiliary KV/cache state groups. Dynamo's current ZMQ wire parser did not model this field, so non-main DSV4 event groups could either fail array-like decoding or be flattened into the normal radix index.

This PR adds a v0 guard: parse group_idx from vLLM-style ZMQ events and ignore any non-main group. Missing group_idx remains accepted for legacy producers such as SGLang and older mocker/vLLM event streams.

Changes

• Add RawKvEvent::Ignored and return it for non-zero group_idx.
• Make convert_event return Option<PlacementEvent> so ignored raw events do not reach the indexer.
• Skip ignored events in the main ZMQ listener before allocating next_event_id.
• Skip ignored events in the standalone indexer listener while still allowing batch cursor/watermark progress.
• Emit group_idx: 0 from the mocker ZMQ path for store/remove events.
• Add parser tests for BlockStored with group_idx = 0, group_idx = 1, and missing group_idx.

Verification

• cargo test -p dynamo-kv-router zmq_wire
• (cd /Users/peabrane/Documents/codes/dynamo/lib/kv-router && cargo clippy --no-deps --all-targets -- -D warnings && cargo fmt) && (cd /Users/peabrane/Documents/codes/dynamo/lib/llm && cargo clippy --no-default-features -- -D warnings && cargo fmt)

Note: cargo test -p dynamo-llm test_convert_event_block_stored --no-default-features reached linking and failed because this local macOS environment is missing libstdc++ (ld: library 'stdc++' not found).

---

Summary by CodeRabbit

• New Features

  • Implemented group-based event filtering to selectively exclude events from non-primary groups, preventing unnecessary processing and improving pipeline efficiency.

• Bug Fixes

  • Enhanced event conversion error handling with graceful failure modes; invalid or malformed events are now silently skipped instead of propagating errors, improving system resilience and preventing cascading failures.

[coderabbitai]

Walkthrough

This PR introduces support for filtering KV cache events based on a group_idx field. It adds a new RawKvEvent::Ignored variant, implements filtering logic during deserialization for non-zero group_idx values, changes convert_event to return Option<PlacementEvent> instead of an unconditional type, and updates all event processing paths and tests to handle the fallible conversion.

Changes

Cohort / File(s)	Summary
ZMQ Wire Format & Deserialization lib/kv-router/src/zmq_wire.rs	Adds RawKvEvent::Ignored variant, implements filtering logic in custom deserialization to mark BlockStored and BlockRemoved events with non-zero group_idx as ignored, changes convert_event return type to Option<PlacementEvent>, and updates tests to verify group filtering behavior and handle the new return type.
Event Processing Paths lib/kv-router/src/standalone_indexer/listener.rs, lib/llm/src/kv_router/publisher/zmq_listener.rs	Updates replay and live event processing loops to pattern-match on Option from convert_event, silently skipping failed conversions, and adds filtering of ignored events before resource allocation.
Event Matching & Handling lib/llm/src/block_manager/kv_consolidator/subscriber.rs	Adds explicit RawKvEvent::Ignored arm in the event processing match statement to handle ignored variants.
Testing & Mocking lib/llm/src/kv_router/publisher/tests.rs, lib/llm/src/mocker.rs	Updates test calls to convert_event with .unwrap() to handle the new Option return type, and adds group_idx field (set to 0) to mock ZMQ events.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 38.46% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly and concisely describes the main change: filtering non-main ZMQ KV event groups by parsing and ignoring non-zero group_idx values.
Description check	✅ Passed	The description covers all required sections: Motivation explains the problem, Changes lists the specific modifications with file-level details, and Verification provides testing commands performed.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

🧹 Nitpick comments (3)

lib/kv-router/src/zmq_wire.rs (3)

102-102: Ignored is exposed as an accepted wire-level variant.

RawKvEvent::Ignored is an internal filtering marker, but it's also:

• Listed in the unknown_variant error messages (L331, L444), and
• Explicitly accepted as incoming "type": "Ignored" (L328, L438-441), and
• Serializable via the derived Serialize + #[serde(tag = "type")].

This lets any producer send {"type": "Ignored"} and have that payload silently dropped downstream. If that's intentional (e.g., for testing or to let producers pre-tag filtered events), a short code comment would make the contract explicit. Otherwise, consider rejecting "Ignored" on the wire and/or marking the variant #[serde(skip)] so it's purely an internal control-flow value.

Also applies to: 327-332, 437-445

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@lib/kv-router/src/zmq_wire.rs` at line 102, RawKvEvent::Ignored is currently
treated as a wire-level variant (it is serialized/deserialized with #[serde(tag
= "type")], accepted as incoming "Ignored" and shown in unknown_variant
messages), which allows producers to send {"type":"Ignored"} and have those
messages silently dropped; either document this as an explicit wire contract or
prevent it being accepted on the wire. To fix, choose one: 1) Make Ignored
internal-only by adding #[serde(skip)] (or remove it from the tagged enum) and
adjust any matching logic in functions handling RawKvEvent to handle the
now-internal marker, and remove "Ignored" from accepted types/unknown_variant
messages (references: RawKvEvent::Ignored, serialize/deserialize derive,
unknown_variant handling); or 2) If it must be a valid wire type, add a comment
explaining the intentional contract and modify deserialization/unknown_variant
logic to treat incoming "Ignored" explicitly (e.g., log/validate) instead of
silently dropping. Ensure all places that match or list Ignored (the accept list
and unknown_variant messages) are updated to reflect the chosen behavior.

---

697-745: Expand group_idx test coverage to BlockRemoved and map format.

The added tests cover BlockStored sequence format with group_idx = 0 / 1 / missing. The parser also gained group_idx handling for:

• BlockRemoved sequence format (line 421, 425)
• BlockStored / BlockRemoved map format (lines 270-272, 298-300, 319-321)
• Explicit "Ignored" tag in both map and seq paths (lines 328, 438-441)

None of these paths have regression coverage. A producer that emits non-main-group BlockRemoved (or uses the map encoding) would silently bypass filtering if those branches regressed. Consider adding analogous tests for at least:

1. BlockRemoved sequence with group_idx = 1 → Ignored
2. A map-encoded BlockStored/BlockRemoved with group_idx = 1 → Ignored

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@lib/kv-router/src/zmq_wire.rs` around lines 697 - 745, Add missing test
coverage for non-main-group and map-encoded cases: create helpers similar to
block_stored_sequence_with_group_idx for BlockRemoved (e.g.,
block_removed_sequence_with_group_idx) and for map-encoded messages (e.g.,
block_stored_map_with_group_idx and block_removed_map_with_group_idx), then add
tests that decode them with from_slice and assert RawKvEvent::Ignored when
group_idx is Some(1) and that they decode to RawKvEvent::BlockStored or
RawKvEvent::BlockRemoved when group_idx is Some(0) or None; reference existing
symbols RawKvEvent::BlockStored, RawKvEvent::BlockRemoved, RawKvEvent::Ignored
and the block_stored_sequence_with_group_idx helper to mirror structure.

---

130-139: Documentation comment would clarify the trailing field order-agnostic design.

The BlockStoredTrailingField untagged enum correctly disambiguates group_idx (u32) from block_mm_infos (array) in msgpack, allowing vLLM to emit either field in any order at positions 9–10 of the sequence. A brief doc comment on the enum would explain this contract and flag the fragility: if vLLM adds a new u32 or array field, the untagged match could misclassify. Also clarify what "up to 2 trailing fields" means in the for _ in 0..2 loop.

The map deserialization path with extra_keys fallback shows this is intentional forward-compatibility—the code accepts both old (extra_keys) and new (block_mm_infos) field formats. Consider documenting that in the enum's doc comment to signal this is deliberate design.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@lib/kv-router/src/zmq_wire.rs` around lines 130 - 139, Add a doc comment to
clarify the design and fragility of the untagged enum BlockStoredTrailingField
and the surrounding parsing logic: document that BlockStoredTrailingField
(variants GroupIdx(u32) and BlockMmInfos(Vec<Option<BlockExtraInfo>>)) exists to
accept either a trailing u32 or an array in positions 9–10 of the vLLM msgpack
sequence (order-agnostic, up to 2 trailing fields processed by the for _ in 0..2
loop), explain the forward-compatibility fallback to the map-path extra_keys
format, and warn that introducing any new u32 or array-typed trailing fields in
vLLM could cause misclassification; also add a one-line doc note near
is_non_main_group describing its purpose in treating group_idx==0 as main group.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Nitpick comments:
In `@lib/kv-router/src/zmq_wire.rs`:
- Line 102: RawKvEvent::Ignored is currently treated as a wire-level variant (it
is serialized/deserialized with #[serde(tag = "type")], accepted as incoming
"Ignored" and shown in unknown_variant messages), which allows producers to send
{"type":"Ignored"} and have those messages silently dropped; either document
this as an explicit wire contract or prevent it being accepted on the wire. To
fix, choose one: 1) Make Ignored internal-only by adding #[serde(skip)] (or
remove it from the tagged enum) and adjust any matching logic in functions
handling RawKvEvent to handle the now-internal marker, and remove "Ignored" from
accepted types/unknown_variant messages (references: RawKvEvent::Ignored,
serialize/deserialize derive, unknown_variant handling); or 2) If it must be a
valid wire type, add a comment explaining the intentional contract and modify
deserialization/unknown_variant logic to treat incoming "Ignored" explicitly
(e.g., log/validate) instead of silently dropping. Ensure all places that match
or list Ignored (the accept list and unknown_variant messages) are updated to
reflect the chosen behavior.
- Around line 697-745: Add missing test coverage for non-main-group and
map-encoded cases: create helpers similar to
block_stored_sequence_with_group_idx for BlockRemoved (e.g.,
block_removed_sequence_with_group_idx) and for map-encoded messages (e.g.,
block_stored_map_with_group_idx and block_removed_map_with_group_idx), then add
tests that decode them with from_slice and assert RawKvEvent::Ignored when
group_idx is Some(1) and that they decode to RawKvEvent::BlockStored or
RawKvEvent::BlockRemoved when group_idx is Some(0) or None; reference existing
symbols RawKvEvent::BlockStored, RawKvEvent::BlockRemoved, RawKvEvent::Ignored
and the block_stored_sequence_with_group_idx helper to mirror structure.
- Around line 130-139: Add a doc comment to clarify the design and fragility of
the untagged enum BlockStoredTrailingField and the surrounding parsing logic:
document that BlockStoredTrailingField (variants GroupIdx(u32) and
BlockMmInfos(Vec<Option<BlockExtraInfo>>)) exists to accept either a trailing
u32 or an array in positions 9–10 of the vLLM msgpack sequence (order-agnostic,
up to 2 trailing fields processed by the for _ in 0..2 loop), explain the
forward-compatibility fallback to the map-path extra_keys format, and warn that
introducing any new u32 or array-typed trailing fields in vLLM could cause
misclassification; also add a one-line doc note near is_non_main_group
describing its purpose in treating group_idx==0 as main group.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: 3cd9fa65-c201-41bb-8905-3399efd8316e

📥 Commits

Reviewing files that changed from the base of the PR and between 7ae0c39 and 3204116.

📒 Files selected for processing (6)

• lib/kv-router/src/standalone_indexer/listener.rs
• lib/kv-router/src/zmq_wire.rs
• lib/llm/src/block_manager/kv_consolidator/subscriber.rs
• lib/llm/src/kv_router/publisher/tests.rs
• lib/llm/src/kv_router/publisher/zmq_listener.rs
• lib/llm/src/mocker.rs

[zhongdaor-nv]

https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/vllm/tests/test_vllm_kv_events_api.py#L75

Need to update the test to include group_idx

[PeaBrane]

Need to update the test to include group_idx

Done in 063171b: updated the vLLM KV event API guard test to expect group_idx on both BlockStored and BlockRemoved, and added serialization checks for the new field positions.