[event] add transactional engine cache event subscriber

[putaopi7]

Summary

This PR adds a standalone subscriber that synchronizes RTP-LLM or vLLM local KV-cache state into KVCM.

• Add a dedicated ST_EVENT_REPORT metadata-only storage type.
• Add authoritative per-host cache snapshots and idempotent incremental updates.
• Add RTP-LLM and vLLM source adapters behind one transactional source interface.
• Commit source cursors only after KVCM acknowledges the corresponding update.
• Integrate node registration, heartbeat, generation fencing, and host-down cleanup.
• Exclude engine-owned cache locations from KVCM capacity accounting and active reclaim.

Data flow

RTP GetCacheStatus v2 / vLLM ZMQ
  -> Source.prepare
  -> normalize BlockRecord
  -> KVCM ReportEvent
  -> KVCM ACK
  -> Source.commit

Source.prepare creates a candidate state without advancing its committed cursor. A KVCM failure therefore applies backpressure to the source instead of allowing unbounded local event accumulation or acknowledging data that was not persisted.

Design

Dedicated event-report storage

Engine-local GPU cache is represented by a new ST_EVENT_REPORT = 8 storage type and event-report:// locations.

This backend owns node liveness and cache-location metadata only. It does not read, write, or delete the actual KV data, which remains owned by RTP-LLM or vLLM.

• Existing Vineyard storage behavior and URI routing remain unchanged.
• Event-report storage is not included in KVCM managed capacity.
• Event-report locations are excluded from active reclaim.
• The storage still participates in replica discovery while its engine node is available.
• Instance groups opt in through event_reporting_storage_candidates.

Authoritative snapshot semantics

EVENT_BLOCK_SNAPSHOT replaces the complete block set for one instance_id + host_ip_port. An empty snapshot clears that host without affecting other hosts or instances.

Snapshots, deltas, and asynchronous host cleanup for the same instance and host are serialized by striped mutexes. Generation and availability fences prevent stale timeout cleanup from removing locations after a node has recovered or re-registered.

Transactional subscriber

• Cold start first prepares an authoritative source snapshot.
• The subscriber then performs NODE_REGISTER -> snapshot ACK -> source commit -> HEARTBEAT.
• If the initial snapshot fails, the subscriber reports HOST_DOWN, re-registers, and retries the same uncommitted snapshot.
• KVCM failures do not advance the source cursor.
• Source failures pause heartbeat; repeated failures report HOST_DOWN before exit.
• A periodic authoritative refresh reconciles any long-lived drift.

Small snapshots use one replacement request. Large snapshots use an empty replacement followed by chunked ADD requests to stay within the request-size limit. A failed transfer is retried from the empty replacement and eventually converges, although the host can be partially visible while chunks are being applied.

RTP-LLM source

• Consumes protocol-v2 snapshots and ordered events from every configured DP endpoint.
• Tracks generation and cursor independently per endpoint.
• Commits the aggregate state only after every endpoint succeeds.
• Requires exactly dp_size distinct endpoints to avoid silently missing a DP rank.

vLLM source

• Consumes sequenced MsgPack/ZMQ cache events and requests bounded replay after a gap.
• Accepts a replay as authoritative only from sequence 0 or after AllBlocksCleared.
• Treats a new sequence-0 stream as a publisher epoch reset.
• Remains unregistered and sends no heartbeat until an authoritative baseline exists.
• Maps the low 64 bits of the vLLM hash to KVCM signed int64, matching the legacy integer event representation.
• Explicitly supports dp_size = 1; multi-publisher aggregation is left for a separate change.

Configuration

KVCM needs a dedicated event-report storage:

{
  "global_unique_name": "engine_cache_events",
  "event_report": {
    "heartbeat_timeout_ms": 30000,
    "cleanup_grace_ms": 300000,
    "liveness_check_interval_ms": 5000
  }
}

The corresponding instance group must include engine_cache_events in event_reporting_storage_candidates. It must not be added to ordinary storage_candidates.

Example RTP subscriber:

python -m kv_cache_manager.cache_event_subscriber \
  --engine rtp \
  --manager-uri http://manager:8080 \
  --instance-group default \
  --instance-id model-a \
  --host-ip-port worker-a:9000 \
  --rtp-endpoints worker-a:9001 \
  --block-size 64 \
  --model-name model-a \
  --cache-group-count 1

The vLLM path additionally requires msgspec and pyzmq; the RTP path has no source-specific Python dependency.

Compatibility

• The new protobuf enum value and instance-group candidate field are additive.
• Existing storage candidates and Vineyard nodes keep their current behavior.
• No engine cache is reported unless the dedicated storage and subscriber are explicitly configured.
• Cache metadata remains isolated by instance_id.

Validation

• Subscriber Python tests: 27/27 passed, including startup ordering, retry, pagination, generation reset, vLLM replay, and multi-endpoint failure cases.
• manager_message_proto_util builds successfully with the updated protobuf contract.
• InstanceGroupTest.EventReportStorageProtoRoundTrip: passed.
• StorageConfigTest: passed.
• EventReportBackendTest: passed.
• CacheManagerTest.ReportEventSnapshotAuthoritativelyReplacesHostBlocks: passed.
• CacheReclaimerTest.TestFilterLocationNeverReclaimsEventReportedStorage: passed.
• Event-report MetaIndexer filter cases: passed.
• DataStorageManagerTest.RegisterEventReportStorage: passed.
• git diff --check: passed.

Known limitations and follow-ups

• Run a real RTP + Subscriber + KVCM three-process integration test with the companion RTP-LLM PR.
• vLLM currently supports one DP publisher only.
• Chunked large snapshots allow temporary partial visibility during transfer; retry is convergent but not atomically visible.
• KVCM keys are 64-bit, so a wider vLLM digest is intentionally truncated to the signed-int64 key domain.

Companion PR

• RTP-LLM: alibaba/rtp-llm#1214

[CLAassistant]

All committers have signed the CLA.